%%%------------------------------------------------------------------------ %% Copyright 2022, 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 MetricReader is an SDK module that provides the %% common configurable aspects of the OpenTelemetry Metrics SDK and %% determines the following capabilities: %% %% * Collecting metrics from the SDK on demand. %% * Handling the ForceFlush and Shutdown signals from the SDK. %% @end %%%------------------------------------------------------------------------- -module(otel_metric_reader). -behaviour(gen_server). -export([start_link/3, collect/1, checkpoint_generation/1, shutdown/1]). -export([init/1, handle_call/3, handle_cast/2, handle_info/2, handle_continue/2, code_change/1]). -include_lib("opentelemetry_api_experimental/include/otel_metrics.hrl"). -include("otel_view.hrl"). -include("otel_metrics.hrl"). -record(state, { exporter, %% eqwalizer:ignore waiting on sup_ref to be exported https://github.com/erlang/otp/pull/7205 provider_sup :: supervisor:sup_ref(), id :: reference(), default_aggregation_mapping :: #{otel_instrument:kind() => module()}, temporality_mapping :: #{otel_instrument:kind() => otel_instrument:temporality()}, export_interval_ms :: integer() | undefined, tref :: reference() | undefined, callbacks_tab :: ets:table(), streams_tab :: ets:table(), metrics_tab :: ets:table(), exemplars_tab :: ets:table(), config :: #{}, resource :: otel_resource:t() | undefined, generation_ref :: atomics:atomics_ref(), producers :: [otel_metric_producer:t()] }). %% -spec start_link(atom(), map()) -> {ok, pid()} | ignore | {error, term()}. %% start_link(ChildId, CallbacksTable, StreamsTable, MetricsTable, Config) -> %% gen_server:start_link({local, ChildId}, ?MODULE, [ChildId, CallbacksTable, StreamsTable, MetricsTable, Config], []). start_link(ReaderId, ProviderSup, Config) -> gen_server:start_link(?MODULE, [ReaderId, ProviderSup, Config], []). collect(ReaderPid) -> ReaderPid ! collect. shutdown(ReaderPid) -> gen_server:call(ReaderPid, shutdown). %% Get the current checkpoint generation. checkpoint_generation(ReaderId) -> GenerationRef = persistent_term:get({?MODULE, ReaderId}), atomics:get(GenerationRef, 1). %% Increment and return the previous checkpoint generation. inc_checkpoint_generation(ReaderId) -> GenerationRef = persistent_term:get({?MODULE, ReaderId}), atomics:add_get(GenerationRef, 1, 1) - 1. init([ReaderId, ProviderSup, Config]) -> ExporterModuleConfig = maps:get(exporter, Config, undefined), Exporter = otel_exporter:init(ExporterModuleConfig), DefaultAggregationMapping = maps:get(default_aggregation_mapping, Config, otel_aggregation:default_mapping()), Temporality = maps:get(default_temporality_mapping, Config, otel_aggregation:default_temporality_mapping()), %% if a periodic reader is needed then this value is set %% somehow need to do a default of 10000 MILLIS, but only if this is a periodic reader ExporterIntervalMs = maps:get(export_interval_ms, Config, undefined), TRef = case ExporterIntervalMs of undefined -> undefined; _ -> erlang:send_after(ExporterIntervalMs, self(), collect) end, GenerationRef = try persistent_term:get({?MODULE, ReaderId}) catch error:badarg -> GenerationRef0 = atomics:new(1, []), persistent_term:put({?MODULE, ReaderId}, GenerationRef0), GenerationRef0 end, %% eqwalizer:fixme get an unbound record error until the fixme for state record is resolved {ok, #state{exporter=Exporter, provider_sup=ProviderSup, id=ReaderId, default_aggregation_mapping=DefaultAggregationMapping, temporality_mapping=Temporality, export_interval_ms=ExporterIntervalMs, tref=TRef, generation_ref=GenerationRef, producers=[], config=Config}, {continue, register_with_server}}. %% eqwalizer:fixme get an unbound record error until the fixme for state record is resolved handle_continue(register_with_server, State=#state{provider_sup=ProviderSup, id=ReaderId, default_aggregation_mapping=DefaultAggregationMapping, temporality_mapping=Temporality}) -> ServerPid = otel_meter_server_sup:provider_pid(ProviderSup), {CallbacksTab, StreamsTab, MetricsTab, ExemplarsTab, Resource, Producers} = otel_meter_server:add_metric_reader(ServerPid, ReaderId, self(), DefaultAggregationMapping, Temporality), {noreply, State#state{callbacks_tab=CallbacksTab, streams_tab=StreamsTab, metrics_tab=MetricsTab, exemplars_tab=ExemplarsTab, resource=Resource, producers=Producers}}. handle_call(shutdown, _From, State) -> {reply, ok, State}; handle_call(_, _From, State) -> {noreply, State}. handle_cast(_, State) -> {noreply, State}. %% eqwalizer:fixme get an unbound record error until the fixme for state record is resolved handle_info(collect, State=#state{exporter=undefined, export_interval_ms=ExporterIntervalMs, tref=TRef}) when TRef =/= undefined andalso ExporterIntervalMs =/= undefined -> erlang:cancel_timer(TRef, [{async, true}]), NewTRef = erlang:send_after(ExporterIntervalMs, self(), collect), {noreply, State#state{tref=NewTRef}}; handle_info(collect, State=#state{id=ReaderId, exporter={ExporterModule, Config}, export_interval_ms=undefined, tref=undefined, callbacks_tab=CallbacksTab, streams_tab=StreamsTab, metrics_tab=MetricsTab, exemplars_tab=ExemplarsTab, resource=Resource, producers=Producers }) -> Metrics = run_collection(CallbacksTab, StreamsTab, MetricsTab, ExemplarsTab, ReaderId, Producers), otel_exporter:export_metrics(ExporterModule, Metrics, Resource, Config), {noreply, State}; handle_info(collect, State=#state{id=ReaderId, exporter={ExporterModule, Config}, export_interval_ms=ExporterIntervalMs, tref=TRef, callbacks_tab=CallbacksTab, streams_tab=StreamsTab, metrics_tab=MetricsTab, exemplars_tab=ExemplarsTab, resource=Resource, producers=Producers }) when TRef =/= undefined andalso ExporterIntervalMs =/= undefined -> erlang:cancel_timer(TRef, [{async, true}]), NewTRef = erlang:send_after(ExporterIntervalMs, self(), collect), Metrics = run_collection(CallbacksTab, StreamsTab, MetricsTab, ExemplarsTab, ReaderId, Producers), otel_exporter:export_metrics(ExporterModule, Metrics, Resource, Config), {noreply, State#state{tref=NewTRef}}; %% no tref or exporter, do nothing at all handle_info(_, State) -> {noreply, State}. code_change(State) -> {ok, State}. %% run_collection(CallbacksTab, StreamsTab, MetricsTab, ExemplarsTab, ReaderId, Producers) -> %% collect from view aggregations table and then export Metrics = collect_(CallbacksTab, StreamsTab, MetricsTab, ExemplarsTab, ReaderId), ExternalMetrics = run_producers(Producers), Metrics ++ ExternalMetrics. run_producers(Producers) -> run_producers(Producers, []). run_producers([], ExternalMetrics) -> ExternalMetrics; run_producers([Producer | Rest], ExternalMetrics) -> run_producers(Rest, run_producer(Producer) ++ ExternalMetrics). run_producer(Producer) -> otel_metric_producer:produce_batch(Producer). -spec collect_(any(), ets:table(), any(), ets:table(), reference()) -> [any()]. collect_(CallbacksTab, StreamsTab, MetricsTab, ExemplarsTab, ReaderId) -> _ = run_callbacks(ReaderId, CallbacksTab, StreamsTab, MetricsTab, ExemplarsTab), %% Need to be able to efficiently get all from VIEW_AGGREGATIONS_TAB that apply to this reader %% for each VIEW_AGGREGATIONS_TAB entry look up metrics from METRICS_TAB using the name %% to select for key `{Name, '_'}'. This gives the current value for each set of attributes %% for an aggregation. %% use the information (temporality) from the VIEW_AGGREGATIONS_TAB entry to reset the %% METRICS_TAB entry value (like setting value back to 0 for DELTA) %% StreamsTab is a `bag' so to iterate over every Stream for %% each Instrument we use `first'/`next' and lookup the list of Streams %% by the key (Instrument) Key = ets:first(StreamsTab), Generation = inc_checkpoint_generation(ReaderId), collect_(CallbacksTab, StreamsTab, MetricsTab, ExemplarsTab, Generation, ReaderId, [], Key). run_callbacks(ReaderId, CallbacksTab, StreamsTab, MetricsTab, ExemplarsTab) -> try ets:lookup_element(CallbacksTab, ReaderId, 2) of Callbacks -> otel_observables:run_callbacks(Callbacks, ReaderId, StreamsTab, MetricsTab, ExemplarsTab) catch error:badarg -> [] end. collect_(_CallbacksTab, _StreamsTab, _MetricsTab, _ExemplarsTab, _Generation, _ReaderId, MetricsAcc, '$end_of_table') -> MetricsAcc; collect_(CallbacksTab, StreamsTab, MetricsTab, ExemplarsTab, Generation, ReaderId, MetricsAcc, Key) -> Streams = ets:lookup_element(StreamsTab, Key, 2), collect_(CallbacksTab, StreamsTab, MetricsTab, ExemplarsTab, Generation, ReaderId, checkpoint_metrics(MetricsTab, ExemplarsTab, Generation, ReaderId, Streams) ++ MetricsAcc, ets:next(StreamsTab, Key)). checkpoint_metrics(MetricsTab, ExemplarsTab, Generation, Id, Streams) -> lists:foldl(fun(#stream{aggregation_module=otel_aggregation_drop}, Acc) -> Acc; (Stream=#stream{name=Name, reader=ReaderId, instrument=Instrument=#instrument{unit=Unit}, aggregation_module=AggregationModule, description=Description }, Acc) when Id =:= ReaderId -> Data = AggregationModule:collect(MetricsTab, ExemplarsTab, Stream, Generation), [metric(Instrument, Name, Description, Unit, Data) | Acc]; (_, Acc) -> Acc end, [], Streams). metric(#instrument{meter={_, Meter=#meter{}}}, Name, Description, Unit, Data) -> #metric{scope=otel_meter_default:scope(Meter), name=Name, description=Description, unit=Unit, data=Data}.