Merge remote-tracking branch 'jinnipark/feature-manager-killer' into blt-overload_lager_event

Signed-off-by: Brian L. Troutwine <brian@troutwine.us>

rebar.config

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

src/lager_app.erl

@@ -134,7 +134,7 @@ start_error_logger_handler(_, HWM, {ok, WhiteList}) ->
     GlStrategy = case application:get_env(lager, error_logger_groupleader_strategy) of
                     undefined ->
                         handle;
-                    {ok, GlStrategy0} when 
+                    {ok, GlStrategy0} when
                             GlStrategy0 =:= handle;
                             GlStrategy0 =:= ignore;
                             GlStrategy0 =:= mirror ->
@@ -146,6 +146,25 @@ start_error_logger_handler(_, HWM, {ok, WhiteList}) ->
                         throw({error, bad_config})
                 end,
 
+    KillerHWM = case application:get_env(lager, killer_hwm) of
+                    undefined -> -1;
+                    {ok, undefined} -> -1;
+                    {ok, KHWM} when is_integer(KHWM), KHWM >= 0 -> KHWM
+                end,
+    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,
+                               [?DEFAULT_SINK, lager_manager_killer,
+                                [KillerHWM, KillerReinstallAfter]]),
+
     case supervisor:start_child(lager_handler_watcher_sup, [error_logger, error_logger_lager_h, [HWM, GlStrategy]]) of
         {ok, _} ->
             [begin error_logger:delete_report_handler(X), X end ||
@@ -154,6 +173,7 @@ start_error_logger_handler(_, HWM, {ok, WhiteList}) ->
             []
     end.
 
+
 %% `determine_async_behavior/3' is called with the results from either
 %% `application:get_env/2' and `proplists:get_value/2'. Since
 %% `application:get_env/2' wraps a successful retrieval in an `{ok,

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, sink=Sink} = State) ->
+    % Brutally kill the manager but stay alive to restore settings.
+    Manager = whereis(Sink),
+    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, sink=Sink} = 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}.