opentelemetry_experimental

%%%------------------------------------------------------------------------
%% 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
%% @end
%%%-------------------------------------------------------------------------
-module(otel_aggregation_sum).

-behaviour(otel_aggregation).

-export([init/2,
         aggregate/7,
         collect/4]).

-include("otel_metrics.hrl").
-include_lib("opentelemetry_api_experimental/include/otel_metrics.hrl").
-include("otel_view.hrl").

-type t() :: #sum_aggregation{}.

-export_type([t/0]).

%% ignore eqwalizer errors in functions using a lot of matchspecs
-eqwalizer({nowarn_function, checkpoint/3}).
-eqwalizer({nowarn_function, aggregate/7}).
-dialyzer({nowarn_function, checkpoint/3}).
-dialyzer({nowarn_function, aggregate/7}).
-dialyzer({nowarn_function, collect/4}).
-dialyzer({nowarn_function, maybe_delete_old_generation/4}).
-dialyzer({nowarn_function, datapoint/7}).

init(#stream{name=Name,
             reader=ReaderId,
             forget=Forget}, Attributes) when is_reference(ReaderId) ->
    Generation = case Forget of
                     true ->
                         otel_metric_reader:checkpoint_generation(ReaderId);
                     _ ->
                         0
                 end,
    StartTime = opentelemetry:timestamp(),
    Key = {Name, Attributes, ReaderId, Generation},
    #sum_aggregation{key=Key,
                     start_time=StartTime,
                     checkpoint=0, %% 0 value is never reported but gets copied to previous_checkpoint
                     %% which is used to add/subtract for conversion of temporality
                     previous_checkpoint=0,
                     int_value=0,
                     float_value=0.0}.

aggregate(Ctx, Tab, ExemplarsTab, #stream{name=Name,
                                          reader=ReaderId,
                                          forget=Forget,
                                          exemplar_reservoir=ExemplarReservoir}, Value, Attributes, DroppedAttributes)
  when is_integer(Value) ->
    Generation = case Forget of
                     true ->
                         otel_metric_reader:checkpoint_generation(ReaderId);
                     _ ->
                         0
                 end,
    Key = {Name, Attributes, ReaderId, Generation},
    try ets:update_counter(Tab, Key, {#sum_aggregation.int_value, Value}) of
        _ ->
            otel_metric_exemplar_reservoir:offer(Ctx, ExemplarReservoir, ExemplarsTab, Key, Value, DroppedAttributes),
            true
    catch
        error:badarg ->
            %% the use of `update_counter' guards against conflicting with another process
            %% doing the update at the same time

            %% the default isn't just given in the first `update_counter' because then
            %% we'd have to call `system_time' for every single measurement taken
            %% _ = ets:update_counter(Tab, Key, {#sum_aggregation.value, Value},
            %%                        init(Key, Options)),
            %% true
            false
    end;
aggregate(Ctx, Tab, ExemplarsTab, #stream{name=Name,
                                          reader=ReaderId,
                                          forget=Forget,
                                          exemplar_reservoir=ExemplarReservoir}, Value, Attributes, DroppedAttributes) ->
    Generation = case Forget of
                     true ->
                         otel_metric_reader:checkpoint_generation(ReaderId);
                     _ ->
                         0
                 end,
    Key = {Name, Attributes, ReaderId, Generation},
    MS = [{#sum_aggregation{key=Key,
                            start_time='$1',
                            checkpoint='$2',
                            previous_checkpoint='$6',
                            int_value='$3',
                            float_value='$4'},
           [],
           [{#sum_aggregation{key={element, 2, '$_'},
                              start_time='$1',
                              checkpoint='$2',
                              previous_checkpoint='$6',
                              int_value='$3',
                              float_value={'+', '$4', {const, Value}}}}]}],
    case ets:select_replace(Tab, MS) of
        1 ->
            otel_metric_exemplar_reservoir:offer(Ctx, ExemplarReservoir, ExemplarsTab, Key, Value, DroppedAttributes),
            true;
        _ ->
            false
    end.

checkpoint(Tab, #stream{name=Name,
                        reader=ReaderId,
                        temporality=?TEMPORALITY_DELTA}, Generation) ->
    MS = [{#sum_aggregation{key={Name, '$1', ReaderId, Generation},
                            start_time='$4',
                            checkpoint='$5',
                            previous_checkpoint='_',
                            int_value='$2',
                            float_value='$3'},
           [{'=:=', '$3', {const, 0.0}}],
           [{#sum_aggregation{key={{Name, '$1', {const, ReaderId}, {const, Generation}}},
                              start_time='$4',
                              checkpoint='$2',
                              previous_checkpoint='$5',
                              int_value=0,
                              float_value=0.0}}]},
          {#sum_aggregation{key={Name, '$1', ReaderId, Generation},
                            start_time='$4',
                            checkpoint='$5',
                            previous_checkpoint='_',
                            int_value='$2',
                            float_value='$3'},
           [],
           [{#sum_aggregation{key={{Name, '$1', {const, ReaderId}, {const, Generation}}},
                              start_time='$4',
                              checkpoint={'+', '$2', '$3'},
                              previous_checkpoint='$5',
                              int_value=0,
                              float_value=0.0}}]}],
    _ = ets:select_replace(Tab, MS),
    ok;
