prometheus

%% @doc
%% A Histogram tracks the size and number of events in buckets.
%% You can use Histograms for aggregatable calculation of quantiles.
%%
%% Example use cases for Histograms:
%% <ul>
%%   <li>Response latency</li>
%%   <li>Request size</li>
%% </ul>
%%
%% Histogram expects `buckets' key in a metric spec. Buckets can be:
%%   - a list of numbers in increasing order;
%%   - :default;
%%   - {:linear, start, step, count};
%%   - {:exponential, start, step, count}
%%
%% Example:
%% <pre lang="erlang">
%% -module(example_instrumenter).
%% setup() ->
%%   prometheus_histogram:declare([{name, http_request_duration_milliseconds},
%%                                 {labels, [method]},
%%                                 {buckets, [100, 300, 500, 750, 1000]},
%%                                 {help, "Http Request execution time."}]).
%%
%% instrument(Time, Method) ->
%%   %% Time must be in native units, otherwise duration_unit must be false
%%   prometheus_histogram:observe(http_request_duration_milliseconds,
%%                                [Method], Time).
%%
%% </pre>
%% @end

-module(prometheus_histogram).

%%% metric
-export([new/1,
         declare/1,
         deregister/1,
         deregister/2,
         set_default/2,
         observe/2,
         observe/3,
         observe/4,
         pobserve/6,
         observe_duration/2,
         observe_duration/3,
         observe_duration/4,
         remove/1,
         remove/2,
         remove/3,
         reset/1,
         reset/2,
         reset/3,
         value/1,
         value/2,
         value/3,
         buckets/1,
         buckets/2,
         buckets/3,
         values/2]
       ).

%%% collector
-export([deregister_cleanup/1,
         collect_mf/2,
         collect_metrics/2]).

-include("prometheus.hrl").

-behaviour(prometheus_metric).
-behaviour(prometheus_collector).

%%====================================================================
%% Macros
%%====================================================================

-define(TABLE, ?PROMETHEUS_HISTOGRAM_TABLE).
-define(BOUNDS_POS, 2).
-define(ISUM_POS, 3).
-define(FSUM_POS, 4).
-define(BUCKETS_START, 5).
-define(WIDTH, 16).

%% ets row layout
%% {Key, NBounds, Sum, Bucket1, Bucket2, ...}
%% NBounds is a list of bounds possibly converted to native units

%%====================================================================
%% Metric API
%%====================================================================

%% @doc Creates a histogram using `Spec'.
%%
%% Raises `{missing_metric_spec_key, Key, Spec}' error if required `Soec' key
%% is missing.<br/>
%% Raises `{invalid_metric_name, Name, Message}' error if metric `Name'
%% is invalid.<br/>
%% Raises `{invalid_metric_help, Help, Message}' error if metric `Help'
%% is invalid.<br/>
%% Raises `{invalid_metric_labels, Labels, Message}' error if `Labels'
%% isn't a list.<br/>
%% Raises `{invalid_label_name, Name, Message}' error if `Name' isn't a valid
%% label name.<br/>
%% Raises `{invalid_value_error, Value, Message}' error if `duration_unit' is
%% unknown or doesn't match metric name.<br/>
%% Raises `{mf_already_exists, {Registry, Name}, Message}' error if a histogram
%% with the same `Spec' already exists.
%%
%% Histogram-specific errors:<br/>
%% Raises `{no_buckets, Buckets}' error if `Buckets' are missing,
%% not a list, empty list or not known buckets spec.<br/>
%% Raises `{invalid_buckets, Buckets, Message}' error if `Buckets'
%% aren't in increasing order.<br/>
%% Raises `{invalid_bound, Bound}' error if `Bound' isn't a number.
%% @end
new(Spec) ->
  Spec1 = validate_histogram_spec(Spec),
  prometheus_metric:insert_new_mf(?TABLE, ?MODULE, Spec1).

