Merge pull request #701 from tsloughter/metric-producers

Metric producers

Author: tsloughter

apps/opentelemetry/src/otel_configuration.erl

@@ -53,6 +53,7 @@
                readers := [#{id := atom(), module => module(), config => map()}],
                exemplars_enabled := boolean(),
                exemplar_filter := always_on | always_off | trace_based,
+               metric_producers := [{module(), term()}],
 
                processors := list(),
                sampler := {atom(), term()},
@@ -94,6 +95,7 @@ new() ->
                    readers => [],
                    exemplars_enabled => false,
                    exemplar_filter => trace_based,
+                   metric_producers => [],
                    processors => [{otel_batch_processor, ?BATCH_PROCESSOR_DEFAULTS}],
                    sampler => {parent_based, #{root => always_on}},
                    sweeper => #{interval => timer:minutes(10),
@@ -318,6 +320,7 @@ config_mappings(general_sdk) ->
      {"OTEL_METRIC_READERS", readers, readers},
      {"OTEL_ERLANG_X_EXEMPLARS_ENABLED", exemplars_enabled, boolean},
      {"OTEL_METRICS_EXEMPLAR_FILTER", exemplar_filter, exemplar_filter},
+     {"OTEL_ERLANG_X_METRIC_PRODUCERS", metric_producers, metric_producers},
      {"OTEL_RESOURCE_DETECTORS", resource_detectors, existing_atom_list},
      {"OTEL_RESOURCE_DETECTOR_TIMEOUT", resource_detector_timeout, integer},
 
@@ -514,6 +517,14 @@ transform(exemplar_filter, ExemplarFilter) when ExemplarFilter =:= "always_off"
     always_off;
 transform(exemplar_filter, _) ->
     trace_based;
+transform(metric_producers, MetricProducers) when is_list(MetricProducers) ->
+    lists:filtermap(fun({ProducerModule, ProducerConfig}) when is_atom(ProducerModule) ->
+                            {true, {ProducerModule, ProducerConfig}};
+                       (ProducerModule) when is_atom(ProducerModule) ->
+                            {true, {ProducerModule, []}};
+                       (_) ->
+                            false
+                    end, MetricProducers);
 transform(views, Views) ->
     Views.
 

apps/opentelemetry/src/otel_simple_processor.erl

@@ -266,12 +266,4 @@ report_cb(#{source := exporter,
             exporter := ExporterModule,
             stacktrace := StackTrace}) ->
     {"exporter threw exception: exporter=~p ~ts",
-     [ExporterModule, format_exception(Kind, Reason, StackTrace)]}.
-
--if(?OTP_RELEASE >= 24).
-format_exception(Kind, Reason, StackTrace) ->
-    erl_error:format_exception(Kind, Reason, StackTrace).
--else.
-format_exception(Kind, Reason, StackTrace) ->
-    io_lib:format("~p:~p ~p", [Kind, Reason, StackTrace]).
--endif.
+     [ExporterModule, otel_utils:format_exception(Kind, Reason, StackTrace)]}.

apps/opentelemetry_api/src/otel_span.erl

@@ -72,7 +72,7 @@ validate_start_opts(Opts) when is_map(Opts) ->
 
 %% @doc Returns whether the span is recording.
 -spec is_recording(SpanCtx) -> boolean() when
-      SpanCtx :: opentelemetry:span_ctx().
+      SpanCtx :: opentelemetry:span_ctx() | undefined.
 is_recording(SpanCtx) ->
     ?is_recording(SpanCtx).
 

apps/opentelemetry_experimental/include/otel_metrics.hrl

@@ -114,7 +114,7 @@
 -record(metric,
         {
          name :: otel_view:name(),
-         scope :: opentelemetry:instrumentation_scope(),
+         scope :: opentelemetry:instrumentation_scope() | undefined,
          description :: otel_instrument:description(),
          unit :: otel_instrument:unit(),
          data :: #sum{} | #gauge{} | #histogram{}

apps/opentelemetry_experimental/src/otel_aggregation.erl

@@ -1,18 +1,16 @@
 -module(otel_aggregation).
 
 -export([maybe_init_aggregate/6,
-         default_mapping/0,
-         ets_lookup_element/4]).
+         default_mapping/0]).
 
 -include_lib("opentelemetry_api_experimental/include/otel_metrics.hrl").
