Merge pull request #427 from aeternity/uw-dont-flush-notify-msgs

Dont flush notify msgs at overload

README.md

@@ -356,6 +356,39 @@ related processes crash, you can set a limit:
 
 It is probably best to keep this number small.
 
+### Event queue flushing
+
+When the high-water mark is exceeded, lager can be configured to flush all
+event notifications in the message queue. This can have unintended consequences
+for other handlers in the same event manager (in e.g. the `error_logger'), as
+events they rely on may be wrongly discarded. By default, this behavior is enabled,
+but can be controlled, for the `error_logger' via:
+
+```erlang
+{error_logger_flush_queue, true | false}
+```
+
+or for a specific sink, using the option:
+
+```erlang
+{flush_queue, true | false}
+
+If `flush_queue` is true, a message queue length threshold can be set, at which
+messages will start being discarded. The default threshold is `0`, meaning that
+if `flush_queue` is true, messages will be discarded if the high-water mark is
+exceeded, regardless of the length of the message queue. The option to control
+the threshold is, for `error_logger`:
+
+```erlang
+{error_logger_flush_threshold, 1000}
+```
+
+and for sinks:
+
+```erlang
+{flush_threshold, 1000}
+```
+
 ### Sink Killer
 
 In some high volume situations, it may be preferable to drop all pending log

include/lager.hrl

@@ -121,6 +121,9 @@
                   lasttime = os:timestamp() :: erlang:timestamp(),
                   %% count of dropped messages this second
                   dropped = 0 :: non_neg_integer(),
+                  %% If true, flush notify messages from msg queue at overload
+                  flush_queue = true :: boolean(),
+                  flush_threshold = 0 :: integer(),
                   %% timer
                   timer = make_ref() :: reference(),
                   %% optional filter fun to avoid counting suppressed messages against HWM totals

src/error_logger_lager_h.erl

@@ -72,7 +72,9 @@ set_high_water(N) ->
 
 -spec init(any()) -> {ok, #state{}}.
 init([HighWaterMark, GlStrategy]) ->
-    Shaper = #lager_shaper{hwm=HighWaterMark, filter=shaper_fun(), id=?MODULE},
+    Flush = lager_app:get_env(lager, error_logger_flush_queue, false),
+    FlushThr = lager_app:get_env(lager, error_logger_flush_threshold, 0),
+    Shaper = #lager_shaper{hwm=HighWaterMark, flush_queue = Flush, flush_threshold = FlushThr, filter=shaper_fun(), id=?MODULE},
     Raw = lager_app:get_env(lager, error_logger_format_raw, false),
     Sink = configured_sink(),
     {ok, #state{sink=Sink, shaper=Shaper, groupleader_strategy=GlStrategy, raw=Raw}}.

src/lager_file_backend.erl

@@ -108,11 +108,12 @@ init(LogFileConfig) when is_list(LogFileConfig) ->
             {error, {fatal, bad_config}};
         Config ->
             %% probabably a better way to do this, but whatever
-            [RelName, Level, Date, Size, Count, HighWaterMark, SyncInterval, SyncSize, SyncOn, CheckInterval, Formatter, FormatterConfig] =
-              [proplists:get_value(Key, Config) || Key <- [file, level, date, size, count, high_water_mark, sync_interval, sync_size, sync_on, check_interval, formatter, formatter_config]],
+            [RelName, Level, Date, Size, Count, HighWaterMark, Flush, SyncInterval, SyncSize, SyncOn, CheckInterval, Formatter, FormatterConfig] =
+              [proplists:get_value(Key, Config) || Key <- [file, level, date, size, count, high_water_mark, flush_queue, sync_interval, sync_size, sync_on, check_interval, formatter, formatter_config]],
+            FlushThr = proplists:get_value(flush_threshold, Config, 0),
             Name = lager_util:expand_path(RelName),
             schedule_rotation(Name, Date),
-            Shaper = #lager_shaper{hwm=HighWaterMark, id=Name},
+            Shaper = lager_util:maybe_flush(Flush, #lager_shaper{hwm=HighWaterMark, flush_threshold = FlushThr, id=Name}),
             State0 = #state{name=Name, level=Level, size=Size, date=Date, count=Count, shaper=Shaper, formatter=Formatter,
                 formatter_config=FormatterConfig, sync_on=SyncOn, sync_interval=SyncInterval, sync_size=SyncSize,
                 check_interval=CheckInterval},

src/lager_util.erl

@@ -29,7 +29,7 @@
     localtime_ms/0, localtime_ms/1, maybe_utc/1, parse_rotation_date_spec/1,
     calculate_next_rotation/1, validate_trace/1, check_traces/4, is_loggable/3,
     trace_filter/1, trace_filter/2, expand_path/1, find_file/2, check_hwm/1, check_hwm/2,
-    make_internal_sink_name/1, otp_version/0
+    make_internal_sink_name/1, otp_version/0, maybe_flush/2
 ]).
 
 -ifdef(TEST).
@@ -556,7 +556,12 @@ check_hwm(Shaper = #lager_shaper{lasttime = Last, dropped = Drop}) ->
     case Last of
         {M, S, N} ->
             %% still in same second, but have exceeded the high water mark
-            NewDrops = discard_messages(Now, Shaper#lager_shaper.filter, 0),
+            NewDrops = case should_flush(Shaper) of
+                           true ->
+                               discard_messages(Now, Shaper#lager_shaper.filter, 0);
+                           false ->
+                               1
+                       end,
             Timer = case erlang:read_timer(Shaper#lager_shaper.timer) of
                         false ->
                             erlang:send_after(trunc((1000000 - N)/1000), self(), {shaper_expired, Shaper#lager_shaper.id});
@@ -570,6 +575,12 @@ check_hwm(Shaper = #lager_shaper{lasttime = Last, dropped = Drop}) ->
             {true, Drop, Shaper#lager_shaper{dropped = 0, mps=1, lasttime = Now}}
     end.
 
+should_flush(#lager_shaper{flush_queue = true, flush_threshold = 0}) ->
+    true;
+should_flush(#lager_shaper{flush_queue = true, flush_threshold = T}) ->
+    {_, L} = process_info(self(), message_queue_len),
+    L > T.
+
 discard_messages(Second, Filter, Count) ->
     {M, S, _} = os:timestamp(),
     case Second of
@@ -611,6 +622,11 @@ otp_version() ->
         end),
     Vsn.
 
+maybe_flush(undefined, #lager_shaper{} = S) ->
+    S;
+maybe_flush(Flag, #lager_shaper{} = S) when is_boolean(Flag) ->
+    S#lager_shaper{flush_queue = Flag}.
+
 -ifdef(TEST).
 
 parse_test() ->