Current section
Files
Jump to
Current section
Files
src/otel_meter_server.erl
%%%------------------------------------------------------------------------
%% Copyright 2019, 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 This module is the SDK's implementation of the MeterProvider. The
%% calls to the server are done from the API module `otel_meter_provider'.
%% This `gen_server' is started as part of the SDK's supervision tree and
%% registers itself as the default MeterProvider by using the atom
%% `otel_meter_provider' as its name.
%%
%% The MeterProvider is where Meter's are created and Views are registered.
%%
%% Each MeterProvider has an associated MetricReader.
%%
%% The MeterProvider "owns" any Instrument created with a Meter from that
%% MeterProvider.
%%
%% For Measumrents on an Instrument the MeterProvider's Views are checked
%% for a match. If no match is found the default aggregation and temporality
%% is used.
%% @end
%%%-------------------------------------------------------------------------
-module(otel_meter_server).
-behaviour(gen_server).
-export([start_link/4,
add_metric_reader/4,
add_metric_reader/5,
get_readers/0,
get_readers/1,
add_instrument/1,
add_instrument/2,
register_callback/3,
register_callback/4,
add_view/2,
add_view/3,
add_view/4,
record/5,
force_flush/0,
force_flush/1,
report_cb/1]).
-export([init/1,
handle_call/3,
handle_cast/2,
handle_info/2,
code_change/1]).
-include_lib("opentelemetry_api_experimental/include/otel_metrics.hrl").
-include_lib("opentelemetry_api_experimental/include/otel_meter.hrl").
-include_lib("kernel/include/logger.hrl").
-include("otel_metrics.hrl").
-include("otel_view.hrl").
-type meter() :: #meter{}.
-record(reader,
{
id :: reference(),
pid :: pid(),
monitor_ref :: reference(),
default_aggregation_mapping :: map(),
default_temporality_mapping :: map()
}).
-type reader() :: #reader{}.
-type view_config() :: #{name => otel_instrument:name() | undefined,
description => unicode:unicode_binary() | undefined,
selector => otel_view:criteria(),
attribute_keys => [opentelemetry:attribute_key()] | undefined,
aggregation_module => module() | undefined,
aggregation_options => map()}.
-export_type([meter/0,
view_config/0]).
-record(state,
{
shared_meter,
instruments_tab :: ets:table(),
callbacks_tab :: ets:table(),
streams_tab :: ets:table(),
metrics_tab :: ets:table(),
exemplars_tab :: ets:table(),
exemplars_enabled :: boolean(),
exemplar_filter :: always_on | always_off | trace_based,
views :: [otel_view:t()],
readers :: [#reader{}],
resource :: otel_resource:t(),
producers :: [otel_metric_producer:t()]
}).
%% I think these have warnings because the new view function is ignored
%% which is because it calls functions that use matchspecs in record defs
-dialyzer({nowarn_function, add_view_/9}).
-dialyzer({nowarn_function, new_view/1}).
-spec start_link(atom(), atom(), otel_resource:t(), otel_configuration:t()) -> {ok, pid()} | ignore | {error, term()}.
start_link(Name, RegName, Resource, Config) ->
gen_server:start_link({local, RegName}, ?MODULE, [Name, RegName, Resource, Config], []).
-spec add_instrument(otel_instrument:t()) -> boolean().
add_instrument(Instrument) ->
add_instrument(?GLOBAL_METER_PROVIDER_REG_NAME, Instrument).
-spec add_instrument(atom(), otel_instrument:t()) -> boolean().
add_instrument(Provider, Instrument) ->
gen_server:call(Provider, {add_instrument, Instrument}).
add_metric_reader(ReaderId, ReaderPid, DefaultAggregationMapping, Temporality) ->
add_metric_reader(?GLOBAL_METER_PROVIDER_REG_NAME, ReaderId, ReaderPid,
DefaultAggregationMapping, Temporality).
add_metric_reader(Provider, ReaderId, ReaderPid, DefaultAggregationMapping, Temporality) ->
gen_server:call(Provider, {add_metric_reader, ReaderId, ReaderPid, DefaultAggregationMapping, Temporality}).
get_readers() ->
get_readers(?GLOBAL_METER_PROVIDER_REG_NAME).
get_readers(Provider) ->
gen_server:call(Provider, get_readers).
-spec register_callback([otel_instrument:t()], otel_instrument:callback(), otel_instrument:callback_args()) -> boolean().
register_callback(Instruments, Callback, CallbackArgs) ->
register_callback(?GLOBAL_METER_PROVIDER_REG_NAME, Instruments, Callback, CallbackArgs).
-spec register_callback(atom(), [otel_instrument:t()], otel_instrument:callback(), otel_instrument:callback_args()) -> boolean().
register_callback(Provider, Instruments, Callback, CallbackArgs) ->
gen_server:call(Provider, {register_callback, Instruments, Callback, CallbackArgs}).
-spec add_view(otel_view:criteria(), otel_view:config()) -> boolean().
add_view(Criteria, Config) ->
add_view(?GLOBAL_METER_PROVIDER_REG_NAME, undefined, Criteria, Config).
-spec add_view(otel_view:name(), otel_view:criteria(), otel_view:config()) -> boolean().
add_view(Name, Criteria, Config) ->
add_view(?GLOBAL_METER_PROVIDER_REG_NAME, Name, Criteria, Config).
-spec add_view(atom(), otel_view:name(), otel_view:criteria(), otel_view:config()) -> boolean().
add_view(Provider, Name, Criteria, Config) ->
gen_server:call(Provider, {add_view, Name, Criteria, Config}).
-spec record(otel_ctx:t(), #meter{}, otel_instrument:t() | otel_instrument:name(), number(), opentelemetry:attributes_map()) -> ok.
record(Ctx, Meter, Name, Number, Attributes) when is_atom(Name) ->
handle_measurement(Ctx, Meter, Name, Number, Attributes);
record(Ctx, Meter, #instrument{name=Name}, Number, Attributes) ->
handle_measurement(Ctx, Meter, Name, Number, Attributes).
-spec force_flush() -> ok.
force_flush() ->
force_flush(?GLOBAL_METER_PROVIDER_REG_NAME).
-spec force_flush(gen_server:server_ref()) -> ok.
force_flush(Provider) ->
gen_server:call(Provider, force_flush).
init([Name, RegName, Resource, Config]) ->
InstrumentsTab = otel_metrics_tables:instruments_tab(RegName),
CallbacksTab = otel_metrics_tables:callbacks_tab(RegName),
StreamsTab = otel_metrics_tables:streams_tab(RegName),
MetricsTab = otel_metrics_tables:metrics_tab(RegName),
ExemplarsTab = otel_metrics_tables:exemplars_tab(RegName),
Meter = #meter{module=otel_meter_default,
instruments_tab=InstrumentsTab,
provider=RegName,
streams_tab=StreamsTab,
metrics_tab=MetricsTab,
exemplars_tab=ExemplarsTab},
%% TODO: don't do this if its already set?
opentelemetry_experimental:set_default_meter(Name, {otel_meter_default, Meter}),
Views = lists:filtermap(fun new_view/1, maps:get(views, Config, [])),
ExemplarsEnabled = maps:get(exemplars_enabled, Config, false),
ExemplarFilter = maps:get(exemplar_filter, Config, trace_based),
Producers = init_producers(maps:get(metric_producers, Config, [])),
{ok, #state{shared_meter=Meter,
instruments_tab=InstrumentsTab,
callbacks_tab=CallbacksTab,
streams_tab=StreamsTab,
metrics_tab=MetricsTab,
exemplars_tab=ExemplarsTab,
exemplars_enabled=ExemplarsEnabled,
exemplar_filter=ExemplarFilter,
views=Views,
readers=[],
resource=Resource,
producers=Producers}}.
init_producers(ProducerConfigs) ->
lists:filtermap(fun({ProducerModule, ProducerConfig}) ->
case otel_metric_producer:init(ProducerModule, ProducerConfig) of
false ->
false;
Producer ->
{true, Producer}
end
end, ProducerConfigs).
handle_call(get_readers, _From, State=#state{readers=Readers}) ->
{reply, Readers, State};
handle_call({add_metric_reader, ReaderId, ReaderPid, DefaultAggregationMapping, Temporality},
_From, State=#state{readers=Readers,
views=Views,
instruments_tab=InstrumentsTab,
callbacks_tab=CallbacksTab,
streams_tab=StreamsTab,
metrics_tab=MetricsTab,
exemplars_tab=ExemplarsTab,
exemplars_enabled=ExemplarsEnabled,
exemplar_filter=ExemplarFilter,
resource=Resource,
producers=Producers}) ->
Reader = metric_reader(ReaderId,
ReaderPid,
DefaultAggregationMapping,
Temporality),
Readers1 = [Reader | Readers],
%% create Streams entries for existing View/Instrument
%% matches for the new Reader
_ = update_streams(InstrumentsTab, CallbacksTab, StreamsTab, Views, Readers1, ExemplarsEnabled, ExemplarFilter),
{reply, {CallbacksTab, StreamsTab, MetricsTab, ExemplarsTab, Resource, Producers}, State#state{readers=Readers1}};
handle_call(resource, _From, State=#state{resource=Resource}) ->
{reply, Resource, State};
handle_call({add_instrument, Instrument}, _From, State=#state{readers=Readers,
views=Views,
instruments_tab=InstrumentsTab,
callbacks_tab=CallbacksTab,
streams_tab=StreamsTab,
exemplars_enabled=ExemplarsEnabled,
exemplar_filter=ExemplarFilter}) ->
_ = add_instrument_(InstrumentsTab, CallbacksTab, StreamsTab, Instrument, Views, Readers, ExemplarsEnabled, ExemplarFilter),
{reply, ok, State};
handle_call({register_callback, Instruments, Callback, CallbackArgs}, _From, State=#state{readers=Readers,
callbacks_tab=CallbacksTab}) ->
_ = register_callback_(CallbacksTab, Instruments, Callback, CallbackArgs, Readers),
{reply, ok, State};
handle_call({get_meter, Name, Vsn, SchemaUrl}, _From, State=#state{shared_meter=Meter}) ->
Scope = opentelemetry:instrumentation_scope(Name, Vsn, SchemaUrl),
MeterTuple = {Meter#meter.module,
Meter#meter{instrumentation_scope=Scope}},
{reply, MeterTuple, State};
handle_call({get_meter, Scope}, _From, State=#state{shared_meter=Meter}) ->
{reply, {Meter#meter.module,
Meter#meter{instrumentation_scope=Scope}}, State};
handle_call({add_view, Name, Criteria, Config}, _From, State=#state{views=Views,
instruments_tab=InstrumentsTab,
callbacks_tab=CallbacksTab,
streams_tab=StreamsTab,
readers=Readers}) ->
add_view_(Name, Criteria, Config, InstrumentsTab, CallbacksTab, StreamsTab, Readers, Views, State);
handle_call(force_flush, _From, State=#state{readers=Readers}) ->
[otel_metric_reader:collect(Pid) || #reader{pid=Pid} <- Readers],
{reply, ok, State}.
handle_cast(_, State) ->
{noreply, State}.
%% TODO: Uncomment when we can drop OTP-23 support
%% handle_info({'DOWN_READER', Ref, process, _Pid, _} , State=#state{readers=Readers}) ->
handle_info({'DOWN', Ref, process, _Pid, _} , State=#state{readers=Readers}) ->
{noreply, State#state{readers=lists:keydelete(Ref, #reader.monitor_ref, Readers)}};
handle_info(_, State) ->
{noreply, State}.
code_change(State) ->
{ok, State}.
%%
add_view_(Name, Criteria, Config, InstrumentsTab, CallbacksTab, StreamsTab, Readers, Views, State=#state{exemplars_enabled=ExemplarsEnabled,
exemplar_filter=ExemplarFilter}) ->
case otel_view:new(Name, Criteria, Config) of
{ok, NewView} ->
_ = update_streams(InstrumentsTab, CallbacksTab, StreamsTab, [NewView], Readers, ExemplarsEnabled, ExemplarFilter),
{reply, true, State#state{views=[NewView | Views]}};
{error, named_wildcard_view} ->
{reply, false, State}
end.
new_view(ViewConfig) ->
Name = maps:get(name, ViewConfig, undefined),
Description = maps:get(description, ViewConfig, undefined),
Selector = maps:get(selector, ViewConfig, undefined),
AttributeKeys = maps:get(attribute_keys, ViewConfig, undefined),
AggregationModule = maps:get(aggregation_module, ViewConfig, undefined),
AggregationOptions = maps:get(aggregation_options, ViewConfig, #{}),
case otel_view:new(Name, Selector, #{description => Description,
attribute_keys => AttributeKeys,
aggregation_module => AggregationModule,
aggregation_options => AggregationOptions
}) of
{ok, View} -> {true, View};
{error, named_wildcard_view} -> false
end.
%% Match the Instrument to views and then store a per-Reader aggregation for the View
add_instrument_(InstrumentsTab, CallbacksTab, StreamsTab,
Instrument=#instrument{meter={_, Meter=#meter{}},
name=Name}, Views, Readers, ExemplarsEnabled, ExemplarFilter) ->
case otel_metrics_tables:insert_instrument(InstrumentsTab, Meter, Name, Instrument) of
true ->
update_streams_(Instrument, CallbacksTab, StreamsTab, Views, Readers, ExemplarsEnabled, ExemplarFilter);
false ->
?LOG_INFO("Instrument ~p already created. Ignoring attempt to create Instrument with the same name in the same Meter.", [Name]),
ok
end.
%% used when a new View is added and the Views must be re-matched with each Instrument
update_streams(InstrumentsTab, CallbacksTab, StreamsTab, Views, Readers, ExemplarsEnabled, ExemplarFilter) ->
otel_metrics_tables:foreach_instrument(InstrumentsTab,
fun(Instrument) ->
update_streams_(Instrument,
CallbacksTab,
StreamsTab,
Views,
Readers,
ExemplarsEnabled,
ExemplarFilter)
end).
update_streams_(Instrument=#instrument{meter={_, Meter=#meter{}},
name=Name}, CallbacksTab, StreamsTab, Views, Readers, ExemplarsEnabled, ExemplarFilter) ->
ViewMatches = otel_view:match_instrument_to_views(Instrument, Views, ExemplarsEnabled, ExemplarFilter),
lists:foreach(fun(Reader=#reader{id=ReaderId}) ->
Matches = per_reader_aggregations(Reader, Instrument, ViewMatches),
[true = otel_metrics_tables:insert_stream(StreamsTab, Meter, Name, M) || M <- Matches],
case {Instrument#instrument.callback, Instrument#instrument.callback_args} of
{undefined, _} ->
ok;
{Callback, CallbackArgs} ->
otel_metrics_tables:insert_callback(CallbacksTab, ReaderId, Callback, CallbackArgs, Instrument)
end
end, Readers).
%% Match the Instrument to views and then store a per-Reader aggregation for the View
register_callback_(CallbacksTab, Instruments, Callback, CallbackArgs, Readers) ->
lists:map(fun(#reader{id=ReaderId}) ->
otel_metrics_tables:insert_callback(CallbacksTab, ReaderId, Callback, CallbackArgs, Instruments)
end, Readers).
metric_reader(ReaderId, ReaderPid, DefaultAggregationMapping, Temporality) ->
%% TODO: Uncomment when we can drop OTP-23 support
%% Ref = erlang:monitor(process, ReaderPid, [{tag, 'DOWN_READER'}]),
Ref = erlang:monitor(process, ReaderPid),
ReaderAggregationMapping = maps:merge(otel_aggregation:default_mapping(),
DefaultAggregationMapping),
#reader{id=ReaderId,
pid=ReaderPid,
monitor_ref=Ref,
default_aggregation_mapping=ReaderAggregationMapping,
default_temporality_mapping=Temporality}.
%% a Measurement's Instrument is matched against Views
%% each matched View+Reader becomes a Stream
%% for each Stream a Measurement updates a Metric (`#metric')
%% active metrics are indexed by the Stream name + the Measurement's Attributes
handle_measurement(Ctx, Meter=#meter{streams_tab=StreamsTab}, Name, Number, Attributes) ->
Streams = otel_metrics_tables:match_streams(StreamsTab, Meter, Name),
update_aggregations(Ctx, Meter, Number, Attributes, Streams).
update_aggregations(Ctx, Meter, Value, Attributes, Streams) ->
lists:foreach(fun(Stream=#stream{instrument=Instrument}) ->
maybe_init_aggregate(Ctx, Meter, Value, Instrument, Stream, Attributes);
(_) ->
ok
end, Streams).
maybe_init_aggregate(_, _Meter, Value, #instrument{kind=Kind} = Instrument, _Stream, _Attributes)
when Value < 0, Kind == ?KIND_COUNTER orelse Kind == ?KIND_HISTOGRAM ->
?LOG_INFO("Discarding negative value for instrument ~s of type ~s", [Instrument#instrument.name, Kind]),
ok;
maybe_init_aggregate(Ctx, #meter{metrics_tab=MetricsTab,
exemplars_tab=ExemplarsTab}, Value, _Instrument, Stream, Attributes) ->
otel_aggregation:maybe_init_aggregate(Ctx, MetricsTab, ExemplarsTab, Stream, Value, Attributes).
%% create an aggregation for each Reader and its possibly unique aggregation/temporality
per_reader_aggregations(Reader, Instrument, Streams) ->
[stream_for_reader(Instrument, Stream, View, Reader)
|| {View, Stream} <- Streams].
stream_for_reader(Instrument=#instrument{kind=Kind}, Stream, View=#view{attribute_keys=AttributeKeys},
Reader=#reader{id=Id,
default_temporality_mapping=ReaderTemporalityMapping}) ->
AggregationModule = aggregation_module(Instrument, View, Reader),
Temporality = maps:get(Kind, ReaderTemporalityMapping, ?TEMPORALITY_CUMULATIVE),
Forget = do_forget(Kind, Temporality),
Stream#stream{
reader=Id,
attribute_keys=AttributeKeys,
aggregation_module=AggregationModule,
forget=Forget,
temporality=Temporality};
stream_for_reader(Instrument=#instrument{kind=Kind}, Stream, View,
Reader=#reader{id=Id,
default_temporality_mapping=ReaderTemporalityMapping}) ->
AggregationModule = aggregation_module(Instrument, View, Reader),
Temporality = maps:get(Kind, ReaderTemporalityMapping, ?TEMPORALITY_CUMULATIVE),
Forget = do_forget(Kind, Temporality),
Stream#stream{
reader=Id,
attribute_keys=undefined,
aggregation_module=AggregationModule,
forget=Forget,
temporality=Temporality}.
%% no aggregation defined for the View, so get the aggregation from the Reader
%% the Reader's mapping of Instrument Kind to Aggregation was merged with the
%% global default, so any missing Kind entries are filled in from the global
%% mapping in `otel_aggregation'
-spec aggregation_module(otel_instrument:t(), otel_view:t(), reader()) -> module().
aggregation_module(#instrument{kind=Kind}, undefined,
#reader{default_aggregation_mapping=ReaderAggregationMapping}) ->
maps:get(Kind, ReaderAggregationMapping);
aggregation_module(#instrument{kind=Kind}, #view{aggregation_module=undefined},
#reader{default_aggregation_mapping=ReaderAggregationMapping}) ->
maps:get(Kind, ReaderAggregationMapping);
aggregation_module(_Instrument, #view{aggregation_module=Module}, _Reader) ->
Module.
report_cb(#{instrument_name := Name,
class := Class,
exception := Exception,
stacktrace := StackTrace}) ->
{"failed to create instrument: name=~ts exception=~ts",
[Name, otel_utils:format_exception(Class, Exception, StackTrace)]};
report_cb(#{view_name := Name,
class := Class,
exception := Exception,
stacktrace := StackTrace}) ->
{"failed to create view: name=~ts exception=~ts",
[Name, otel_utils:format_exception(Class, Exception, StackTrace)]}.
do_forget(_, ?TEMPORALITY_DELTA) ->
true;
do_forget(?KIND_OBSERVABLE_COUNTER, _) ->
true;
do_forget(?KIND_OBSERVABLE_GAUGE, _) ->
true;
do_forget(?KIND_OBSERVABLE_UPDOWNCOUNTER, _) ->
true;
do_forget(_, _) ->
false.