Add safety valve

src/jaeger_passage.erl

@@ -18,9 +18,11 @@
 %% Options for {@link start_tracer/3}.
 
 -type start_tracer_option() :: {reporter_id, jaeger_passage_reporter:reporter_id()}
+                             | {max_queue_len, non_neg_integer() | infinity}
                              | jaeger_passage_reporter:start_option().
 %% <ul>
-%%   <li><b>reporter_id</b>: In {@link start_tracer/3} function, it starts a reporter with this identifier. The default value is `Tracer'.</li>
+%%   <li>`reporter_id': In {@link start_tracer/3} function, it starts a reporter with this identifier. The default value is `Tracer'.</li>
+%%   <li>`max_queue_len': While the queue length of the reporter process exceeds this value, all new spans are rejected by the sampler. The default value is `1024'. Note that this value only affects the root spans (e.g., the child spans never be discarded by this limitation).</li>
 %% </ul>
 
 %%------------------------------------------------------------------------------
@@ -70,13 +72,22 @@ start_tracer(Tracer, Sampler) ->
       Tracer :: passage:tracer_id(),
       Sampler :: passage_sampler:sampler(),
       Options :: start_tracer_options().
-start_tracer(Tracer, Sampler, Options) ->
+start_tracer(Tracer, Sampler0, Options) ->
     ReporterId = proplists:get_value(reporter_id, Options, Tracer),
+    MaxQueueLen = proplists:get_value(max_queue_len, Options, 1024),
     Context = jaeger_passage_span_context,
+    Sampler1 =
+        case MaxQueueLen of
+            infinity -> Sampler0;
+            Max      ->
+                (is_integer(Max) andalso Max >= 0) orelse
+                    error(badarg, [Tracer, Sampler0, Options]),
+                jaeger_passage_sampler_queue_limit:new(Sampler0, ReporterId, Max)
+        end,
     case jaeger_passage_reporter:start(ReporterId, Options) of
         {error, Reason} -> {error, Reason};
         {ok, Reporter}  ->
-            passage_tracer_registry:register(Tracer, Context, Sampler, Reporter)
+            passage_tracer_registry:register(Tracer, Context, Sampler1, Reporter)
     end.
 
 %% @doc Stops the tracer which has been started by {@link start_tracer/3}.

src/jaeger_passage_local_ns.erl

@@ -10,6 +10,7 @@
 %%------------------------------------------------------------------------------
 -export([child_spec/0]).
 -export([reporter_name/1]).
+-export([whereis_reporter/1]).
 
 %%------------------------------------------------------------------------------
 %% Application Internal Functions
@@ -20,4 +21,8 @@ child_spec() ->
 
 -spec reporter_name(jaeger_passage_reporter:reporter_id()) -> local:otp_name().
 reporter_name(TracerId) ->
-    local:otp_name({?MODULE, {reporeter, TracerId}}).
+    local:otp_name({?MODULE, {reporter, TracerId}}).
+
+-spec whereis_reporter(jaeger_passage_reporter:reporter_id()) -> pid() | undefined.
+whereis_reporter(TracerId) ->
+    local:whereis_name({?MODULE, {reporter, TracerId}}).

src/jaeger_passage_sampler_queue_limit.erl

@@ -0,0 +1,42 @@
+%% @copyright 2017 Takeru Ohta <phjgt308@gmail.com>
+%%
+%% @private
+-module(jaeger_passage_sampler_queue_limit).
+
+-behaviour(passage_sampler).
+
+%%------------------------------------------------------------------------------
+%% Application Internal API
+%%------------------------------------------------------------------------------
+-export([new/3]).
+
+%%------------------------------------------------------------------------------
+%% 'passage_sampler' Callback API
+%%------------------------------------------------------------------------------
+-export([is_sampled/3]).
+
+%%------------------------------------------------------------------------------
+%% Application Internal Functions
+%%------------------------------------------------------------------------------
+-spec new(BaseSampler, ReporterId, non_neg_integer()) -> passage_sampler:sampler() when
+      BaseSampler :: passage_sampler:sampler(),
+      ReporterId :: jaeger_passage_reporter:reporter_id().
+new(BaseSampler, Reporter, MaxReporterQueueLen) ->
+    passage_sampler:new(?MODULE, {BaseSampler, Reporter, MaxReporterQueueLen}).
+
+%%------------------------------------------------------------------------------
+%% 'passage_sampler' Callback Functions
+%%------------------------------------------------------------------------------
+%% @private
+is_sampled({BaseSampler, ReporterId, MaxQueueLen}, OperationName, Tags) ->
+    case jaeger_passage_local_ns:whereis_reporter(ReporterId) of
+        undefined -> false;
+        Pid       ->
+            case erlang:process_info(Pid, message_queue_len) of
+                {_, N} ->
+                    (N =< MaxQueueLen) andalso
+                        passage_sampler:is_sampled(BaseSampler, OperationName, Tags);
+                _ ->
+                    false
+            end
+    end.