Add lager backend that logs to amq.rabbitmq.log

Fixes #1456

I used the `lager_console_backend` as a model for creating `lager_rabbit_backend`

Only try to declare amq.rabbitmq.log exchange once every five seconds as vhost will not be available immediately

4 files changed, 233 insertions, 140 deletions

diff --git a/src/lager_rabbit_backend.erl b/src/lager_rabbit_backend.erl
new file mode 100644
index 0000000000..87c0c6be8d
--- /dev/null
+++ b/src/lager_rabbit_backend.erl
@@ -0,0 +1,222 @@
+%% The contents of this file are subject to the Mozilla Public License
+%% Version 1.1 (the "License"); you may not use this file except in
+%% compliance with the License. You may obtain a copy of the License
+%% at http://www.mozilla.org/MPL/
+%%
+%% Software distributed under the License is distributed on an "AS IS"
+%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
+%% the License for the specific language governing rights and
+%% limitations under the License.
+%%
+%% The Original Code is RabbitMQ.
+%%
+%% The Initial Developer of the Original Code is GoPivotal, Inc.
+%% Copyright (c) 2007-2017 Pivotal Software, Inc.  All rights reserved.
+%%
+
+%% @doc RabbitMQ backend for lager.
+%% Configuration is a proplist with the following keys:
+%% <ul>
+%%    <li>`level' - log level to use</li>
+%%    <li>`formatter' - the module to use when formatting log messages. Defaults to
+%%                      `lager_default_formatter'</li>
+%%    <li>`formatter_config' - the format configuration string. Defaults to
+%%                             `time [ severity ] message'</li>
+%% </ul>
+
+-module(lager_rabbit_backend).
+
+-behaviour(gen_event).
+
+-export([init/1, terminate/2, code_change/3, handle_call/2, handle_event/2,
+         handle_info/2]).
+
+-include("rabbit.hrl").
+-include("rabbit_framing.hrl").
+
+-include_lib("lager/include/lager.hrl").
+
+-record(state, {level :: {'mask', integer()},
+                formatter :: atom(),
+                format_config :: any(),
+                init_exchange_ts = undefined :: rabbit_types:timestamp(),
+                exchange = undefined :: #resource{}}).
+
+-ifdef(TEST).
+-include_lib("eunit/include/eunit.hrl").
+-compile([{parse_transform, lager_transform}]).
+-endif.
+
+-define(INIT_EXCHANGE_INTERVAL_SECS, 5).
+-define(TERSE_FORMAT, [time, " [", severity, "] ", message]).
+-define(DEFAULT_FORMAT_CONFIG, ?TERSE_FORMAT).
+-define(FORMAT_CONFIG_OFF, []).
+
+-ifdef(TEST).
+-define(DEPRECATED(_Msg), ok).
+-else.
+-define(DEPRECATED(Msg),
+        io:format(user, "WARNING: This is a deprecated lager_rabbit_backend configuration. Please use \"~w\" instead.~n", [Msg])).
+-endif.
+
+-define(LOG_EXCH_NAME, <<"amq.rabbitmq.log">>).
+
+init([Level]) when is_atom(Level) ->
+    ?DEPRECATED([{level, Level}]),
+    init([{level, Level}]);
+init([Level, true]) when is_atom(Level) -> % for backwards compatibility
+    ?DEPRECATED([{level, Level}, {formatter_config, [{eol, "\\r\\n\\"}]}]),
+    init([{level, Level}, {formatter_config, ?FORMAT_CONFIG_OFF}]);
+init([Level, false]) when is_atom(Level) -> % for backwards compatibility
+    ?DEPRECATED([{level, Level}]),
+    init([{level, Level}]);
+
+init(Options) when is_list(Options) ->
+    true = validate_options(Options),
+    Level = get_option(level, Options, undefined),
+    try lager_util:config_to_mask(Level) of
+        L ->
+            DefaultOptions = [{formatter, lager_default_formatter},
+                              {formatter_config, ?DEFAULT_FORMAT_CONFIG}],
+            [Formatter, Config] = [get_option(K, Options, Default) || {K, Default} <- DefaultOptions],
+            State0 = #state{level=L,
+                            formatter=Formatter,
+                            format_config=Config},
+            State1 = maybe_init_exchange(State0),
+            {ok, State1}
+    catch
+        _:_ ->
+            {error, {fatal, bad_log_level}}
+    end;
+init(Level) when is_atom(Level) ->
+    ?DEPRECATED([{level, Level}]),
+    init([{level, Level}]);
+init(Other) ->
+    {error, {fatal, {bad_lager_rabbit_backend_config, Other}}}.
+
+validate_options([]) -> true;
+validate_options([{level, L}|T]) when is_atom(L) ->
+    case lists:member(L, ?LEVELS) of
+        false ->
+            throw({error, {fatal, {bad_level, L}}});
+        true ->
+            validate_options(T)
+    end;
+validate_options([{formatter, M}|T]) when is_atom(M) ->
+    validate_options(T);
+validate_options([{formatter_config, C}|T]) when is_list(C) ->
+    validate_options(T);
+validate_options([H|_]) ->
+    throw({error, {fatal, {bad_lager_rabbit_backend_config, H}}}).
+
+get_option(K, Options, Default) ->
+   case lists:keyfind(K, 1, Options) of
+       {K, V} -> V;
+       false -> Default
+   end.
+
+handle_call(get_loglevel, #state{level=Level} = State) ->
+    {ok, Level, State};
+handle_call({set_loglevel, Level}, State) ->
+    try lager_util:config_to_mask(Level) of
+        Levels ->
+            {ok, ok, State#state{level=Levels}}
+    catch
+        _:_ ->
+            {ok, {error, bad_log_level}, State}
+    end;
+handle_call(_Request, State) ->
+    {ok, ok, State}.
+
+handle_event({log, _Message} = Event, State0) ->
+    State1 = maybe_init_exchange(State0),
+    handle_log_event(Event, State1);
+handle_event(_Event, State) ->
+    {ok, State}.
+
+handle_info(_Info, State) ->
+    {ok, State}.
+
+terminate(_Reason, _State) ->
+    ok.
+
+code_change(_OldVsn, State, _Extra) ->
+    {ok, State}.
+
+%% @private
+handle_log_event({log, _Message}, #state{exchange=undefined} = State) ->
+    % NB: tried to define the exchange but still undefined,
+    % so not logging this message. Note: we can't log this dropped
+    % message because it will start an infinite loop
+    {ok, State};
+handle_log_event({log, Message},
+    #state{level=L, exchange=LogExch,
+           formatter=Formatter, format_config=FormatConfig} = State) ->
+    case lager_util:is_loggable(Message, L, ?MODULE) of
+        true ->
+            %% 0-9-1 says the timestamp is a "64 bit POSIX timestamp". That's
+            %% second resolution, not millisecond.
+            RoutingKey = rabbit_data_coercion:to_binary(lager_msg:severity(Message)),
+            Timestamp = os:system_time(seconds),
+            Node = rabbit_data_coercion:to_binary(node()),
+            Headers = [{<<"node">>, longstr, Node}],
+            AmqpMsg = #'P_basic'{content_type = <<"text/plain">>,
+                                 timestamp    = Timestamp,
+                                 headers      = Headers},
+            Body = rabbit_data_coercion:to_binary(Formatter:format(Message, FormatConfig)),
+            case rabbit_basic:publish(LogExch, RoutingKey, AmqpMsg, Body) of
+                {ok, _DeliveredQPids} -> ok;
+                {error, not_found}    -> ok
+            end,
+            {ok, State};
+        false ->
+            {ok, State}
+    end.
+
+%% @private
+maybe_init_exchange(#state{exchange=undefined, init_exchange_ts=undefined} = State) ->
+    Now = erlang:monotonic_time(second),
+    handle_init_exchange(init_exchange(true), Now, State);
+maybe_init_exchange(#state{exchange=undefined, init_exchange_ts=Timestamp} = State) ->
+    Now = erlang:monotonic_time(second),
+    Result = init_exchange(Now - Timestamp > ?INIT_EXCHANGE_INTERVAL_SECS),
+    handle_init_exchange(Result, Now, State);
+maybe_init_exchange(State) ->
+    State.
+
+%% @private
+init_exchange(true) ->
+    {ok, DefaultVHost} = application:get_env(rabbit, default_vhost),
+    VHost = rabbit_misc:r(DefaultVHost, exchange, ?LOG_EXCH_NAME),
+    try
+        #exchange{} = rabbit_exchange:declare(VHost, topic, true, false, true, [], ?INTERNAL_USER),
+        {ok, #resource{virtual_host=DefaultVHost, kind=exchange, name=?LOG_EXCH_NAME}}
+    catch
+        ErrType:Err ->
+            rabbit_log:debug("Could not initialize exchange '~s' in vhost '~s', reason: ~p:~p",
+                             [?LOG_EXCH_NAME, DefaultVHost, ErrType, Err]),
+            {ok, undefined}
+    end;
+init_exchange(_) ->
+    {ok, undefined}.
+
+%% @private
+handle_init_exchange({ok, undefined}, Now, State) ->
+    State#state{init_exchange_ts=Now};
+handle_init_exchange({ok, Exchange}, Now, State) ->
+    State#state{exchange=Exchange, init_exchange_ts=Now}.
+
+-ifdef(TEST).
+console_config_validation_test_() ->
+    Good = [{level, info}],
+    Bad1 = [{level, foo}],
+    Bad2 = [{larval, info}],
+    AllGood = [{level, info}, {formatter, my_formatter},
+               {formatter_config, ["blort", "garbage"]}],
+    [
+     ?_assertEqual(true, validate_options(Good)),
+     ?_assertThrow({error, {fatal, {bad_level, foo}}}, validate_options(Bad1)),
+     ?_assertThrow({error, {fatal, {bad_lager_rabbit_backend_config, {larval, info}}}}, validate_options(Bad2)),
+     ?_assertEqual(true, validate_options(AllGood))
+    ].
+-endif.
diff --git a/src/rabbit.erl b/src/rabbit.erl
index 0d0ff2f9fc..077657bfca 100644
--- a/src/rabbit.erl
+++ b/src/rabbit.erl
@@ -173,30 +173,19 @@
                    [{description, "message delivery logic ready"},
                     {requires,    core_initialized}]}).
 
--rabbit_boot_step({log_relay,
-                   [{description, "error log relay"},
-                    {mfa,         {rabbit_sup, start_child,
-                                   [rabbit_error_logger_lifecycle,
-                                    supervised_lifecycle,
-                                    [rabbit_error_logger_lifecycle,
-                                     {rabbit_error_logger, start, []},
-                                     {rabbit_error_logger, stop,  []}]]}},
-                    {requires,    routing_ready},
-                    {enables,     networking}]}).
-
 -rabbit_boot_step({direct_client,
                    [{description, "direct client"},
                     {mfa,         {rabbit_direct, boot, []}},
-                    {requires,    log_relay}]}).
+                    {requires,    routing_ready}]}).
 
 -rabbit_boot_step({connection_tracking,
                    [{description, "sets up internal storage for node-local connections"},
                     {mfa,         {rabbit_connection_tracking, boot, []}},
-                    {requires,    log_relay}]}).
+                    {requires,    routing_ready}]}).
 
 -rabbit_boot_step({networking,
                    [{mfa,         {rabbit_networking, boot, []}},
-                    {requires,    log_relay}]}).
+                    {requires,    routing_ready}]}).
 
 -rabbit_boot_step({notify_cluster,
                    [{description, "notify cluster nodes"},
diff --git a/src/rabbit_error_logger.erl b/src/rabbit_error_logger.erl
deleted file mode 100644
index f7b2fb82db..0000000000
--- a/src/rabbit_error_logger.erl
+++ /dev/null
@@ -1,123 +0,0 @@
-%% The contents of this file are subject to the Mozilla Public License
-%% Version 1.1 (the "License"); you may not use this file except in
-%% compliance with the License. You may obtain a copy of the License
-%% at http://www.mozilla.org/MPL/
-%%
-%% Software distributed under the License is distributed on an "AS IS"
-%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
-%% the License for the specific language governing rights and
-%% limitations under the License.
-%%
-%% The Original Code is RabbitMQ.
-%%
-%% The Initial Developer of the Original Code is GoPivotal, Inc.
-%% Copyright (c) 2007-2017 Pivotal Software, Inc.  All rights reserved.
-%%
-
--module(rabbit_error_logger).
--include("rabbit.hrl").
--include("rabbit_framing.hrl").
-
--define(LOG_EXCH_NAME, <<"amq.rabbitmq.log">>).
-
--behaviour(gen_event).
-
--export([start/0, stop/0]).
-
--export([init/1, terminate/2, code_change/3, handle_call/2, handle_event/2,
-         handle_info/2]).
-
-
-%%----------------------------------------------------------------------------
-
--spec start() -> 'ok'.
--spec stop() -> 'ok'.
-
-%%----------------------------------------------------------------------------
-
-start() ->
-    {ok, DefaultVHost} = application:get_env(default_vhost),
-    case error_logger:add_report_handler(?MODULE, [DefaultVHost]) of
-        ok ->
-            ok;
-        {error, {no_such_vhost, DefaultVHost}} ->
-            rabbit_log:warning("Default virtual host '~s' not found; "
-                               "exchange '~s' disabled~n",
-                               [DefaultVHost, ?LOG_EXCH_NAME]),
-            ok
-    end.
-
-stop() ->
-    case error_logger:delete_report_handler(rabbit_error_logger) of
-        ok                        -> ok;
-        {error, module_not_found} -> ok
-    end.
-
-%%----------------------------------------------------------------------------
-
-init([DefaultVHost]) ->
-    #exchange{} = rabbit_exchange:declare(
-                    rabbit_misc:r(DefaultVHost, exchange, ?LOG_EXCH_NAME),
-                    topic, true, false, true, [], ?INTERNAL_USER),
-    {ok, #resource{virtual_host = DefaultVHost,
-                   kind = exchange,
-                   name = ?LOG_EXCH_NAME}}.
-
-terminate(_Arg, _State) ->
-    ok.
-
-code_change(_OldVsn, State, _Extra) ->
-    {ok, State}.
-
-handle_call(_Request, State) ->
-    {ok, not_understood, State}.
-
-handle_event(Event, State) ->
-    safe_handle_event(fun handle_event0/2, Event, State).
-
-handle_event0({Kind, _Gleader, {_Pid, Format, Data}}, State) ->
-    ok = publish(Kind, Format, Data, State),
-    {ok, State};
-handle_event0(_Event, State) ->
-    {ok, State}.
-
-handle_info(_Info, State) ->
-    {ok, State}.
-
-publish(error, Format, Data, State) ->
-    publish1(<<"error">>, Format, Data, State);
-publish(warning_msg, Format, Data, State) ->
-    publish1(<<"warning">>, Format, Data, State);
-publish(info_msg, Format, Data, State) ->
-    publish1(<<"info">>, Format, Data, State);
-publish(_Other, _Format, _Data, _State) ->
-    ok.
-
-publish1(RoutingKey, Format, Data, LogExch) ->
-    %% 0-9-1 says the timestamp is a "64 bit POSIX timestamp". That's
-    %% second resolution, not millisecond.
-    Timestamp = os:system_time(seconds),
-
-    Args = [truncate:term(A, ?LOG_TRUNC) || A <- Data],
-    Headers = [{<<"node">>, longstr, list_to_binary(atom_to_list(node()))}],
-    case rabbit_basic:publish(LogExch, RoutingKey,
-                              #'P_basic'{content_type = <<"text/plain">>,
-                                         timestamp    = Timestamp,
-                                         headers      = Headers},
-                              list_to_binary(io_lib:format(Format, Args))) of
-        {ok, _DeliveredQPids} -> ok;
-        {error, not_found}    -> ok
-    end.
-
-
-safe_handle_event(HandleEvent, Event, State) ->
-    try
-        HandleEvent(Event, State)
-    catch
-        _:Error ->
-            io:format(
-              "Error in log handler~n====================~n"
-              "Event: ~P~nError: ~P~nStack trace: ~p~n~n",
-              [Event, 30, Error, 30, erlang:get_stacktrace()]),
-            {ok, State}
-    end.
diff --git a/src/rabbit_lager.erl b/src/rabbit_lager.erl
index 14aa3d31f6..13b77007f2 100644
--- a/src/rabbit_lager.erl
+++ b/src/rabbit_lager.erl
@@ -181,7 +181,7 @@ configure_lager() ->
     GeneratedHandlers = generate_lager_handlers(),
 
     %% If there are lager handlers configured,
-    %% both lager and generate RabbitMQ handlers  are used.
+    %% both lager and generate RabbitMQ handlers are used.
     %% This is because it's hard to decide clear preference rules.
     %% RabbitMQ handlers can be set to [] to use only lager handlers.
     Handlers = case application:get_env(lager, handlers, undefined) of
@@ -262,7 +262,8 @@ generate_lager_handlers(LogHandlersConfig) ->
                     Backend = lager_backend(file),
                     generate_handler(Backend, HandlerConfig)
             end;
-        ({Other, HandlerConfig}) when Other =:= console; Other =:= syslog ->
+        ({Other, HandlerConfig}) when
+              Other =:= console; Other =:= syslog; Other =:= rabbit ->
             case proplists:get_value(enabled, HandlerConfig, false) of
                 false -> [];
                 true  ->
@@ -275,7 +276,8 @@ generate_lager_handlers(LogHandlersConfig) ->
 
 lager_backend(file) -> lager_file_backend;
 lager_backend(console) -> lager_console_backend;
-lager_backend(syslog) -> lager_syslog_backend.
+lager_backend(syslog) -> lager_syslog_backend;
+lager_backend(rabbit) -> lager_rabbit_backend.
 
 generate_handler(Backend, HandlerConfig) ->
     [{Backend,
@@ -290,6 +292,9 @@ default_handler_config(lager_syslog_backend) ->
 default_handler_config(lager_console_backend) ->
     [{level, default_config_value(level)},
      {formatter_config, default_config_value(formatter_config)}];
+default_handler_config(lager_rabbit_backend) ->
+    [{level, default_config_value(level)},
+     {formatter_config, default_config_value(formatter_config)}];
 default_handler_config(lager_file_backend) ->
     [{level, default_config_value(level)},
      {formatter_config, default_config_value(formatter_config)},