Merge branch 'master' of https://github.com/ericliang/lager into intg

include/lager.hrl

@@ -101,3 +101,15 @@
     end)).
 -endif.
 
+-record(lager_shaper, {
+		  %% how many messages per second we try to deliver
+		  hwm = undefined :: 'undefined' | pos_integer(),
+		  %% how many messages we've received this second
+		  mps = 0 :: non_neg_integer(),
+		  %% the current second
+		  lasttime = os:timestamp() :: erlang:timestamp(),
+		  %% count of dropped messages this second
+		  dropped = 0 :: non_neg_integer()
+		 }).
+
+-type lager_shaper() :: #lager_shaper{}.

src/error_logger_lager_h.erl

@@ -33,15 +33,8 @@
 
 -export([format_reason/1]).
 
--record(state, {
-        %% how many messages per second we try to deliver
-        hwm = undefined :: 'undefined' | pos_integer(),
-        %% how many messages we've received this second
-        mps = 0 :: non_neg_integer(),
-        %% the current second
-        lasttime = os:timestamp() :: erlang:timestamp(),
-        %% count of dropped messages this second
-        dropped = 0 :: non_neg_integer(),
+-record(state, { 
+        shaper :: lager_shaper(),
         %% group leader strategy
         groupleader_strategy :: handle | ignore | mirror
     }).
@@ -77,30 +70,26 @@ set_high_water(N) ->
 
 -spec init(any()) -> {ok, #state{}}.
 init([HighWaterMark, GlStrategy]) ->
-    {ok, #state{hwm=HighWaterMark, groupleader_strategy=GlStrategy}}.
+    Shaper = #lager_shaper{hwm=HighWaterMark},
+    {ok, #state{shaper=Shaper, groupleader_strategy=GlStrategy}}.
 
-handle_call({set_high_water, N}, State) ->
-    {ok, ok, State#state{hwm = N}};
+handle_call({set_high_water, N}, #state{shaper=Shaper} = State) ->
+	NewShaper = Shaper#lager_shaper{hwm=N},
+    {ok, ok, State#state{shaper = NewShaper}};
 handle_call(_Request, State) ->
     {ok, unknown_call, State}.
 
-handle_event(Event, #state{groupleader_strategy=GlStrategy0}=State) ->
-    case check_hwm(State) of
-        {true, NewState} when is_pid(element(2, Event)) ->
-            case element(2, Event) of
-                GL when node(GL) =/= node(), GlStrategy0 =:= ignore ->
-                    gen_event:notify({error_logger, node(GL)}, Event),
-                    {ok, State};
-                GL when node(GL) =/= node(), GlStrategy0 =:= mirror ->
-                    gen_event:notify({error_logger, node(GL)}, Event),
-                    log_event(Event, NewState);
-                _ ->
-                    log_event(Event, NewState)
-            end;
-        {true, NewState} ->
-            log_event(Event, NewState);
-        {false, NewState} ->
-            {ok, NewState}
+handle_event(Event, #state{shaper=Shaper} = State) ->
+    case lager_util:check_hwm(Shaper) of
+        {true, 0, NewShaper} ->
+            eval_gl(Event, State#state{shaper=NewShaper});
+        {true, Drop, #lager_shaper{hwm=Hwm} = NewShaper} when Drop > 0 ->
+	    ?LOGFMT(warning, self(), 
+                "lager_error_logger_h dropped ~p messages in the last second that exceeded the limit of ~p messages/sec", 
+                [Drop, Hwm]),
+            eval_gl(Event, State#state{shaper=NewShaper});
+        {false, _, NewShaper} ->
+            {ok, State#state{shaper=NewShaper}}
     end.
 
 handle_info(_Info, State) ->
@@ -114,49 +103,19 @@ code_change(_OldVsn, State, _Extra) ->
 
 %% internal functions
 
-check_hwm(State = #state{hwm = undefined}) ->
-    {true, State};
-check_hwm(State = #state{mps = Mps, hwm = Hwm}) when Mps < Hwm ->
-    %% haven't hit high water mark yet, just log it
-    {true, State#state{mps=Mps+1}};
-check_hwm(State = #state{hwm = Hwm, lasttime = Last, dropped = Drop}) ->
-    %% are we still in the same second?
-    {M, S, _} = Now = os:timestamp(),
-    case Last of
-        {M, S, _} ->
-            %% still in same second, but have exceeded the high water mark
-            NewDrops = discard_messages(Now, 0),
-            {false, State#state{dropped=Drop+NewDrops}};
-        _ ->
-            %% different second, reset all counters and allow it
-            case Drop > 0 of
-                true ->
-                    ?LOGFMT(warning, self(), "lager_error_logger_h dropped ~p messages in the last second that exceeded the limit of ~p messages/sec",
-                        [Drop, Hwm]);
-                false ->
-                    ok
-            end,
-            {true, State#state{dropped = 0, mps=1, lasttime = Now}}
-    end.
-
-discard_messages(Second, Count) ->
-    {M, S, _} = os:timestamp(),
-    case Second of
-        {M, S, _} ->
-            receive
-                %% we only discard gen_event notifications, because
-                %% otherwise we might discard gen_event internal
-                %% messages, such as trapped EXITs
-                {notify, _Event} ->
-                    discard_messages(Second, Count+1);
-                {_From, _Tag, {sync_notify, _Event}} ->
-                    discard_messages(Second, Count+1)
-            after 0 ->
-                    Count
-            end;
-        _ ->
-            Count
-    end.
+eval_gl(Event, #state{groupleader_strategy=GlStrategy0}=State) when is_pid(element(2, Event)) ->
+    case element(2, Event) of
+         GL when node(GL) =/= node(), GlStrategy0 =:= ignore ->
+            gen_event:notify({error_logger, node(GL)}, Event),
+            {ok, State};
+         GL when node(GL) =/= node(), GlStrategy0 =:= mirror ->
+            gen_event:notify({error_logger, node(GL)}, Event),
+            log_event(Event, State);
+         _ ->
+            log_event(Event, State)
+    end;
+eval_gl(Event, State) ->
+    log_event(Event, State).
 
 log_event(Event, State) ->
     case Event of

src/lager.erl

@@ -30,7 +30,7 @@
         trace/2, trace/3, trace_file/2, trace_file/3, trace_file/4, trace_console/1, trace_console/2,
         name_all_sinks/0, clear_all_traces/0, stop_trace/1, stop_trace/3, status/0,
         get_loglevel/1, get_loglevel/2, set_loglevel/2, set_loglevel/3, set_loglevel/4, get_loglevels/1,
-        update_loglevel_config/1, posix_error/1,
+        update_loglevel_config/1, posix_error/1, set_loghwm/2, set_loghwm/3, set_loghwm/4,
         safe_format/3, safe_format_chop/3, dispatch_log/5, dispatch_log/6, dispatch_log/9,
         do_log/9, do_log/10, pr/2, pr/3]).
 
@@ -427,6 +427,18 @@ get_loglevels(Sink) ->
     [gen_event:call(Sink, Handler, get_loglevel, infinity) ||
         Handler <- gen_event:which_handlers(Sink)].
 
+%% @doc Set the loghwm for the default sink.
+set_loghwm(Handler, Hwm) when is_integer(Hwm) ->
+    set_loghwm(?DEFAULT_SINK, Handler, Hwm).
+
+%% @doc Set the loghwm for a particular backend.
+set_loghwm(Sink, Handler, Hwm) when is_integer(Hwm) ->
+    gen_event:call(Sink, Handler, {set_loghwm, Hwm}, infinity).
+
+%% @doc Set the loghwm (log high water mark) for file backends with multiple identifiers
+set_loghwm(Sink, Handler, Ident, Hwm) when is_integer(Hwm) ->
+    gen_event:call(Sink, {Handler, Ident}, {set_loghwm, Hwm}, infinity).
+
 %% @private
 add_trace_to_loglevel_config(Trace, Sink) ->
     {MinLevel, Traces} = lager_config:get({Sink, loglevel}),
@@ -529,3 +541,4 @@ is_record_known(Record, Module) ->
                     end
             end
     end.
+

src/lager_file_backend.erl

@@ -63,6 +63,7 @@
         size = 0 :: integer(),
         date :: undefined | string(),
         count = 10 :: integer(),
+		shaper :: lager_shaper(),
         formatter :: atom(),
         formatter_config :: any(),
         sync_on :: {'mask', integer()},
@@ -74,7 +75,8 @@
 
 -type option() :: {file, string()} | {level, lager:log_level()} |
                   {size, non_neg_integer()} | {date, string()} |
-                  {count, non_neg_integer()} | {sync_interval, non_neg_integer()} |
+                  {count, non_neg_integer()} | {high_water_mark, non_neg_integer()} |
+				  {sync_interval, non_neg_integer()} |
                   {sync_size, non_neg_integer()} | {sync_on, lager:log_level()} |
                   {check_interval, non_neg_integer()} | {formatter, atom()} |
                   {formatter_config, term()}.
@@ -102,11 +104,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, SyncInterval, SyncSize, SyncOn, CheckInterval, Formatter, FormatterConfig] =
-              [proplists:get_value(Key, Config) || Key <- [file, level, date, size, count, sync_interval, sync_size, sync_on, check_interval, formatter, formatter_config]],
+            [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]],
             Name = lager_util:expand_path(RelName),
             schedule_rotation(Name, Date),
-            State0 = #state{name=Name, level=Level, size=Size, date=Date, count=Count, formatter=Formatter,
+            Shaper = #lager_shaper{hwm=HighWaterMark},
+            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},
             State = case lager_util:open_logfile(Name, {SyncSize, SyncInterval}) of
@@ -130,15 +133,44 @@ handle_call({set_loglevel, Level}, #state{name=Ident} = State) ->
     end;
 handle_call(get_loglevel, #state{level=Level} = State) ->
     {ok, Level, State};
+handle_call({set_loghwm, Hwm}, #state{shaper=Shaper, name=Name} = State) ->
+	case validate_logfile_proplist([{file, Name}, {high_water_mark, Hwm}]) of
+        false ->
+            {ok, {error, bad_log_hwm}, State};
+        _ ->
+			NewShaper = Shaper#lager_shaper{hwm=Hwm},
+			?INT_LOG(notice, "Changed loghwm of ~s to ~p", [Name, Hwm]),
+			{ok, {last_loghwm, Shaper#lager_shaper.hwm}, State#state{shaper=NewShaper}}
+	end;
 handle_call(_Request, State) ->
     {ok, ok, State}.
 
 %% @private
 handle_event({log, Message},
-    #state{name=Name, level=L,formatter=Formatter,formatter_config=FormatConfig} = State) ->
+    #state{name=Name, level=L, shaper=Shaper, 
+		   formatter=Formatter,formatter_config=FormatConfig} = State) ->
     case lager_util:is_loggable(Message,L,{lager_file_backend, Name}) of
         true ->
-            {ok,write(State, lager_msg:timestamp(Message), lager_msg:severity_as_int(Message), Formatter:format(Message,FormatConfig)) };
+			case lager_util:check_hwm(Shaper) of
+				{true, Drop, #lager_shaper{hwm=Hwm} = NewShaper} ->
+					NewState = case Drop > 0 of
+								   true ->
+									   Report = io_lib:format("lager_file_backend dropped ~p messages in the last second that exceeded the limit of ~p messages/sec", [Drop, Hwm]),
+									   ReportMsg = lager_msg:new(Report, warning, [], []),
+									   write(State, 
+											 lager_msg:timestamp(ReportMsg), 
+											 lager_msg:severity_as_int(ReportMsg), 
+											 Formatter:format(ReportMsg, FormatConfig));
+								   false ->
+									   State
+							   end,
+					{ok,write(NewState#state{shaper=NewShaper}, 
+							  lager_msg:timestamp(Message), 
+							  lager_msg:severity_as_int(Message), 
+							  Formatter:format(Message,FormatConfig)) };
+				{false, _, NewShaper} ->
+					{ok, State#state{shaper=NewShaper}}
+			end;
         false ->
             {ok, State}
     end;
@@ -301,6 +333,13 @@ validate_logfile_proplist([{count, Count}|Tail], Acc) ->
         _ ->
             throw({bad_config, "Invalid rotation count", Count})
     end;
+validate_logfile_proplist([{high_water_mark, HighWaterMark}|Tail], Acc) ->
+    case HighWaterMark of
+        Hwm when is_integer(Hwm), Hwm >= 0 ->
+            validate_logfile_proplist(Tail, [{high_water_mark, Hwm}|Acc]);
+        _ ->
+            throw({bad_config, "Invalid high water mark", HighWaterMark})
+    end;
 validate_logfile_proplist([{date, Date}|Tail], Acc) ->
     case lager_util:parse_rotation_date_spec(Date) of
         {ok, Spec} ->
@@ -788,6 +827,10 @@ config_validation_test_() ->
             ?_assertEqual(false,
                 validate_logfile_proplist([{file, "test.log"}, {count, infinity}]))
         },
+        {"bad high water mark",
+            ?_assertEqual(false,
+                validate_logfile_proplist([{file, "test.log"}, {high_water_mark, infinity}]))
+        },
         {"bad date",
             ?_assertEqual(false,
                 validate_logfile_proplist([{file, "test.log"}, {date, "midnight"}]))

src/lager_util.erl

@@ -23,7 +23,7 @@
         open_logfile/2, ensure_logfile/4, rotate_logfile/2, format_time/0, format_time/1,
         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]).
+        trace_filter/1, trace_filter/2, expand_path/1, check_hwm/1]).
 
 -ifdef(TEST).
 -include_lib("eunit/include/eunit.hrl").
@@ -487,6 +487,43 @@ expand_path(RelPath) ->
             RelPath
     end.
 
+%% Log rate limit, i.e. high water mark for incoming messages
+
+check_hwm(Shaper = #lager_shaper{hwm = undefined}) ->
+    {true, 0, Shaper};
+check_hwm(Shaper = #lager_shaper{mps = Mps, hwm = Hwm}) when Mps < Hwm ->
+    %% haven't hit high water mark yet, just log it
+    {true, 0, Shaper#lager_shaper{mps=Mps+1}};
+check_hwm(Shaper = #lager_shaper{lasttime = Last, dropped = Drop}) ->
+    %% are we still in the same second?
+    {M, S, _} = Now = os:timestamp(),
+    case Last of
+        {M, S, _} ->
+            %% still in same second, but have exceeded the high water mark
+            NewDrops = discard_messages(Now, 0),
+            {false, 0, Shaper#lager_shaper{dropped=Drop+NewDrops}};
+        _ ->
+            %% different second, reset all counters and allow it
+            {true, Drop, Shaper#lager_shaper{dropped = 0, mps=1, lasttime = Now}}
+    end.
+
+discard_messages(Second, Count) ->
+    {M, S, _} = os:timestamp(),
+    case Second of
+        {M, S, _} ->
+            receive
+                %% we only discard gen_event notifications, because
+                %% otherwise we might discard gen_event internal
+                %% messages, such as trapped EXITs
+                {notify, _Event} ->
+                    discard_messages(Second, Count+1)
+            after 0 ->
+                    Count
+            end;
+        _ ->
+            Count
+    end.
+
 -ifdef(TEST).
 
 parse_test() ->