Add lager_manager_killer and tests for it.

rebar.config

@@ -10,3 +10,5 @@
 
 {cover_enabled, true}.
 {edoc_opts, [{stylesheet_file, "./priv/edoc.css"}]}.
+
+{eunit_opts, [verbose]}.

src/lager_app.erl

@@ -62,6 +62,32 @@ start(_StartType, _StartArgs) ->
             throw({error, bad_config})
     end,
 
+    case application:get_env(lager, killer_hwm) of
+        undefined ->
+            ok;
+        {ok, undefined} ->
+            undefined;
+        {ok, KillerHWM} when is_integer(KillerHWM), KillerHWM >= 0 ->
+            KillerReinstallAfter =
+            case application:get_env(lager, killer_reinstall_after) of
+                undefined ->
+                    5000;
+                {ok, undefined} ->
+                    5000;
+                {ok, V} when is_integer(V), V >= 0 ->
+                    V;
+                {ok, BadKillerReinstallAfter} ->
+                    error_logger:error_msg("Invalid value for 'cooldown': ~p~n", [BadKillerReinstallAfter]),
+                    throw({error, bad_config})
+            end,
+            _ = supervisor:start_child(lager_handler_watcher_sup,
+                    [lager_event, lager_manager_killer, [KillerHWM, KillerReinstallAfter]]),
+            ok;
+        {ok, BadKillerHWM} ->
+            error_logger:error_msg("Invalid value for 'floodline': ~p~n", [BadKillerHWM]),
+            throw({error, bad_config})
+    end,
+
     Handlers = case application:get_env(lager, handlers) of
         undefined ->
             [{lager_console_backend, info},

src/lager_handler_watcher.erl

@@ -61,6 +61,14 @@ handle_info({gen_event_EXIT, Module, normal}, #state{module=Module} = State) ->
     {stop, normal, State};
 handle_info({gen_event_EXIT, Module, shutdown}, #state{module=Module} = State) ->
     {stop, normal, State};
+handle_info({gen_event_EXIT, Module, {'EXIT', {kill_me, [KillerHWM, KillerReinstallAfter]}}},
+        #state{module=Module, event=Event} = State) ->
+    % Brutally kill the manager but stay alive to restore settings.
+    Manager = whereis(Event),
+    unlink(Manager),
+    exit(Manager, kill),
+    erlang:send_after(KillerReinstallAfter, self(), reinstall_handler),
+    {noreply, State#state{config=[KillerHWM, KillerReinstallAfter]}};
 handle_info({gen_event_EXIT, Module, Reason}, #state{module=Module,
         config=Config, event=Event} = State) ->
     case lager:log(error, self(), "Lager event handler ~p exited with reason ~s",

src/lager_manager_killer.erl

@@ -0,0 +1,44 @@
+-module(lager_manager_killer).
+-author("Sungjin Park <jinni.park@gmail.com>").
+-behavior(gen_event).
+
+-export([init/1, handle_event/2, handle_call/2, handle_info/2, terminate/2, code_change/3]).
+
+-include("lager.hrl").
+
+-record(state, {
+  killer_hwm :: non_neg_integer(),
+  killer_reinstall_after :: non_neg_integer()
+}).
+
+init([KillerHWM, KillerReinstallAfter]) ->
+  {ok, #state{killer_hwm=KillerHWM, killer_reinstall_after=KillerReinstallAfter}}.
+
+handle_call(get_loglevel, State) ->
+  {ok, {mask, ?LOG_NONE}, State};
+handle_call({set_loglevel, _Level}, State) ->
+  {ok, ok, State};
+handle_call(get_settings, State = #state{killer_hwm=KillerHWM, killer_reinstall_after=KillerReinstallAfter}) ->
+  {ok, [KillerHWM, KillerReinstallAfter], State};
+handle_call(_Request, State) ->
+  {ok, ok, State}.
+
+handle_event({log, _Message}, State = #state{killer_hwm=KillerHWM, killer_reinstall_after=KillerReinstallAfter}) ->
+  {message_queue_len, Len} = process_info(self(), message_queue_len),
+  case Len > KillerHWM of
+    true ->
+      exit({kill_me, [KillerHWM, KillerReinstallAfter]});
+    _ ->
+      {ok, State}
+  end;
+handle_event(_Event, State) ->
+  {ok, State}.
+
+handle_info(_Info, State) ->
+  {ok, State}.
+
+terminate(_Reason, _State) ->
+  ok.
+
+code_change(_OldVsn, State, _Extra) ->
+  {ok, State}.

test/lager_manager_killer_test.erl

@@ -0,0 +1,55 @@
+-module(lager_manager_killer_test).
+-author("Sungjin Park <jinni.park@gmail.com>").
+
+-compile([{parse_transform, lager_transform}]).
+
+-ifdef(TEST).
+-include_lib("eunit/include/eunit.hrl").
+
+overload_test() ->
+  application:stop(lager),
+  application:load(lager),
+  Delay = 1000, % sleep 1 sec on every log
+  KillerHWM = 10, % kill the manager if there are more than 10 pending logs
+  KillerReinstallAfter = 1000, % reinstall killer after 1 sec
+  application:set_env(lager, handlers, [{lager_slow_backend, Delay}]),
+  application:set_env(lager, async_threshold, undefined),
+  application:set_env(lager, killer_hwm, KillerHWM),
+  application:set_env(lager, killer_reinstall_after, KillerReinstallAfter),
+  ensure_started(lager),
+  lager_config:set(async, true),
+  Manager = whereis(lager_event),
+  erlang:trace(all, true, [procs]),
+  [lager:info("~p'th message", [N]) || N <- lists:seq(1,KillerHWM+2)],
+  Margin = 100,
+  receive
+    {trace, Manager, exit, killed} ->
+      ?debugFmt("Manager ~p killed", [Manager])
+  after Delay+Margin ->
+    ?assert(false)
+  end,
+  receive
+    {trace, _Sup, spawn, Pid, Fun} ->
+      ?assert(process_info(Pid, registered_name) =:= {registered_name, lager_event}),
+      ?debugFmt("Manager ~p start with ~p", [Pid, Fun]),
+      ?assertNot(lists:member(lager_manager_killer, gen_event:which_handlers(lager_event)))
+  after Margin ->
+      ?assert(false)
+  end,
+  erlang:trace(all, false, [procs]),
+  timer:sleep(KillerReinstallAfter),
+  ?assert(proplists:get_value(lager_manager_killer, gen_event:which_handlers(lager_event))),
+  ?assert(gen_event:call(lager_event, lager_manager_killer, get_settings) =:= [KillerHWM, KillerReinstallAfter]),
+  ?debugFmt("Killer reinstalled with [~p, ~p]", [KillerHWM, KillerReinstallAfter]),
+  application:stop(lager).
+
+ensure_started(App) ->
+  case application:start(App) of
+    ok ->
+      ok;
+    {error, {not_started, Dep}} ->
+      ensure_started(Dep),
+      ensure_started(App)
+  end.
+
+-endif.

test/lager_slow_backend.erl

@@ -0,0 +1,34 @@
+-module(lager_slow_backend).
+-author("Sungjin Park <jinni.park@gmail.com>").
+-behavior(gen_event).
+
+-export([init/1, handle_call/2, handle_event/2, handle_info/2, terminate/2, code_change/3]).
+
+-include("lager.hrl").
+
+-record(state, {
+  delay :: non_neg_integer()
+}).
+
+init(Delay) ->
+  {ok, #state{delay=Delay}}.
+
+handle_call(get_loglevel, State) ->
+  {ok, lager_util:config_to_mask(debug), State};
+handle_call(_Request, State) ->
+  {ok, ok, State}.
+
+handle_event({log, _Message}, State) ->
+  timer:sleep(State#state.delay),
+  {ok, State};
+handle_event(_Event, State) ->
+  {ok, State}.
+
+handle_info(_Info, State) ->
+  {ok, State}.
+
+terminate(_Reason, _State) ->
+  ok.
+
+code_change(_OldVsn, State, _Extra) ->
+  {ok, State}.