Current section
Files
Jump to
Current section
Files
src/otel_aggregation_histogram_explicit.erl
%%%------------------------------------------------------------------------
%% 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_histogram_explicit).
-export([init/2,
aggregate/3,
checkpoint/5,
collect/4]).
-include("otel_metrics.hrl").
-type t() :: #explicit_histogram_aggregation{}.
-export_type([t/0]).
-define(DEFAULT_BOUNDARIES, [0.0, 5.0, 10.0, 25.0, 50.0, 75.0, 100.0, 250.0, 500.0, 1000.0]).
-define(MIN_DOUBLE, -9223372036854775807.0). %% the proto representation of size `fixed64'
init(Key, Options) ->
Boundaries = maps:get(boundaries, Options, ?DEFAULT_BOUNDARIES),
RecordMinMax = maps:get(record_min_max, Options, true),
#explicit_histogram_aggregation{key=Key,
start_time_unix_nano=erlang:system_time(nanosecond),
boundaries=Boundaries,
zeroed_counts=zero_buckets(length(Boundaries)),
bucket_counts=zero_buckets(length(Boundaries)),
record_min_max=RecordMinMax,
min=infinity, %% works because any atom is > any integer
max=?MIN_DOUBLE,
sum=0
}.
aggregate(Table, Key, Value) ->
case ets:lookup(Table, Key) of
[Current] ->
%% TODO: needs to be changed to work with concurrent updates
%% at this time there are no concurrent updates, so nothing is lost
%% a basic compare and swap could fail
%% consider making each bucket its own record
ets:insert(Table, aggregate(Value, Current));
_ ->
%% since we need the options to initialize a histogram `false' is
%% returned and `otel_metric_server' will initialize the histogram
false
end.
aggregate(MeasurementValue,
Aggregation=#explicit_histogram_aggregation{record_min_max=true,
boundaries=Boundaries,
bucket_counts=Buckets,
min=Min,
max=Max,
sum=Sum}) ->
Buckets1 = bump_bucket_counts(MeasurementValue, Boundaries, Buckets),
Aggregation#explicit_histogram_aggregation{bucket_counts=Buckets1,
min=min(Min, MeasurementValue),
max=max(Max, MeasurementValue),
sum=Sum+MeasurementValue};
aggregate(MeasurementValue,
Aggregation=#explicit_histogram_aggregation{boundaries=Boundaries,
bucket_counts=Buckets,
sum=Sum}) ->
Buckets1 = bump_bucket_counts(MeasurementValue, Boundaries, Buckets),
Aggregation#explicit_histogram_aggregation{bucket_counts=Buckets1,
sum=Sum+MeasurementValue}.
-dialyzer({nowarn_function, checkpoint/5}).
%% TODO: handle delta temporary checkpoints
checkpoint(Tab, Name, ?AGGREGATION_TEMPORALITY_DELTA, _, CollectionStartNano) ->
MS = [{#explicit_histogram_aggregation{key='$1',
start_time_unix_nano='_',
boundaries='$2',
record_min_max='$3',
checkpoint='_',
zeroed_counts='$4',
bucket_counts='$5',
min='$6',
max='$7',
sum='$8'
},
[{'=:=', {element, 1, '$1'}, {const, Name}}],
[{#explicit_histogram_aggregation{key='$1',
start_time_unix_nano={const, CollectionStartNano},
boundaries='$2',
record_min_max='$3',
checkpoint={#explicit_histogram_checkpoint{bucket_counts='$5',
min='$6',
max='$7',
sum='$8'}},
zeroed_counts='$4',
bucket_counts='$4',
min=infinity,
max=?MIN_DOUBLE,
sum=0}}]}],
_ = ets:select_replace(Tab, MS),
ok;
checkpoint(Tab, Name, _, _, _CollectionStartNano) ->
MS = [{#explicit_histogram_aggregation{key='$1',
start_time_unix_nano='$2',
boundaries='$3',
record_min_max='$4',
checkpoint='_',
zeroed_counts='$9',
bucket_counts='$5',
min='$6',
max='$7',
sum='$8'
},
[{'=:=', {element, 1, '$1'}, {const, Name}}],
[{#explicit_histogram_aggregation{key='$1',
start_time_unix_nano='$2',
boundaries='$3',
record_min_max='$4',
checkpoint={#explicit_histogram_checkpoint{bucket_counts='$5',
min='$6',
max='$7',
sum='$8'}},
zeroed_counts='$9',
bucket_counts='$5',
min='$6',
max='$7',
sum='$8'}}]}],
_ = ets:select_replace(Tab, MS),
ok.
collect(Tab, Name, _, CollectionStartTime) ->
Select = [{'$1',
[{'==', Name, {element, 1, {element, 2, '$1'}}}],
['$1']}],
AttributesAggregation = ets:select(Tab, Select),
[datapoint(CollectionStartTime, SumAgg) || SumAgg <- AttributesAggregation].
%%
datapoint(CollectionStartNano, #explicit_histogram_aggregation{
key={_, Attributes},
start_time_unix_nano=StartTimeUnixNano,
boundaries=Boundaries,
checkpoint=#explicit_histogram_checkpoint{bucket_counts=Buckets,
min=Min,
max=Max,
sum=Sum}
}) ->
#histogram_datapoint{
attributes=Attributes,
start_time_unix_nano=StartTimeUnixNano,
time_unix_nano=CollectionStartNano,
count=lists:sum(erlang:tuple_to_list(Buckets)),
sum=Sum,
bucket_counts=Buckets,
explicit_bounds=Boundaries,
exemplars=[],
flags=0,
min=Min,
max=Max
}.
zero_buckets(Size) ->
erlang:list_to_tuple(lists:duplicate(Size, 0)).
find_bucket(Boundaries, Value) ->
find_bucket(Boundaries, Value, 1).
find_bucket([X | _Rest], Value, Pos) when Value =< X ->
Pos;
find_bucket([_X], _Value, Pos) ->
Pos;
find_bucket([_X | Rest], Value, Pos) ->
find_bucket(Rest, Value, Pos+1);
find_bucket(_, _, Pos) ->
Pos.
bump_bucket_counts(MeasurementValue, Boundaries, Buckets) ->
Pos = find_bucket(Boundaries, MeasurementValue),
setelement(Pos, Buckets, element(Pos, Buckets) + 1).