opentelemetry_experimental

%%%------------------------------------------------------------------------
%% 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/1,
         callbacks_table_name/1,
         view_aggregation_table_name/1,
         metrics_table_name/1,
         add_instrument/2,
         register_callback/4,
         add_view/3,
         add_view/4,
         record/4,
         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/include/opentelemetry.hrl").
%% need to move shared records out of otel_span.hrl
-include_lib("opentelemetry/include/otel_span.hrl").
-include_lib("opentelemetry_api_experimental/include/otel_metrics.hrl").
-include_lib("kernel/include/logger.hrl").
-include("otel_metrics.hrl").
-include("otel_view.hrl").

-type meter() :: #meter{}.

-export_type([meter/0]).

-record(reader,
        {
         child_id :: atom(),
         module                      :: module(),
         config                      :: term(),
         callbacks_tab               :: ets:tid() | atom(),
         view_aggregation_tab        :: ets:tid() | atom(),
         metrics_tab                 :: ets:tid() | atom(),
         default_aggregation_mapping :: map(),
         default_temporality_mapping :: map()
        }).

-type reader() :: #reader{}.

-record(state,
        {
         shared_meter,

         views :: [otel_view:t()],
         readers :: [#reader{}],

         resource :: otel_resource:t()
        }).

-spec start_link(otel_configuration:t()) -> {ok, pid()} | ignore | {error, term()}.
start_link(Config) ->
    start_link(?DEFAULT_METER_PROVIDER, Config).

-spec start_link(atom(), otel_configuration:t()) -> {ok, pid()} | ignore | {error, term()}.
start_link(Provider, Config) ->
    gen_server:start_link({local, Provider}, ?MODULE, [Provider, Config], []).

-spec add_instrument(atom(), otel_instrument:t()) -> boolean().
add_instrument(Provider, Instrument) ->
    gen_server:call(Provider, {add_instrument, Instrument}).

-spec register_callback(atom(), [otel_instrument:t()], otel_instrument:callback(), term()) -> boolean().
register_callback(Provider, Instruments, Callback, CallbackArgs) ->
    gen_server:call(Provider, {register_callback, Instruments, Callback, CallbackArgs}).

-spec add_view(atom(), otel_view:criteria(), otel_view:config()) -> boolean().
add_view(Provider, Criteria, Config) ->
    add_view(Provider, undefined, 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(atom(), otel_instrument:t(), number(), opentelemetry:attributes_map()) -> ok.
record(Provider, Instrument, Number, Attributes) ->
    gen_server:cast(Provider, {record,  #measurement{instrument=Instrument,
                                                     value=Number,
                                                     attributes=Attributes}}).

callbacks_table_name(ChildId) ->
    list_to_atom(atom_to_list(ChildId) ++ "_callbacks_tab").

view_aggregation_table_name(ChildId) ->
    list_to_atom(atom_to_list(ChildId) ++ "_view_aggregation_tab").

metrics_table_name(ChildId) ->
    list_to_atom(atom_to_list(ChildId) ++ "_metrics_tab").

force_flush(Provider) ->
    gen_server:call(Provider, force_flush).

init([Name, Config]) ->
    Resource = otel_resource_detector:get_resource(),

    Meter = #meter{module=otel_meter_default,
                   provider=Name},

    opentelemetry_experimental:set_default_meter({otel_meter_default, Meter}),

    Readers = add_metric_readers(Config),

    {ok, #state{shared_meter=Meter,
                views=[],
                readers=Readers,
                resource=Resource}}.

handle_call({record, Measurement}, _From, State=#state{readers=Readers,
                                                       views=Views}) ->
    handle_measurement(Measurement, Readers, Views),
    {reply, ok, State};
handle_call(resource, _From, State=#state{resource=Resource}) ->
    {reply, Resource, State};
handle_call({add_instrument, Instrument}, _From, State=#state{readers=Readers,
                                                              views=Views}) ->
    _ = add_instrument_(Instrument, Views, Readers),
    {reply, ok, State};