--include("otel_metrics.hrl").
 -include("otel_view.hrl").
 
 %% -type t() :: drop | sum | last_value | histogram.
 -type t() :: otel_aggregation_drop:t() | otel_aggregation_sum:t() |
              otel_aggregation_last_value:t() | otel_aggregation_histogram_explicit:t().
 
--type key() :: {atom(), opentelemetry:attributes_map(), reference() | undefined, number()}.
+-type key() :: {atom(), opentelemetry:attributes_map(), reference(), number()}.
 
 -type options() :: map().
 
@@ -43,8 +41,8 @@
       Generation :: integer().
 
 maybe_init_aggregate(Ctx, MetricsTab, ExemplarsTab, Stream=#stream{aggregation_module=AggregationModule,
-                                                                           attribute_keys=AttributeKeys},
-                             Value, Attributes) ->
+                                                                   attribute_keys=AttributeKeys},
+                     Value, Attributes) ->
     {FilteredAttributes, DroppedAttributes} = filter_attributes(AttributeKeys, Attributes),
     case AggregationModule:aggregate(Ctx, MetricsTab, ExemplarsTab, Stream, Value, FilteredAttributes, DroppedAttributes) of
         true ->
@@ -76,16 +74,3 @@ split(Keys, Map) ->
                         {Value, DroppedAcc1} = maps:take(Key, DroppedAcc),
                         {KeptAcc#{Key => Value}, DroppedAcc1}
                 end, {#{}, Map}, Keys).
-
--if(?OTP_RELEASE >= 26).
-ets_lookup_element(Tab, Key, Pos, Default) ->
-    ets:lookup_element(Tab, Key, Pos, Default).
--else.
-ets_lookup_element(Tab, Key, Pos, Default) ->
-    try
-        ets:lookup_element(Tab, Key, Pos)
-    catch
-        error:badarg ->
-            Default
-    end.
--endif.

apps/opentelemetry_experimental/src/otel_aggregation_histogram_explicit.erl

@@ -149,7 +149,7 @@ default_buckets() ->
 init(#stream{name=Name,
              reader=ReaderId,
              aggregation_options=Options,
-             forget=Forget}, Attributes) ->
+             forget=Forget}, Attributes) when is_reference(ReaderId) ->
     Generation = case Forget of
                      true ->
                          otel_metric_reader:checkpoint_generation(ReaderId);
@@ -174,16 +174,15 @@ aggregate(Ctx, Table, ExemplarsTab, #stream{name=Name,
                                             reader=ReaderId,
                                             aggregation_options=Options,
                                             forget=Forget,
-                                            exemplar_reservoir=ExemplarReservoir}, Value, Attributes, DroppedAttributes) ->
+                                            exemplar_reservoir=ExemplarReservoir}, Value, Attributes, DroppedAttributes) when is_reference(ReaderId) ->
     Generation = case Forget of
                      true ->
                          otel_metric_reader:checkpoint_generation(ReaderId);
                      _ ->
                          0
                  end,
-    Key = {Name, Attributes, ReaderId, Generation},
     ExplicitBucketBoundaries = maps:get(explicit_bucket_boundaries, Options, ?DEFAULT_BOUNDARIES),
-    case otel_aggregation:ets_lookup_element(Table, Key, #explicit_histogram_aggregation.bucket_counts, false) of
+    case otel_metrics_tables:lookup_explicit_histogram_bucket_counts(Table, Name, Attributes, ReaderId, Generation) of
         false ->
             %% since we need the options to initialize a histogram `false' is
             %% returned and `otel_metric_server' will initialize the histogram
@@ -199,6 +198,7 @@ aggregate(Ctx, Table, ExemplarsTab, #stream{name=Name,
             BucketIdx = find_bucket(ExplicitBucketBoundaries, Value),
             counters:add(BucketCounts, BucketIdx, 1),
 
+            Key = {Name, Attributes, ReaderId, Generation},
             MS = ?AGGREATE_MATCH_SPEC(Key, Value, BucketCounts),
             case ets:select_replace(Table, MS) of
                 1 ->

apps/opentelemetry_experimental/src/otel_aggregation_sum.erl

@@ -42,7 +42,7 @@
 
 init(#stream{name=Name,
              reader=ReaderId,
-             forget=Forget}, Attributes) ->
+             forget=Forget}, Attributes) when is_reference(ReaderId) ->
     Generation = case Forget of
                      true ->
                          otel_metric_reader:checkpoint_generation(ReaderId);
