Introduce hard span storage size limit in sweeper (#121)

introduce hard span storage size limit in sweeper

Author: deadtrickster

elvis.config

@@ -11,6 +11,7 @@
                   {elvis_style, dont_repeat_yourself, #{min_complexity => 20}},
                   {elvis_style, line_length, #{limit => 120}},
                   {elvis_style, state_record_and_type, disable},
+                  {elvis_style, no_if_expression, disable},
                   {elvis_style, function_naming_convention, #{regex => "^_{0,2}([a-z][a-z0-9]*_?)*_{0,2}$"}},
                   %% sequential reporter calls other reporters dynamically
                   %% stat view proxies measures

src/oc_span_sweeper.erl

@@ -29,7 +29,8 @@
 
 -record(data, {interval :: integer() | infinity,
                strategy :: drop | finish | failed_attribute_and_finish | fun((opencensus:span()) -> ok),
-               ttl :: integer()}).
+               ttl :: integer() | infinity,
+               storage_size :: integer() | infinity}).
 
 start_link() ->
     gen_statem:start_link({local, ?MODULE}, ?MODULE, [], []).
@@ -40,54 +41,84 @@ init([]) ->
     Interval = maps:get(interval, SweeperConfig, timer:minutes(5)),
     Strategy = maps:get(strategy, SweeperConfig, drop),
     TTL = maps:get(span_ttl, SweeperConfig, timer:minutes(5)),
+    StorageSize = maps:get(storage_size, SweeperConfig, infinity),
     {ok, ready, #data{interval=Interval,
                       strategy=Strategy,
-                      ttl=erlang:convert_time_unit(TTL, millisecond, native)},
+                      ttl=maybe_convert_time_unit(TTL),
+                      storage_size=StorageSize},
      [hibernate, {state_timeout, Interval, sweep}]}.
 
+
+maybe_convert_time_unit(infinity) ->
+    infinity;
+maybe_convert_time_unit(TTL) ->
+    erlang:convert_time_unit(TTL, millisecond, native).
+
 callback_mode() ->
     handle_event_function.
 
-handle_event(state_timeout, sweep, _, #data{interval=Interval,
-                                            strategy=drop,
-                                            ttl=TTL}) ->
+handle_event(state_timeout, sweep, _, #data{interval=Interval} = Data) ->
+    do_gc(Data),
+    {keep_state_and_data, [hibernate, {state_timeout, Interval, sweep}]};
+handle_event(_, _, _, _Data) ->
+    keep_state_and_data.
+
+code_change(_, State, Data, _) ->
+    {ok, State, Data}.
+
+terminate(_Reason, _State, _Data) ->
+    ok.
+
+%%
+do_gc(#data{strategy=Strategy,
+            ttl=TTL,
+            storage_size=infinity}) ->
+    sweep_spans(Strategy, TTL);
+do_gc(#data{strategy=Strategy,
+            ttl=TTL,
+            storage_size=MaxSize}) ->
+
+    StorageSize = ets:info(?SPAN_TAB, memory) * erlang:system_info({wordsize, external}),
+
+    if
+        StorageSize >= 2 * MaxSize ->
+            %% High overload kill storage.
+            ets:delete_all_objects(?SPAN_TAB);
+        StorageSize >= MaxSize ->
+            %% Low overload, reduce TTL
+            sweep_spans(Strategy, overload_ttl(TTL));
+        true ->
+            sweep_spans(Strategy, TTL)
+    end.
+
+overload_ttl(infinity) ->
+    infinity;
+overload_ttl(TTL) ->
+    TTL div 10.
+
+sweep_spans(_, infinity) ->
+    ok;
+sweep_spans(drop, TTL) ->
     TooOld = erlang:monotonic_time() - TTL,
     case ets:select_delete(?SPAN_TAB, expired_match_spec(TooOld, true)) of
         0 ->
             ok;
         NumDeleted ->
             ?LOG_INFO("sweep old spans: ttl=~p num_dropped=~p", [TTL, NumDeleted])
-    end,
-    {keep_state_and_data, [hibernate, {state_timeout, Interval, sweep}]};
-handle_event(state_timeout, sweep, _, #data{interval=Interval,
-                                            strategy=finish,
-                                            ttl=TTL}) ->
+    end;
+sweep_spans(finish, TTL) ->
     Expired = select_expired(TTL),
     [finish_span(Span) || Span <- Expired],
-    {keep_state_and_data, [hibernate, {state_timeout, Interval, sweep}]};
-handle_event(state_timeout, sweep, _, #data{interval=Interval,
-                                            strategy=failed_attribute_and_finish,
-                                            ttl=TTL}) ->
+    ok;
+sweep_spans(failed_attribute_and_finish, TTL) ->
     Expired = select_expired(TTL),
     [finish_span(oc_span:put_attribute(<<"finished_by_sweeper">>, true, Span)) || Span <- Expired],
-    {keep_state_and_data, [hibernate, {state_timeout, Interval, sweep}]};
-handle_event(state_timeout, sweep, _, #data{interval=Interval,
-                                            strategy=Fun,
-                                            ttl=TTL}) when is_function(Fun) ->
+    ok;
+sweep_spans(Fun, TTL) when is_function(Fun) ->
     Expired = select_expired(TTL),
     [Fun(Span) || Span <- Expired],
-    {keep_state_and_data, [hibernate, {state_timeout, Interval, sweep}]};
-handle_event(_, _, _, _Data) ->
-    keep_state_and_data.
-
-code_change(_, State, Data, _) ->
-    {ok, State, Data}.
-
-terminate(_Reason, _State, _Data) ->
     ok.
 
-%%
-
 %% ignore these functions because dialyzer doesn't like match spec use of '_'
 -dialyzer({nowarn_function, expired_match_spec/2}).
 -dialyzer({nowarn_function, finish_span/1}).

test/oc_sweeper_SUITE.erl

@@ -14,7 +14,8 @@
 -include("opencensus.hrl").
 
 all() ->
-    [drop,
+    [storage_size,
+     drop,
      finish,
      failed_attribute_and_finish].
 
@@ -25,6 +26,17 @@ init_per_suite(Config) ->
 end_per_suite(_Config) ->
     ok.
 
+init_per_testcase(storage_size, Config) ->
+    application:set_env(opencensus, sweeper, #{interval => 250,
+                                               strategy => finish,
+                                               span_ttl => 500,
+                                               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()}),
+    {ok, _} = application:ensure_all_started(opencensus),
+    Config;
 init_per_testcase(Type, Config) ->
     application:set_env(opencensus, sweeper, #{interval => 250,
                                                strategy => Type,
@@ -40,6 +52,32 @@ end_per_testcase(_, _Config) ->
     ok = application:stop(opencensus),
     ok.
 
+storage_size(_Config) ->
+    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),
+
+    %% wait until the sweeper sweeps away the parent span
+    ?UNTIL(ets:tab2list(?SPAN_TAB) =:= []),
+
+    %% sleep long enough that the reporter would have run again for sure
+    timer:sleep(10),
+
+    %% should be no reported spans
+    ?assertEqual(no_span, receive
+                              {span, #span{name=N}} when N =:= SpanName1 ->
+                                  got_span
+                          after
+                              0 ->
+                                  no_span
+                          end).
+
 drop(_Config) ->
     SpanName1 = <<"span-1">>,
     SpanCtx = oc_trace:start_span(SpanName1, undefined),