Merge pull request #697 from tsloughter/observe-exemplars

Add a fresh context to each observable callback and test observe exemplars

Author: tsloughter

apps/opentelemetry_api/include/otel_ctx.hrl

@@ -0,0 +1,4 @@
+-define(current_ctx, otel_ctx:get_current()).
+
+-define(with_ctx(Ctx, Fun),
+        otel_ctx:with_ctx(Ctx, Fun)).

apps/opentelemetry_api/src/otel_ctx.erl

@@ -37,6 +37,7 @@
 
          attach/1,
          detach/1,
+         with_ctx/2,
          get_current/0,
 
          text_map_extractor/2,
@@ -159,12 +160,27 @@ attach(Ctx) ->
     erlang:put(?CURRENT_CTX, Ctx).
 
 %% @doc Detaches the given context from the current process.
--spec detach(token()) -> ok.
+-spec detach(token()) -> t() | undefined.
 detach(Token) ->
     %% at this time `Token' is a context
     update_logger_process_metadata(Token),
     erlang:put(?CURRENT_CTX, Token).
 
+%% @doc Attaches a context and runs a function, detaching the context at the end.
+%%
+%% Returns the detached context.
+-spec with_ctx(t(), fun(() -> term())) -> {term(), t()}.
+with_ctx(Ctx, Fun) ->
+    Token = otel_ctx:attach(Ctx),
+    try
+        Result = Fun(),
+        {Result, otel_ctx:detach(Token)}
+    catch
+        C:T:S ->
+            otel_ctx:detach(Token),
+            erlang:raise(C, T, S)
+    end.
+
 
 %% Extractor and Injector setup functions
 

apps/opentelemetry_experimental/src/otel_aggregation.erl

@@ -71,6 +71,10 @@ default_mapping() ->
 
 split(Keys, Map) ->
     lists:foldl(fun(Key, {KeptAcc, DroppedAcc}) ->
-                        {Value, DroppedAcc1} = maps:take(Key, DroppedAcc),
-                        {KeptAcc#{Key => Value}, DroppedAcc1}
+                        case maps:take(Key, DroppedAcc) of
+                            {Value, DroppedAcc1} ->
+                                {KeptAcc#{Key => Value}, DroppedAcc1};
+                            error ->
+                                {KeptAcc, DroppedAcc}
+                        end
                 end, {#{}, Map}, Keys).

apps/opentelemetry_experimental/src/otel_meter_server.erl

@@ -37,6 +37,8 @@
 -export([start_link/4,
          add_metric_reader/4,
          add_metric_reader/5,
+         get_readers/0,
+         get_readers/1,
          add_instrument/1,
          add_instrument/2,
          register_callback/3,
@@ -129,6 +131,12 @@ add_metric_reader(ReaderId, ReaderPid, DefaultAggregationMapping, Temporality) -
 add_metric_reader(Provider, ReaderId, ReaderPid, DefaultAggregationMapping, Temporality) ->
     gen_server:call(Provider, {add_metric_reader, ReaderId, ReaderPid, DefaultAggregationMapping, Temporality}).
 
+get_readers() ->
+    get_readers(?GLOBAL_METER_PROVIDER_REG_NAME).
+
+get_readers(Provider) ->
+    gen_server:call(Provider, get_readers).
+
 -spec register_callback([otel_instrument:t()], otel_instrument:callback(), otel_instrument:callback_args()) -> boolean().
 register_callback(Instruments, Callback, CallbackArgs) ->
     register_callback(?GLOBAL_METER_PROVIDER_REG_NAME, Instruments, Callback, CallbackArgs).
@@ -208,6 +216,9 @@ init_producers(ProducerConfigs) ->
                             end
                     end, ProducerConfigs).
 
+handle_call(get_readers, _From, State=#state{readers=Readers}) ->
+
+    {reply, Readers, State};
 handle_call({add_metric_reader, ReaderId, ReaderPid, DefaultAggregationMapping, Temporality},
             _From, State=#state{readers=Readers,
                                 views=Views,

apps/opentelemetry_experimental/src/otel_metric_exemplar_reservoir_aligned_histogram.erl

@@ -58,9 +58,12 @@ offer(Ctx, ExemplarsTab, Key, Value, FilteredAttributes, #state{bucket_boundarie
 %% @doc Return all exemplars for a `Key' and then delete them.
 -spec collect(ets:table(), term(), #state{}) -> [otel_metric_exemplar:exemplar()].
 collect(ExemplarsTab, Key, _State) ->
-    Exemplars = [E || [E] <- ets:match(ExemplarsTab, {{Key, '_'}, '$1'})],
+    Exemplars = ets:select(ExemplarsTab, [{{{'$1', '_'}, '$2'},
+                                           [{'==', '$1', {const, Key}}],
+                                           ['$2']}]),
 
-    _ = ets:match_delete(ExemplarsTab, {{Key, '_'}, '_'}),
+    _ = ets:select_delete(ExemplarsTab, [{{'$1', '_'}, [{'==', '$1', {const, Key}}], [true]},
+                                         {{{'$1', '_'}, '_'}, [{'==', '$1', {const, Key}}], [true]}]),
 
     Exemplars.
 

apps/opentelemetry_experimental/src/otel_metric_exemplar_reservoir_simple.erl

@@ -67,10 +67,12 @@ offer(Ctx, ExemplarsTab, Key, Value, FilteredAttributes, #state{num_buckets=NumB
 %% @doc Return all exemplars for a `Key' and then delete them.
 -spec collect(ets:table(), term(), #state{}) -> [otel_metric_exemplar:exemplar()].
 collect(ExemplarsTab, Key, _State) ->
-    Exemplars = [E || [E] <- ets:match(ExemplarsTab, {{Key, '_'}, '$1'})],
+    Exemplars = ets:select(ExemplarsTab, [{{{'$1', '_'}, '$2'},
+                                           [{'==', '$1', {const, Key}}],
+                                           ['$2']}]),
 
-    _ = ets:select_delete(ExemplarsTab, [{{Key, '_'}, [], [true]},
-                                         {{{Key, '_'}, '_'}, [], [true]}]),
+    _ = ets:select_delete(ExemplarsTab, [{{'$1', '_'}, [{'==', '$1', {const, Key}}], [true]},
+                                         {{{'$1', '_'}, '_'}, [{'==', '$1', {const, Key}}], [true]}]),
 
     Exemplars.
 

apps/opentelemetry_experimental/src/otel_observables.erl

@@ -22,6 +22,7 @@
 -include_lib("kernel/include/logger.hrl").
 -include_lib("opentelemetry_api_experimental/include/otel_metrics.hrl").
 -include("otel_view.hrl").
+-include_lib("opentelemetry_api/include/otel_ctx.hrl").
 
 -type callbacks() :: [{otel_instrument:callback(), otel_instrument:callback_args(), otel_instrument:t()}].
 
@@ -30,8 +31,11 @@
 run_callbacks(Callbacks, ReaderId, StreamTab, MetricsTab, ExemplarsTab) ->
     lists:foreach(fun({Callback, CallbackArgs, Instruments})
                         when is_list(Instruments) ->
-                          Ctx = otel_ctx:new(),
-                          Results = Callback(CallbackArgs),
+                          Ctx0 = otel_ctx:new(),
+                          {Results, Ctx} = ?with_ctx(Ctx0, fun() ->
+                                                                   Results0 = Callback(CallbackArgs),
+                                                                   Results0
+                                                           end),
                           handle_instruments_observations(Ctx,
                                                           Results,
                                                           Instruments,
@@ -40,8 +44,11 @@ run_callbacks(Callbacks, ReaderId, StreamTab, MetricsTab, ExemplarsTab) ->
                                                           ExemplarsTab,
                                                           ReaderId);
                      ({Callback, CallbackArgs, Instrument}) ->
-                          Ctx = otel_ctx:new(),
-                          Results = Callback(CallbackArgs),
+                          Ctx0 = otel_ctx:new(),
+                          {Results, Ctx} = ?with_ctx(Ctx0, fun() ->
+                                                                   Results0 = Callback(CallbackArgs),
+                                                                   Results0
+                                                           end),
                           %% when not a list of instruments it isn't expecting named observation
                           %% results so we use handle_instrument instead of handle_instruments
                           %% but we can't type that correctly so have to use a `fixme'