%% @doc Creates a histogram using `Spec'.
%% If a histogram with the same `Spec' exists returns `false'.
%%
%% Raises `{missing_metric_spec_key, Key, Spec}' error if required `Soec' key
%% is missing.<br/>
%% Raises `{invalid_metric_name, Name, Message}' error if metric `Name'
%% is invalid.<br/>
%% Raises `{invalid_metric_help, Help, Message}' error if metric `Help'
%% is invalid.<br/>
%% Raises `{invalid_metric_labels, Labels, Message}' error if `Labels'
%% isn't a list.<br/>
%% Raises `{invalid_label_name, Name, Message}' error if `Name' isn't a valid
%% label name.<br/>
%% Raises `{invalid_value_error, Value, MessagE}' error if `duration_unit' is
%% unknown or doesn't match metric name.<br/>
%%
%% Histogram-specific errors:<br/>
%% Raises `{no_buckets, Buckets}' error if `Buckets' are missing,
%% not a list, empty list or not known buckets spec.<br/>
%% Raises `{invalid_buckets, Buckets, Message}' error if `Buckets'
%% aren't in increasing order.<br/>
%% Raises `{invalid_bound, Bound}' error if `Bound' isn't a number.
%% @end
declare(Spec) ->
  Spec1 = validate_histogram_spec(Spec),
  prometheus_metric:insert_mf(?TABLE, ?MODULE, Spec1).

%% @equiv deregister(default, Name)
deregister(Name) ->
  deregister(default, Name).

%% @doc
%% Removes all histogram series with name `Name' and
%% removes Metric Family from `Registry'.
%%
%% After this call new/1 for `Name' and `Registry' will succeed.
%%
%% Returns `{true, _}' if `Name' was a registered histogram.
%% Otherwise returns `{false, _}'.
%% @end
deregister(Registry, Name) ->
  try
    MF = prometheus_metric:check_mf_exists(?TABLE, Registry, Name),
    Buckets = prometheus_metric:mf_data(MF),
    prometheus_metric:deregister_mf(?TABLE, Registry, Name),
    Select = deregister_select(Registry, Name, Buckets),
    NumDeleted = ets:select_delete(?TABLE, Select),
    {true, NumDeleted > 0}
  catch
    _:_ -> {false, false}
  end.

%% @private
set_default(Registry, Name) ->
  insert_placeholders(Registry, Name, []).

%% @equiv observe(default, Name, [], Value)
observe(Name, Value) ->
  observe(default, Name, [], Value).

%% @equiv observe(default, Name, LabelValues, Value)
observe(Name, LabelValues, Value) ->
  observe(default, Name, LabelValues, Value).

%% @doc Observes the given `Value'.
%%
%% Raises `{invalid_value, Value, Message}' if `Value'
%% isn't an integer.<br/>
%% Raises `{unknown_metric, Registry, Name}' error if histogram with named
%% `Name' can't be found in `Registry'.<br/>
%% Raises `{invalid_metric_arity, Present, Expected}' error if labels count
%% mismatch.
%% @end
observe(Registry, Name, LabelValues, Value) when is_integer(Value) ->
  Key = key(Registry, Name, LabelValues),
  case ets:lookup(?TABLE, Key) of
    [Metric] ->
      BucketPosition = calculate_histogram_bucket_position(Metric, Value),
      ets:update_counter(?TABLE, Key,
                         [{?ISUM_POS, Value},
                          {?BUCKETS_START + BucketPosition, 1}]);
    [] ->
      insert_metric(Registry, Name, LabelValues, Value, fun observe/4)
  end,
  ok;
observe(Registry, Name, LabelValues, Value) when is_number(Value) ->
  Key = key(Registry, Name, LabelValues),
  case ets:lookup(?TABLE, Key) of
    [Metric] ->
      fobserve_impl(Key, Metric, Value);
    [] ->
      insert_metric(Registry, Name, LabelValues, Value,
                    fun(_, _, _, _) ->
                        observe(Registry, Name, LabelValues, Value)
                    end)
  end;