@@ -270,8 +270,7 @@ datapoint(Tab, ExemplarReservoir, ExemplarsTab, Time, _, ?TEMPORALITY_DELTA, #su
     %% can't use `previous_checkpoint' because with delta metrics have their generation changed
     %% at each collection
     PreviousCheckpoint =
-        otel_aggregation:ets_lookup_element(Tab, {Name, Attributes, ReaderId, Generation-1},
-                                            #sum_aggregation.checkpoint, 0),
+        otel_metrics_tables:lookup_sum_checkpoint(Tab, Name, Attributes, ReaderId, Generation-1),
     Exemplars = otel_metric_exemplar_reservoir:collect(ExemplarReservoir, ExemplarsTab, Key),
     #datapoint{
        attributes=Attributes,

apps/opentelemetry_experimental/src/otel_meter_default.erl

@@ -29,6 +29,7 @@
          record/4,
          record/5]).
 
+-include_lib("opentelemetry_api/include/opentelemetry.hrl").
 -include_lib("kernel/include/logger.hrl").
 -include_lib("opentelemetry_api_experimental/include/otel_metrics.hrl").
 -include("otel_metrics.hrl").
@@ -44,14 +45,8 @@ create_instrument(Meter, Name, Kind, Opts) ->
     _ = otel_meter_server:add_instrument(Provider, Instrument),
     Instrument.
 
-lookup_instrument(Meter={_, #meter{instruments_tab=Tab}}, Name) ->
-    try ets:lookup_element(Tab, {Meter, Name}, 2) of
-        Instrument ->
-            Instrument
-    catch
-        _:_ ->
-            undefined
-    end.
+lookup_instrument({_, Meter=#meter{instruments_tab=InstrumentsTab}}, Name) ->
+    otel_metrics_tables:lookup_instrument(InstrumentsTab, Meter, Name).
 
 -spec create_instrument(otel_meter:t(), otel_instrument:name(), otel_instrument:kind(), otel_instrument:callback(), otel_instrument:callback_args(), otel_instrument:opts()) -> otel_instrument:t().
 create_instrument(Meter, Name, Kind, Callback, CallbackArgs, Opts) ->
@@ -67,7 +62,10 @@ register_callback({_, #meter{provider=Provider}}, Instruments, Callback, Callbac
 register_callback(_, _, _, _) ->
     ok.
 
+-spec scope({module(), #meter{}} | #meter{}) -> opentelemetry:instrumentation_scope() | undefined.
 scope({_, #meter{instrumentation_scope=Scope}}) ->
+    Scope;
+scope(#meter{instrumentation_scope=Scope}) ->
     Scope.
 
 validate_name(Name) when is_atom(Name) ->
@@ -116,15 +114,11 @@ validate_explicit_bucket_boundaries(Name, Value) ->
 record(Ctx, Instrument=#instrument{}, Number) ->
     record(Ctx, Instrument, Number, #{}).
 
-record(Ctx, Meter={_,#meter{}}, NameOrInstrument, Number) ->
+record(Ctx, Meter={_, #meter{}}, NameOrInstrument, Number) ->
     record(Ctx, Meter, NameOrInstrument, Number, #{});
 
-record(Ctx, Instrument=#instrument{meter={_, #meter{streams_tab=StreamTab,
-                                                    metrics_tab=MetricsTab,
-                                                    exemplars_tab=ExemplarsTab}}}, Number, Attributes) ->
-    otel_meter_server:record(Ctx, StreamTab, MetricsTab, ExemplarsTab, Instrument, Number, Attributes).
+record(Ctx, Instrument=#instrument{meter={_, Meter=#meter{}}}, Number, Attributes) ->
+    otel_meter_server:record(Ctx, Meter, Instrument, Number, Attributes).
 
-record(Ctx, Meter={_, #meter{streams_tab=StreamTab,
-                             metrics_tab=MetricsTab,
-                             exemplars_tab=ExemplarsTab}}, NameOrInstrument, Number, Attributes) ->
-    otel_meter_server:record(Ctx, Meter, StreamTab, MetricsTab, ExemplarsTab, NameOrInstrument, Number, Attributes).
+record(Ctx, {_, Meter}, NameOrInstrument, Number, Attributes) ->
+    otel_meter_server:record(Ctx, Meter, NameOrInstrument, Number, Attributes).