Allow registering multiple span reporters (#130)

hauleth · deadtrickster · commit 60077b8224c0 · 2019-02-08T09:59:47.000+01:00
Merge `oc_reporter_sequential` with `oc_reporter` to allow defining multiple reporters at once. Additionally this allows registering new reporters during runtime (no deregistering available). This makes `oc_reporter` API similar to the `oc_stats_reporter`. Marked as breaking change as: - configuration option `reporter` was removed and replaced with option `reporters`, which takes a list of `{Reporter, Options}` tuples or just `Reporter` atoms. In the second case `[]` will be used as `Options`. - removed `oc_reporter_sequential` as it is no longer needed - removed `oc_reporter_noop` as it can be replaced by setting `reporters` configuration option to `[]` (empty list, this is default) Close #116
diff --git a/src/oc_reporter.erl b/src/oc_reporter.erl
@@ -20,8 +20,12 @@
 
 -behaviour(gen_server).
 
+-compile({no_auto_import, [register/2]}).
+
 -export([start_link/0,
-         store_span/1]).
+         store_span/1,
+         register/1,
+         register/2]).
 
 -export([init/1,
          handle_call/3,
@@ -46,8 +50,7 @@
 %% until it returns.
 -callback report(nonempty_list(opencensus:span()), opts()) -> ok.
 
--record(state, {reporter :: module(),
-                reporter_config :: #{},
+-record(state, {reporters :: [{module(), term()}],
                 send_interval_ms :: integer(),
                 timer_ref :: reference()}).
 
@@ -59,6 +62,18 @@ start_link() ->
     maybe_init_ets(),
     gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
 
+%% @doc
+%% @equiv register(Reporter, []).
+%% @end
+register(Reporter) -> register(Reporter, []).
+
+%% @doc
+%% Register new traces reporter `Reporter' with `Config'.
+%% @end
+-spec register(module(), term()) -> ok.
+register(Reporter, Options) ->
+    gen_server:call(?MODULE, {register, init_reporter({Reporter, Options})}).
+
 -spec store_span(opencensus:span()) -> true | {error, invalid_span} | {error, no_report_buffer}.
 store_span(Span=#span{}) ->
     try
@@ -73,27 +88,26 @@ store_span(_) ->
 
 init(_Args) ->
     SendInterval = application:get_env(opencensus, send_interval_ms, 500),
-    {Reporter, ReporterOpts} = application:get_env(opencensus, reporter, {oc_reporter_noop, []}),
-    ReporterConfig = Reporter:init(ReporterOpts),
+    Reporters = [init_reporter(Config) || Config <- application:get_env(opencensus, reporters, [])],
     Ref = erlang:send_after(SendInterval, self(), report_spans),
-    {ok, #state{reporter=Reporter,
-                reporter_config=ReporterConfig,
+    {ok, #state{reporters=Reporters,
                 send_interval_ms=SendInterval,
                 timer_ref=Ref}}.
 
+handle_call({register, Reporter}, _From, #state{reporters=Reporters} = State) ->
+    {reply, ok, State#state{reporters=[Reporter | Reporters]}};
 handle_call(_, _From, State) ->
     {noreply, State}.
 
 handle_cast(_, State) ->
     {noreply, State}.
 
-handle_info(report_spans, State=#state{reporter=Reporter,
-                                       reporter_config=Config,
+handle_info(report_spans, State=#state{reporters=Reporters,
                                        send_interval_ms=SendInterval,
                                        timer_ref=Ref}) ->
     erlang:cancel_timer(Ref),
     Ref1 = erlang:send_after(SendInterval, self(), report_spans),
-    send_spans(Reporter, Config),
+    send_spans(Reporters),
     {noreply, State#state{timer_ref=Ref1}}.
 
 code_change(_, State, _) ->
@@ -103,6 +117,11 @@ terminate(_, #state{timer_ref=Ref}) ->
     erlang:cancel_timer(Ref),
     ok.
 
+init_reporter({Reporter, Config}) ->
+    {Reporter, Reporter:init(Config)};
+init_reporter(Reporter) when is_atom(Reporter) ->
+    {Reporter, Reporter:init([])}.
+
 maybe_init_ets() ->
     case ets:info(?BUFFER_STATUS, name) of
         undefined ->
@@ -115,7 +134,7 @@ maybe_init_ets() ->
             ok
     end.
 
-send_spans(Reporter, Config) ->
+send_spans(Reporters) ->
     [{_, Buffer}] = ets:lookup(?BUFFER_STATUS, current_buffer),
     NewBuffer = case Buffer of
                     ?BUFFER_1 ->
@@ -129,8 +148,9 @@ send_spans(Reporter, Config) ->
             ok;
         Spans ->
             ets:delete_all_objects(Buffer),
-            report(Reporter, Spans, Config)
-
+            [report(Reporter, Spans, Config)
+             || {Reporter, Config} <- Reporters],
+            ok
     end.
 
 report(undefined, _, _) ->
diff --git a/src/oc_reporter_noop.erl b/src/oc_reporter_noop.erl
diff --git a/src/oc_reporter_sequential.erl b/src/oc_reporter_sequential.erl
diff --git a/test/oc_reporter_pid.erl b/test/oc_reporter_pid.erl
@@ -22,10 +22,8 @@
 -export([init/1,
          report/2]).
 
-init(_) ->
-    application:get_env(opencensus, pid_reporter, #{}).
+init(Pid) -> Pid.
 
-report(Spans, Opts) ->
-    Pid = maps:get(pid, Opts),
+report(Spans, Pid) ->
     [Pid ! {span, Span} || Span <- Spans],
     ok.
diff --git a/test/oc_reporters_SUITE.erl b/test/oc_reporters_SUITE.erl
@@ -13,6 +13,7 @@
 
 all() ->
     [pid_reporter,
+     dynamically_register_reporter,
      sequential_reporter].
 
 init_per_suite(Config) ->
@@ -25,15 +26,16 @@ end_per_suite(_Config) ->
 
 init_per_testcase(pid_reporter, Config) ->
     application:set_env(opencensus, send_interval_ms, 1),
-    application:set_env(opencensus, reporter, {oc_reporter_pid, []}),
-    application:set_env(opencensus, pid_reporter, #{pid => self()}),
+    application:set_env(opencensus, reporters, [{oc_reporter_pid, self()}]),
+    {ok, _} = application:ensure_all_started(opencensus),
+    Config;
+init_per_testcase(dynamically_register_reporter, Config) ->
+    application:set_env(opencensus, send_interval_ms, 1),
     {ok, _} = application:ensure_all_started(opencensus),
     Config;
 init_per_testcase(sequential_reporter, Config) ->
     application:set_env(opencensus, send_interval_ms, 1),
-    application:set_env(opencensus, reporter, {oc_reporter_sequential, [{oc_reporter_pid, []},
-                                                                        {oc_reporter_pid, []}]}),
-    application:set_env(opencensus, pid_reporter, #{pid => self()}),
+    application:set_env(opencensus, reporters, [{oc_reporter_pid, self()}, {oc_reporter_pid, self()}]),
     {ok, _} = application:ensure_all_started(opencensus),
     Config.
 
@@ -65,6 +67,39 @@ pid_reporter(_Config) ->
                                                             andalso is_integer(O), S#span.start_time),
                                   ?assertMatch({ST, O} when is_integer(ST)
                                                             andalso is_integer(O), S#span.end_time)
+                          after
+                            5000 -> ct:fail("Do not received any message in requested time (5s)")
+                          end
+                  end, [SpanName1, ChildSpanName1]).
+
+dynamically_register_reporter(_Config) ->
+    oc_reporter:register(oc_reporter_pid, self()),
+
+    SpanName1 = <<"span-1">>,
+    SpanCtx = oc_trace:start_span(SpanName1, undefined),
+
+    ChildSpanName1 = <<"child-span-1">>,
+    ChildSpanCtx = oc_trace:start_span(ChildSpanName1, SpanCtx),
+
+    [ChildSpanData] = ets:lookup(?SPAN_TAB, ChildSpanCtx#span_ctx.span_id),
+    ?assertEqual(ChildSpanName1, ChildSpanData#span.name),
+    ?assertEqual(SpanCtx#span_ctx.span_id, ChildSpanData#span.parent_span_id),
+
+    oc_trace:finish_span(ChildSpanCtx),
+    oc_trace:finish_span(SpanCtx),
+
+    %% Order the spans are reported is undefined, so use a selective receive to make
+    %% sure we get them all
+    lists:foreach(fun(Name) ->
+                          receive
+                              {span, S=#span{name = Name}} ->
+                                  %% Verify the end time and duration are set when the span was finished
+                                  ?assertMatch({ST, O} when is_integer(ST)
+                                                            andalso is_integer(O), S#span.start_time),
+                                  ?assertMatch({ST, O} when is_integer(ST)
+                                                            andalso is_integer(O), S#span.end_time)
+                          after
+                            5000 -> ct:fail("Do not received any message in requested time (5s)")
                           end
                   end, [SpanName1, ChildSpanName1]).
 
@@ -90,4 +125,3 @@ sequential_reporter(_Config) ->
                          end, SortedNames), %% receive order is undefined though
 
     ?assertMatch(SortedNames, lists:sort(Received)).
-
diff --git a/test/oc_sweeper_SUITE.erl b/test/oc_sweeper_SUITE.erl
@@ -34,8 +34,7 @@ init_per_testcase(storage_size, Config) ->
                                                storage_size => 100}),
 
     application:set_env(opencensus, send_interval_ms, 1),
-    application:set_env(opencensus, reporter, {oc_reporter_pid, []}),
-    application:set_env(opencensus, pid_reporter, #{pid => self()}),
+    application:set_env(opencensus, reporters, [{oc_reporter_pid, self()}]),
     {ok, _} = application:ensure_all_started(opencensus),
     Config;
 init_per_testcase(Type, Config) ->
@@ -44,8 +43,7 @@ init_per_testcase(Type, Config) ->
                                                span_ttl => 500}),
 
     application:set_env(opencensus, send_interval_ms, 1),
-    application:set_env(opencensus, reporter, {oc_reporter_pid, []}),
-    application:set_env(opencensus, pid_reporter, #{pid => self()}),
+    application:set_env(opencensus, reporters, [{oc_reporter_pid, self()}]),
     {ok, _} = application:ensure_all_started(opencensus),
     Config.
 
@@ -104,6 +102,8 @@ drop(_Config) ->
                                       andalso is_integer(O), S#span.start_time),
             ?assertMatch({ST, O} when is_integer(ST)
                                       andalso is_integer(O), S#span.end_time)
+    after
+      1000 -> ct:fail("Do not received any message after 1s")
     end,
 
     %% sleep long enough that the reporter would have run again for sure
@@ -137,6 +137,8 @@ finish(_Config) ->
                                                             andalso is_integer(O), S#span.start_time),
                                   ?assertMatch({ST, O} when is_integer(ST)
                                                             andalso is_integer(O), S#span.end_time)
+                          after
+                            1000 -> ct:fail("Do not received any message after 1s")
                           end
                   end, [SpanName1, ChildSpanName1]).
 
@@ -167,4 +169,6 @@ failed_attribute_and_finish(_Config) ->
                                       andalso is_integer(O), S#span.start_time),
             ?assertMatch({ST, O} when is_integer(ST)
                                       andalso is_integer(O), S#span.end_time)
+    after
+      1000 -> ct:fail("Do not received any message after 1s")
     end.
diff --git a/test/oc_transform_SUITE.erl b/test/oc_transform_SUITE.erl
@@ -25,7 +25,7 @@ end_per_suite(_Config) ->
     ok.
 
 init_per_testcase(_, Config) ->
-    application:set_env(opencensus, reporter, {oc_reporter_pid, []}),
+    application:set_env(opencensus, reporters, [{oc_reporter_pid, self()}]),
     application:set_env(opencensus, pid_reporter, #{pid => self()}),
 
     {ok, _} = application:ensure_all_started(opencensus),
@@ -52,7 +52,7 @@ trace_transform(_Config) ->
                               ?assertMatch({ST, O} when is_integer(ST)
                                                       andalso is_integer(O), S#span.end_time)
                       after 1000 ->
-                              error(timeout)
+                              ct:fail("Did not receive any message in 1s")
                       end
                   end, [SpanName1, <<"oc_transform_SUITE:traced_function/0">>, <<"my_name">>]).
 
diff --git a/test/ocp_SUITE.erl b/test/ocp_SUITE.erl
@@ -30,7 +30,7 @@ init_per_testcase(_, Config) ->
     Tab = ets:new(reporter_tab, [public, {write_concurrency, true},
                                  {read_concurrency, true}, {keypos, #span.span_id}]),
     application:set_env(opencensus, send_interval_ms, 1),
-    application:set_env(opencensus, reporter, {oc_tab_reporter, []}),
+    application:set_env(opencensus, reporters, [{oc_tab_reporter, []}]),
     application:set_env(opencensus, tab_reporter, #{tid => Tab}),
     {ok, _} = application:ensure_all_started(opencensus),
     [{tid, Tab} | Config].
diff --git a/test/opencensus_SUITE.erl b/test/opencensus_SUITE.erl
@@ -27,7 +27,7 @@ init_per_testcase(_, Config) ->
     Tab = ets:new(reporter_tab, [public, {write_concurrency, true},
                                  {read_concurrency, true}, {keypos, #span.span_id}]),
     application:set_env(opencensus, send_interval_ms, 1),
-    application:set_env(opencensus, reporter, {oc_tab_reporter, []}),
+    application:set_env(opencensus, reporters, [{oc_tab_reporter, []}]),
     application:set_env(opencensus, tab_reporter, #{tid => Tab}),
     application:set_env(opencensus, sampler, {oc_sampler_always, []}),
     {ok, _} = application:ensure_all_started(opencensus),