handle_call({register_callback, Instruments, Callback, CallbackArgs}, _From, State=#state{readers=Readers,
                                                                            views=Views}) ->
    _ = register_callback_(Instruments, Callback, CallbackArgs, Views, 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}) ->
    %% TODO: drop View if Criteria is a wildcard instrument name and View name is not undefined
    View = otel_view:new(Name, Criteria, Config),
    {reply, true, State#state{views=[View | Views]}};
handle_call(force_flush, _From, State=#state{readers=Readers}) ->
    [otel_metric_reader:collect(ChildId) || #reader{child_id=ChildId} <- Readers],
    {reply, ok, State}.

handle_cast({record, Measurement}, State=#state{readers=Readers,
                                                views=Views}) ->
    handle_measurement(Measurement, Readers, Views),
    {noreply, State}.

handle_info(_, State) ->
    {noreply, State}.

code_change(State) ->
    {ok, State}.

%%

%% Match the Instrument to views and then store a per-Reader aggregation for the View
add_instrument_(Instrument, Views, Readers) ->
    ViewMatches = otel_view:match_instrument_to_views(Instrument, Views),
    lists:map(fun(Reader=#reader{callbacks_tab=CallbackTab,
                                 view_aggregation_tab=ViewAggregationTab}) ->
                      Matches = per_reader_aggregations(Reader, Instrument, ViewMatches),
                      _ = ets:insert(ViewAggregationTab, {Instrument, Matches}),
                      case {Instrument#instrument.callback, Instrument#instrument.callback_args} of
                          {undefined, _} ->
                              ok;
                          {Callback, CallbackArgs} ->
                              ets:insert(CallbackTab, {Callback, CallbackArgs, [Instrument]})
                      end
              end, Readers).

%% Match the Instrument to views and then store a per-Reader aggregation for the View
register_callback_(Instruments, Callback, CallbackArgs, Views, Readers) ->
    lists:map(fun(Instrument) ->
                      ViewMatches = otel_view:match_instrument_to_views(Instrument, Views),
                      lists:map(fun(Reader=#reader{callbacks_tab=CallbackTab,
                                                   view_aggregation_tab=ViewAggregationTab}) ->
                                        Matches = per_reader_aggregations(Reader, Instrument, ViewMatches),
                                        _ = ets:insert(ViewAggregationTab, {Instrument, Matches}),
                                        ets:insert(CallbackTab, {Callback, CallbackArgs, Instruments})
                                end, Readers)
              end, Instruments).

add_metric_readers(Config) ->
    ReaderConfigs = maps:get(readers, Config, []),
    [metric_reader(ChildId, ReaderModule, ReaderConfig) ||
        #{id := ChildId, module := ReaderModule, config := ReaderConfig} <- ReaderConfigs].

metric_reader(ChildId, ReaderModule, ReaderConfig) ->
    CallbacksTableName = callbacks_table_name(ChildId),
    ViewAggregationTableName = view_aggregation_table_name(ChildId),
    MetricsTableName = metrics_table_name(ChildId),

    CallbacksTable = ets:new(CallbacksTableName, [bag, protected, named_table, {keypos, 1}]),
    ViewAggregationTable = ets:new(ViewAggregationTableName, [set, protected, named_table, {keypos, 1}]),
    MetricsTable = ets:new(MetricsTableName, [set, public, named_table, {keypos, 2}]),

    ReaderAggregationMapping = maps:merge(otel_aggregation:default_mapping(),
                                          maps:get(default_aggregation_mapping, ReaderConfig, #{})),
    ReaderTemporalityMapping = maps:merge(otel_aggregation:temporality_mapping(),
                                          maps:get(default_temporality_mapping, ReaderConfig, #{})),

    #reader{child_id=ChildId,
            module=ReaderModule,
            config=ReaderConfig,
            callbacks_tab=CallbacksTable,
            view_aggregation_tab=ViewAggregationTable,
            metrics_tab=MetricsTable,
            default_aggregation_mapping=ReaderAggregationMapping,
            default_temporality_mapping=ReaderTemporalityMapping}.


%% a Measurement's Instrument is matched against Views
%% each matched View+Reader becomes a ViewAggregation
%% for each ViewAggregation a Measurement updates a Metric (`#metric')
%% active metrics are indexed by the ViewAggregation name + the Measurement's Attributes

handle_measurement(Measurement=#measurement{instrument=Instrument},
                   Readers,
                   Views) ->
    ViewMatches = otel_view:match_instrument_to_views(Instrument, Views),
    lists:map(fun(Reader=#reader{view_aggregation_tab=ViewAggregationTab}) ->
                      case ets:lookup(ViewAggregationTab, Instrument) of
                          [] ->
                              %% this instrument hasn't been seen before
                              Matches = per_reader_aggregations(Reader, Instrument, ViewMatches),
                              true = ets:insert(ViewAggregationTab, {Instrument, Matches}),
                              update_aggregations(Measurement, Reader, Matches);
                          [{_, Matches}] ->
                              %% TODO: matches need to be  updated when a new view is added
                              update_aggregations(Measurement, Reader, Matches)
                      end
              end, Readers).

update_aggregations(#measurement{attributes=Attributes,
                                 value=Value}, #reader{metrics_tab=MetricsTab}, ViewAggregations) ->
    lists:map(fun(#view_aggregation{name=Name,
                                    aggregation_module=AggregationModule,
                                    aggregation_options=Options}) ->
                      case AggregationModule:aggregate(MetricsTab, {Name, Attributes}, Value) of
                              true ->
                                  ok;
                              false ->
                                  %% entry doesn't exist, create it and rerun the aggregate function
                                  Metric = AggregationModule:init({Name, Attributes}, Options),
                                  _ = ets:insert(MetricsTab, Metric),
                                  AggregationModule:aggregate(MetricsTab, {Name, Attributes}, Value)
                          end
              end, ViewAggregations).

%% create an aggregation for each Reader and its possibly unique aggregation/temporality
per_reader_aggregations(Reader, Instrument, ViewAggregations) ->
    [view_aggregation_for_reader(Instrument, ViewAggregation, View, Reader)
     || {View, ViewAggregation} <- ViewAggregations].

view_aggregation_for_reader(Instrument=#instrument{kind=Kind}, ViewAggregation, View,
                            Reader=#reader{default_temporality_mapping=ReaderTemporalityMapping}) ->
    AggregationModule = aggregation_module(Instrument, View, Reader),
    Temporality = maps:get(Kind, ReaderTemporalityMapping, ?AGGREGATION_TEMPORALITY_UNSPECIFIED),

    ViewAggregation#view_aggregation{
      aggregation_module=AggregationModule,
      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}, #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)]}.