checkpoint(Tab, #stream{name=Name,
                        reader=ReaderId,
                        forget=Forget,
                        temporality=?TEMPORALITY_CUMULATIVE}, Generation0) ->
    Generation = case Forget of
                     true ->
                         Generation0;
                     _ ->
                         0
                 end,
    MS = [{#sum_aggregation{key={Name, '$1', ReaderId, Generation},
                            start_time='$2',
                            checkpoint='$5',
                            previous_checkpoint='$6',
                            int_value='$3',
                            float_value='$4'},
           [{'=:=', '$4', {const, 0.0}}],
           [{#sum_aggregation{key={{Name, '$1', {const, ReaderId}, {const, Generation}}},
                              start_time='$2',
                              checkpoint='$3',
                              previous_checkpoint={'+', '$5', '$6'},
                              int_value=0,
                              float_value=0.0}}]},
          {#sum_aggregation{key={Name, '$1', ReaderId, Generation},
                            start_time='$2',
                            checkpoint='$5',
                            previous_checkpoint='$6',
                            int_value='$3',
                            float_value='$4'},
           [],
           [{#sum_aggregation{key={{Name, '$1', {const, ReaderId}, {const, Generation}}},
                              start_time='$2',
                              checkpoint={'+', '$3', '$4'},
                              previous_checkpoint={'+', '$5', '$6'},
                              int_value=0,
                              float_value=0.0}}]}],
    _ = ets:select_replace(Tab, MS),
    ok.

collect(Tab, ExemplarsTab, Stream=#stream{name=Name,
                                          reader=ReaderId,
                                          instrument=#instrument{temporality=InstrumentTemporality},
                                          temporality=Temporality,
                                          is_monotonic=IsMonotonic,
                                          forget=Forget,
                                          exemplar_reservoir=ExemplarReservoir}, Generation0) ->
    CollectionStartTime = opentelemetry:timestamp(),
    Generation = case Forget of
                     true ->
                         Generation0;
                     _ ->
                         0
                 end,

    checkpoint(Tab, Stream, Generation),

    %% eqwalizer:ignore matchspecs mess with the typing
    Select = [{#sum_aggregation{key={Name, '_', ReaderId, Generation}, _='_'}, [], ['$_']}],
    AttributesAggregation = ets:select(Tab, Select),
    Result = #sum{aggregation_temporality=Temporality,
                  is_monotonic=IsMonotonic,
                  datapoints=[datapoint(Tab, ExemplarReservoir, ExemplarsTab, CollectionStartTime, InstrumentTemporality, Temporality, SumAgg) || SumAgg <- AttributesAggregation]},

    %% would be nice to do this in the reader so its not duplicated in each aggregator
    maybe_delete_old_generation(Tab, Name, ReaderId, Generation),

    Result.

%% 0 means it is either cumulative or the first generation with nothing older to delete
maybe_delete_old_generation(_Tab, _Name, _ReaderId, 0) ->
    ok;
maybe_delete_old_generation(Tab, Name, ReaderId, Generation) ->
    %% delete all older than the Generation instead of just the previous in case a
    %% a crash had happened between incrementing the Generation counter and doing
    %% the delete in a previous collection cycle
    %% eqwalizer:ignore matchspecs mess with the typing
    Select = [{#sum_aggregation{key={Name, '_', ReaderId, '$1'}, _='_'},
               [{'<', '$1', {const, Generation}}],
               [true]}],
    ets:select_delete(Tab, Select).

%% nothing special to do if the instrument temporality and view temporality are the same
datapoint(_Tab, ExemplarReservoir, ExemplarsTab, CollectionStartTime, Temporality, Temporality, #sum_aggregation{key=Key={_, Attributes, _, _},
                                                                                                                 start_time=StartTime,
                                                                                                                 checkpoint=Value}) ->
    Exemplars = otel_metric_exemplar_reservoir:collect(ExemplarReservoir, ExemplarsTab, Key),
    #datapoint{
       attributes=Attributes,
       start_time=StartTime,
       time=CollectionStartTime,
       value=Value,
       exemplars=Exemplars,
       flags=0
      };
%% converting an instrument of delta temporality to cumulative means we need to add the
%% previous value to the current because the actual value is only a delta
datapoint(_Tab, ExemplarReservoir, ExemplarsTab, Time, _, ?TEMPORALITY_CUMULATIVE, #sum_aggregation{key=Key={_Name, Attributes, _ReaderId, _Generation},
                                                                                                    start_time=StartTime,
                                                                                                    previous_checkpoint=PreviousCheckpoint,
                                                                                                    checkpoint=Value}) ->
    Exemplars = otel_metric_exemplar_reservoir:collect(ExemplarReservoir, ExemplarsTab, Key),
    #datapoint{
       attributes=Attributes,
       start_time=StartTime,
       time=Time,
       value=Value + PreviousCheckpoint,
       exemplars=Exemplars,
       flags=0
      };
%% converting an instrument of cumulative temporality to delta means subtracting the
%% value of the previous collection, if one exists.
%% because we use a generation counter to reset delta aggregates the previous value
%% has to be looked up with an ets lookup of the previous generation
datapoint(Tab, ExemplarReservoir, ExemplarsTab, Time, _, ?TEMPORALITY_DELTA, #sum_aggregation{key=Key={Name, Attributes, ReaderId, Generation},
                                                                                              start_time=StartTime,
                                                                                              checkpoint=Value}) ->
    %% converting from cumulative to delta by grabbing the last generation and subtracting it
    %% can't use `previous_checkpoint' because with delta metrics have their generation changed
    %% at each collection
    PreviousCheckpoint =
        otel_metrics_tables:lookup_sum_checkpoint(Tab, Name, Attributes, ReaderId, Generation-1),
    Exemplars = otel_metric_exemplar_reservoir:collect(ExemplarReservoir, ExemplarsTab, Key),
    #datapoint{
       attributes=Attributes,
       start_time=StartTime,
       time=Time,
       value=Value - PreviousCheckpoint,
       exemplars=Exemplars,
       flags=0
      }.