%%%------------------------------------------------------------------------ %% Copyright 2023, OpenTelemetry Authors %% Licensed under the Apache License, Version 2.0 (the "License"); %% you may not use this file except in compliance with the License. %% You may obtain a copy of the License at %% %% http://www.apache.org/licenses/LICENSE-2.0 %% %% Unless required by applicable law or agreed to in writing, software %% distributed under the License is distributed on an "AS IS" BASIS, %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. %% See the License for the specific language governing permissions and %% limitations under the License. %% %% @doc %% @end %%%------------------------------------------------------------------------- -module(otel_observables). -export([run_callbacks/5]). -include_lib("kernel/include/logger.hrl"). -include_lib("opentelemetry_api_experimental/include/otel_metrics.hrl"). -include("otel_view.hrl"). -include_lib("opentelemetry_api/include/otel_ctx.hrl"). -type callbacks() :: [{otel_instrument:callback(), otel_instrument:callback_args(), otel_instrument:t()}]. %% call each callback and associate the result with the Instruments it observes -spec run_callbacks(callbacks(), reference(), ets:table(), ets:table(), ets:table()) -> ok. run_callbacks(Callbacks, ReaderId, StreamTab, MetricsTab, ExemplarsTab) -> lists:foreach(fun({Callback, CallbackArgs, Instruments}) when is_list(Instruments) -> Ctx0 = otel_ctx:new(), {Results, Ctx} = ?with_ctx(Ctx0, fun() -> Results0 = Callback(CallbackArgs), Results0 end), handle_instruments_observations(Ctx, Results, Instruments, StreamTab, MetricsTab, ExemplarsTab, ReaderId); ({Callback, CallbackArgs, Instrument}) -> Ctx0 = otel_ctx:new(), {Results, Ctx} = ?with_ctx(Ctx0, fun() -> Results0 = Callback(CallbackArgs), Results0 end), %% when not a list of instruments it isn't expecting named observation %% results so we use handle_instrument instead of handle_instruments %% but we can't type that correctly so have to use a `fixme' handle_instrument_observations(Ctx, %% eqwalizer:fixme can maybe do better typing Results, Instrument, StreamTab, MetricsTab, ExemplarsTab, ReaderId) end, Callbacks). %% lookup Streams for Instrument and aggregate each observation -spec handle_instrument_observations(otel_ctx:t(), [otel_instrument:observation()], otel_instrument:t(), ets:table(), ets:table(), ets:table(), reference()) -> ok. handle_instrument_observations(Ctx, Results, #instrument{meter={_, Meter}, name=Name}, StreamTab, MetricsTab, ExemplarsTab, ReaderId) -> try ets:lookup_element(StreamTab, {Meter, Name}, 2) of Streams -> [handle_observations(Ctx, MetricsTab, ExemplarsTab, Stream, Results) || #stream{reader=Id}=Stream <- Streams, Id =:= ReaderId], ok catch error:badarg -> %% no Views for this Instrument, so nothing to do ok end. %% handle results for a multi-instrument callback -spec handle_instruments_observations(otel_ctx:t(), [otel_instrument:named_observations()], [otel_instrument:t()], ets:table(), ets:table(), ets:table(), reference()) -> ok. handle_instruments_observations(_Ctx, [], _Instruments, _StreamTab, _MetricsTab, _ExemplarsTab, _ReaderId) -> ok; handle_instruments_observations(Ctx, [{InstrumentName, Results} | Rest], Instruments, StreamTab, MetricsTab, ExemplarsTab, ReaderId) -> case lists:keyfind(InstrumentName, #instrument.name, Instruments) of false -> ?LOG_DEBUG("Unknown Instrument ~p used in metric callback", [InstrumentName]); Instrument -> handle_instrument_observations(Ctx, Results, Instrument, StreamTab, MetricsTab, ExemplarsTab, ReaderId) end, handle_instruments_observations(Ctx, Rest, Instruments, StreamTab, MetricsTab, ExemplarsTab, ReaderId); handle_instruments_observations(Ctx, [Result | Rest], Instruments, StreamTab, MetricsTab, ExemplarsTab, ReaderId) -> ?LOG_DEBUG("Each multi-instrument callback result must be a tuple of " "type {atom(), [{number(), map()}]} but got ~p", [Result]), handle_instruments_observations(Ctx, Rest, Instruments, StreamTab, MetricsTab, ExemplarsTab, ReaderId); handle_instruments_observations(_Ctx, Results, _Instruments, _StreamTab, _MetricsTab, _ExemplarsTab, _ReaderId) -> ?LOG_DEBUG("Multi-instrument callback result must be a list of type " "[{atom(), [{number(), map()}]}] but got ~p", [Results]), ok. %% update aggregation for each observation handle_observations(_Ctx, _MetricsTab, _ExemplarsTab, _Stream, []) -> ok; handle_observations(Ctx, MetricsTab, ExemplarsTab, Stream, [{Number, Attributes} | Rest]) when is_number(Number), is_map(Attributes) -> _ = otel_aggregation:maybe_init_aggregate(Ctx, MetricsTab, ExemplarsTab, Stream, Number, Attributes), handle_observations(Ctx, MetricsTab, ExemplarsTab, Stream, Rest); handle_observations(Ctx, MetricsTab, ExemplarsTab, Stream, [Result | Rest]) -> ?LOG_DEBUG("Each metric callback result must be of type {number(), map()} but got ~p", [Result]), handle_observations(Ctx, MetricsTab, ExemplarsTab, Stream, Rest); handle_observations(_Ctx, _MetricsTab, _ExemplarsTab, _Stream, Result) -> ?LOG_DEBUG("Metric callback return must be a list of type [{number(), map()}] or " "[{atom(), [{number(), map()}]}] but got", [Result]), ok.