observe(_Registry, _Name, _LabelValues, Value) ->
  erlang:error({invalid_value, Value, "observe accepts only numbers"}).

%% @private
pobserve(Registry, Name, LabelValues, Buckets, BucketPos, Value) when is_integer(Value) ->
  Key = key(Registry, Name, LabelValues),
  try
    ets:update_counter(?TABLE, Key,
                       [{?ISUM_POS, Value}, {?BUCKETS_START + BucketPos, 1}])
  catch error:badarg ->
      insert_metric(Registry, Name, LabelValues, Value,
                    fun(_, _, _, _) ->
                        pobserve(Registry, Name, LabelValues, Buckets,
                                 BucketPos, Value)
                    end)
  end,
  ok;
pobserve(Registry, Name, LabelValues, Buckets, BucketPos, Value) when is_number(Value) ->
  Key = key(Registry, Name, LabelValues),
  case
    fobserve_impl(Key, Buckets, BucketPos, Value) of
    0 ->
      insert_metric(Registry, Name, LabelValues, Value,
                    fun(_, _, _, _) ->
                        fobserve_impl(Key, Buckets, BucketPos, Value)
                    end);
    1 ->
      ok
  end;
pobserve(_Registry, _Name, _LabelValues, _Buckets, _Pos,  Value) ->
  erlang:error({invalid_value, Value, "pobserve accepts only numbers"}).


%% @equiv observe_duration(default, Name, [], Fun)
observe_duration(Name, Fun) ->
  observe_duration(default, Name, [], Fun).

%% @equiv observe_duration(default, Name, LabelValues, Fun)
observe_duration(Name, LabelValues, Fun) ->
  observe_duration(default, Name, LabelValues, Fun).

%% @doc Tracks the amount of time spent executing `Fun'.
%%
%% Raises `{unknown_metric, Registry, Name}' error if histogram with named
%% `Name' can't be found in `Registry'.<br/>
%% Raises `{invalid_metric_arity, Present, Expected}' error if labels count
%% mismatch.
%% Raises `{invalid_value, Value, Message}' if `Fun'
%% isn't a function.<br/>
%% @end
observe_duration(Registry, Name, LabelValues, Fun) when is_function(Fun) ->
  Start = erlang:monotonic_time(),
  try
    Fun()
  after
    observe(Registry, Name, LabelValues, erlang:monotonic_time() - Start)
  end;
observe_duration(_Regsitry, _Name, _LabelValues, Fun) ->
  erlang:error({invalid_value, Fun, "observe_duration accepts only functions"}).

%% @equiv remove(default, Name, [])
remove(Name) ->
  remove(default, Name, []).

%% @equiv remove(default, Name, LabelValues)
remove(Name, LabelValues) ->
  remove(default, Name, LabelValues).

%% @doc Removes histogram series identified by `Registry', `Name'
%% and `LabelValues'.
%%
%% Raises `{unknown_metric, Registry, Name}' error if histogram with name
%% `Name' can't be found in `Registry'.<br/>
%% Raises `{invalid_metric_arity, Present, Expected}' error if labels count
%% mismatch.
%% @end
remove(Registry, Name, LabelValues) ->
  prometheus_metric:check_mf_exists(?TABLE, Registry, Name, LabelValues),
  case lists:flatten([ets:take(?TABLE,
                               {Registry, Name, LabelValues, Scheduler})
                      || Scheduler <- schedulers_seq()]) of
    [] -> false;
    _ -> true
  end.

%% @equiv reset(default, Name, [])
reset(Name) ->
  reset(default, Name, []).

%% @equiv reset(default, Name, LabelValues)
reset(Name, LabelValues) ->
  reset(default, Name, LabelValues).

%% @doc Resets the value of the histogram identified by `Registry', `Name'
%% and `LabelValues'.
%%
%% Raises `{unknown_metric, Registry, Name}' error if histogram with name
%% `Name' can't be found in `Registry'.<br/>
%% Raises `{invalid_metric_arity, Present, Expected}' error if labels count
%% mismatch.
%% @end
reset(Registry, Name, LabelValues) ->
  MF = prometheus_metric:check_mf_exists(?TABLE, Registry, Name, LabelValues),
  Buckets = prometheus_metric:mf_data(MF),
  UpdateSpec = generate_update_spec(Buckets),

  case lists:usort([ets:update_element(?TABLE,
                                       {Registry, Name, LabelValues, Scheduler},
                                       [{?ISUM_POS, 0}, {?FSUM_POS, 0}] ++ UpdateSpec)
                    || Scheduler <- schedulers_seq()]) of
    [_, _] -> true;
    [true] -> true;
    _ -> false
  end.

%% @equiv value(default, Name, [])
value(Name) ->
  value(default, Name, []).

%% @equiv value(default, Name, LabelValues)
value(Name, LabelValues) ->
  value(default, Name, LabelValues).

%% @doc Returns the value of the histogram identified by `Registry', `Name'
%% and `LabelValues'. If there is no histogram for `LabelValues',
%% returns `undefined'.
%%
%% If duration unit set, sum will be converted to the duration unit.
%% {@link prometheus_time. Read more here.}
%%
%% Raises `{unknown_metric, Registry, Name}' error if histogram named `Name'
%% can't be found in `Registry'.<br/>
%% Raises `{invalid_metric_arity, Present, Expected}' error if labels count
%% mismatch.
%% @end
value(Registry, Name, LabelValues) ->
  MF = prometheus_metric:check_mf_exists(?TABLE, Registry, Name, LabelValues),

  RawValues = [ets:lookup(?TABLE, {Registry, Name, LabelValues, Scheduler})
               || Scheduler <- schedulers_seq()],
  case lists:flatten(RawValues) of
    [] -> undefined;
    Values -> {reduce_buckets_counters(Values), reduce_sum(MF, Values)}
  end.

values(Registry, Name) ->
  case prometheus_metric:check_mf_exists(?TABLE, Registry, Name) of
    false -> [];
    MF ->
      DU = prometheus_metric:mf_duration_unit(MF),
      Labels = prometheus_metric:mf_labels(MF),
      Bounds = prometheus_metric:mf_data(MF),
      MFValues = load_all_values(Registry, Name, Bounds),
      [begin
         [ISum, FSum | BCounters] = reduce_label_values(LabelValues, MFValues),
         Bounds1 = lists:zipwith(fun(Bound, Bucket) ->
                                     {Bound, Bucket}
                                 end,
                                 Bounds, BCounters),
         {lists:zip(Labels, LabelValues),  Bounds1,
          prometheus_time:maybe_convert_to_du(DU, ISum + FSum)}
       end ||
        LabelValues <- collect_unique_labels(MFValues)]
  end.

%% @equiv buckets(default, Name, [])
buckets(Name) ->
  buckets(default, Name, []).

%% @equiv buckets(default, Name, LabelValues)
buckets(Name, LabelValues) ->
  buckets(default, Name, LabelValues).

%% @doc Returns buckets of the histogram identified by `Registry', `Name'
%% and `LabelValues'.
%% @end
buckets(Registry, Name, LabelValues) ->
  MF = prometheus_metric:check_mf_exists(?TABLE, Registry, Name, LabelValues),
  prometheus_metric:mf_data(MF).

%%====================================================================
%% Collector API
%%====================================================================

%% @private
deregister_cleanup(Registry) ->
  [delete_metrics(Registry, Buckets)
   || [_, _, _, _, Buckets] <- prometheus_metric:metrics(?TABLE, Registry)],
  true = prometheus_metric:deregister_mf(?TABLE, Registry),
  ok.

%% @private
collect_mf(Registry, Callback) ->
  [Callback(create_histogram(Name, Help, {CLabels, Labels, Registry, DU, Buckets})) ||
    [Name, {Labels, Help}, CLabels, DU, Buckets]
      <- prometheus_metric:metrics(?TABLE, Registry)],
  ok.

%% @private
collect_metrics(Name, {CLabels, Labels, Registry, DU, Bounds}) ->
  MFValues = load_all_values(Registry, Name, Bounds),
  [begin
     Stat = reduce_label_values(LabelValues, MFValues),
     create_histogram_metric(CLabels, Labels, DU, Bounds, LabelValues, Stat)
   end ||
    LabelValues <- collect_unique_labels(MFValues)].

%%====================================================================
%% Private Parts
%%====================================================================

validate_histogram_spec(Spec) ->
  Labels = prometheus_metric_spec:labels(Spec),
  validate_histogram_labels(Labels),
  RBuckets = prometheus_metric_spec:get_value(buckets, Spec, default),
  Buckets = prometheus_buckets:new(RBuckets),
  [{data, Buckets}|Spec].

validate_histogram_labels(Labels) ->
  [raise_error_if_le_label_found(Label) || Label <- Labels].

raise_error_if_le_label_found("le") ->
  erlang:error({invalid_metric_label_name, "le",
                "histogram cannot have a label named \"le\""});
raise_error_if_le_label_found(Label) ->
  Label.

insert_metric(Registry, Name, LabelValues, Value, CB) ->
  insert_placeholders(Registry, Name, LabelValues),
  CB(Registry, Name, LabelValues, Value).

fobserve_impl(Key, Metric, Value) ->
  Buckets = metric_buckets(Metric),
  BucketPos = calculate_histogram_bucket_position(Metric, Value),
  fobserve_impl(Key, Buckets, BucketPos, Value).

fobserve_impl(Key, Buckets, BucketPos, Value) ->
  ets:select_replace(?TABLE, generate_select_replace(Key, Buckets, BucketPos, Value)).

insert_placeholders(Registry, Name, LabelValues) ->
  MF = prometheus_metric:check_mf_exists(?TABLE, Registry, Name, LabelValues),
  MFBuckets = prometheus_metric:mf_data(MF),
  DU = prometheus_metric:mf_duration_unit(MF),
  Fun = fun (Bucket) ->
            prometheus_time:maybe_convert_to_native(DU, Bucket)
        end,
  BoundCounters = lists:duplicate(length(MFBuckets), 0),
  MetricSpec =
    [key(Registry, Name, LabelValues), lists:map(Fun, MFBuckets), 0, 0]
    ++ BoundCounters,
  ets:insert_new(?TABLE, list_to_tuple(MetricSpec)).

calculate_histogram_bucket_position(Metric, Value) ->
  Buckets = metric_buckets(Metric),
  prometheus_buckets:position(Buckets, Value).

generate_select_replace(Key, Bounds, BucketPos, Value) ->
  BoundPlaceholders = gen_query_bound_placeholders(Bounds),
  HistMatch = list_to_tuple([Key, '$2', '$3', '$4'] ++ BoundPlaceholders),
  BucketUpdate = lists:sublist(BoundPlaceholders, BucketPos)
    ++ [{'+', gen_query_placeholder(?BUCKETS_START + BucketPos), 1}]
    ++ lists:nthtail(BucketPos + 1, BoundPlaceholders),
  HistUpdate = list_to_tuple([{Key}, '$2', '$3', {'+', '$4', Value}] ++ BucketUpdate),
  [{HistMatch,
    [],
    [{HistUpdate}]}].

buckets_seq(Buckets) ->
  lists:seq(?BUCKETS_START, ?BUCKETS_START + length(Buckets) - 1).

generate_update_spec(Buckets) ->
  [{Index, 0} || Index <- buckets_seq(Buckets)].

gen_query_placeholder(Index) ->
  list_to_atom("$" ++ integer_to_list(Index)).

gen_query_bound_placeholders(Buckets) ->
                  [gen_query_placeholder(Index) || Index <- buckets_seq(Buckets)].

augment_counters([Start | Counters]) ->
  augment_counters(Counters, [Start], Start).

augment_counters([], LAcc, _CAcc) ->
  LAcc;
augment_counters([Counter | Counters], LAcc, CAcc) ->
  augment_counters(Counters, LAcc ++ [CAcc + Counter], CAcc + Counter).

metric_buckets(Metric) ->
  element(?BOUNDS_POS, Metric).

reduce_buckets_counters(Metrics) ->
  ABuckets =
    [sub_tuple_to_list(Metric, ?BUCKETS_START,
                       ?BUCKETS_START + length(metric_buckets(Metric)))
     || Metric <- Metrics],
  [lists:sum(Bucket) || Bucket <- transpose(ABuckets)].

transpose([[]|_]) -> [];
transpose(M) ->
  [lists:map(fun hd/1, M) | transpose(lists:map(fun tl/1, M))].

reduce_sum(Metrics) ->
  lists:sum([element(?ISUM_POS, Metric) + element(?FSUM_POS, Metric)
             || Metric <- Metrics]).

reduce_sum(MF, Metrics) ->
  DU = prometheus_metric:mf_duration_unit(MF),
  prometheus_time:maybe_convert_to_du(DU, reduce_sum(Metrics)).

create_histogram_metric(CLabels, Labels, DU, Bounds, LabelValues, [ISum, FSum | Buckets]) ->
  BCounters = augment_counters(Buckets),
  Bounds1 = lists:zipwith(fun(Bound, Bucket) ->
                              {Bound, Bucket}
                          end,
                          Bounds, BCounters),

  prometheus_model_helpers:histogram_metric(
    CLabels ++ lists:zip(Labels, LabelValues),
    Bounds1,
    lists:last(BCounters),
    prometheus_time:maybe_convert_to_du(DU, ISum + FSum)).

load_all_values(Registry, Name, Bounds) ->
  BoundPlaceholders = gen_query_bound_placeholders(Bounds),
  QuerySpec = [{Registry, Name, '$1', '_'}, '_', '$3', '$4'] ++ BoundPlaceholders,

  ets:match(?TABLE, list_to_tuple(QuerySpec)).

deregister_select(Registry, Name, Buckets) ->
  BoundCounters = lists:duplicate(length(Buckets), '_'),
  MetricSpec = [{Registry, Name, '_', '_'}, '_', '_', '_'] ++ BoundCounters,
  [{list_to_tuple(MetricSpec), [], [true]}].

delete_metrics(Registry, Buckets) ->
  BoundCounters = lists:duplicate(length(Buckets), '_'),
  MetricSpec = [{Registry, '_', '_', '_'}, '_', '_', '_'] ++ BoundCounters,
  ets:match_delete(?TABLE, list_to_tuple(MetricSpec)).

sub_tuple_to_list(Tuple, Pos, Size) when Pos < Size ->
  [element(Pos, Tuple) | sub_tuple_to_list(Tuple, Pos + 1, Size)];
sub_tuple_to_list(_Tuple, _Pos, _Size) -> [].

schedulers_seq() ->
  lists:seq(0, ?WIDTH-1).

key(Registry, Name, LabelValues) ->
  X = erlang:system_info(scheduler_id),
  Rnd = X band (?WIDTH-1),
  {Registry, Name, LabelValues, Rnd}.

collect_unique_labels(MFValues) ->
  lists:usort([L || [L | _] <- MFValues]).

reduce_label_values(Labels, MFValues) ->
  [lists:sum(C)
   || C <- transpose([V || [L | V] <- MFValues, L == Labels])].

create_histogram(Name, Help, Data) ->
  prometheus_model_helpers:create_mf(Name, Help, histogram, ?MODULE, Data).