Write per queue upgrade log to a separate file

Author: Daniil Fedotov

scripts/rabbitmq-env

@@ -236,9 +236,11 @@ rmq_normalize_path_var RABBITMQ_PLUGINS_DIR
 [ "x" = "x$RABBITMQ_LOGS" ] && RABBITMQ_LOGS=${LOGS}
 [ "x" != "x$RABBITMQ_LOGS" ] && export RABBITMQ_LOGS_source=environment
 [ "x" = "x$RABBITMQ_LOGS" ] && RABBITMQ_LOGS="${RABBITMQ_LOG_BASE}/${RABBITMQ_NODENAME}.log"
+[ "x" = "x$RABBITMQ_UPGRADE_LOG" ] && RABBITMQ_UPGRADE_LOG="${RABBITMQ_LOG_BASE}/${RABBITMQ_NODENAME}_upgrade.log"
 
-rmq_normalize_path_var \
-    RABBITMQ_LOGS
+rmq_normalize_path_var RABBITMQ_LOGS
+
+rmq_normalize_path_var RABBITMQ_UPGRADE_LOG
 
 [ "x" = "x$RABBITMQ_CTL_ERL_ARGS" ] && RABBITMQ_CTL_ERL_ARGS=${CTL_ERL_ARGS}
 
@@ -254,7 +256,8 @@ rmq_check_if_shared_with_mnesia \
     RABBITMQ_PLUGINS_EXPAND_DIR \
     RABBITMQ_ENABLED_PLUGINS_FILE \
     RABBITMQ_PLUGINS_DIR \
-    RABBITMQ_LOGS
+    RABBITMQ_LOGS \
+    RABBITMQ_UPGRADE_LOG
 
 ##--- End of overridden <var_name> variables
 

scripts/rabbitmq-server

@@ -156,9 +156,11 @@ RABBITMQ_LISTEN_ARG=
 if [ "$RABBITMQ_LOGS" = '-' ]; then
     SASL_ERROR_LOGGER=tty
     RABBIT_LAGER_HANDLER=tty
+    RABBITMQ_LAGER_HANDLER_UPGRADE=tty
 else
     SASL_ERROR_LOGGER=false
     RABBIT_LAGER_HANDLER='"'${RABBITMQ_LOGS}'"'
+    RABBITMQ_LAGER_HANDLER_UPGRADE='"'${RABBITMQ_UPGRADE_LOG}'"'
 fi
 
 # we need to turn off path expansion because some of the vars, notably
@@ -206,6 +208,7 @@ start_rabbitmq_server() {
         -sasl sasl_error_logger "$SASL_ERROR_LOGGER" \
         -rabbit lager_log_root "\"$RABBITMQ_LOG_BASE\"" \
         -rabbit lager_handler "$RABBIT_LAGER_HANDLER" \
+        -rabbit lager_handler_upgrade "$RABBITMQ_LAGER_HANDLER_UPGRADE" \
         -rabbit enabled_plugins_file "\"$RABBITMQ_ENABLED_PLUGINS_FILE\"" \
         -rabbit plugins_dir "\"$RABBITMQ_PLUGINS_DIR\"" \
         -rabbit plugins_expand_dir "\"$RABBITMQ_PLUGINS_EXPAND_DIR\"" \

src/rabbit_lager.erl

@@ -210,7 +210,7 @@ configure_lager() ->
     %% messages to the default sink. To know the list of expected extra
     %% sinks, we look at the 'lager_extra_sinks' compilation option.
     Sinks0 = application:get_env(lager, extra_sinks, []),
-    Sinks1 = configure_extra_sinks(Sinks0, 
+    Sinks1 = configure_extra_sinks(Sinks0,
                                    [error_logger | list_expected_sinks()]),
     %% TODO Waiting for basho/lager#303
     %% Sinks2 = lists:keystore(error_logger_lager_event, 1, Sinks1,
@@ -231,18 +231,26 @@ configure_lager() ->
 configure_extra_sinks(Sinks, [SinkName | Rest]) ->
     Sink0 = proplists:get_value(SinkName, Sinks, []),
     Sink1 = case proplists:is_defined(handlers, Sink0) of
-        false -> lists:keystore(handlers, 1, Sink0,
-                                {handlers,
-                                 [{lager_forwarder_backend,
-                                   lager_util:make_internal_sink_name(lager)
-                                  }]});
+        false -> default_sink_config(SinkName, Sink0);
         true  -> Sink0
     end,
     Sinks1 = lists:keystore(SinkName, 1, Sinks, {SinkName, Sink1}),
     configure_extra_sinks(Sinks1, Rest);
 configure_extra_sinks(Sinks, []) ->
     Sinks.
 
+default_sink_config(rabbit_log_upgrade_lager_event, Sink) ->
+    io:format("I AM UPGRADE SINK"),
+    Handlers = lager_handlers(application:get_env(rabbit,
+                                                  lager_handler_upgrade,
+                                                  tty)),
+    lists:keystore(handlers, 1, Sink, {handlers, Handlers});
+default_sink_config(_, Sink) ->
+    lists:keystore(handlers, 1, Sink,
+                   {handlers,
+                    [{lager_forwarder_backend,
+                      lager_util:make_internal_sink_name(lager)}]}).
+
 list_expected_sinks() ->
     case application:get_env(rabbit, lager_extra_sinks) of
         {ok, List} ->

src/rabbit_log.erl

@@ -79,6 +79,7 @@ make_internal_sink_name(rabbit_log_channel) -> rabbit_log_channel_lager_event;
 make_internal_sink_name(rabbit_log_mirroring) -> rabbit_log_mirroring_lager_event;
 make_internal_sink_name(rabbit_log_queue) -> rabbit_log_queue_lager_event;
 make_internal_sink_name(rabbit_log_federation) -> rabbit_log_federation_lager_event;
+make_internal_sink_name(rabbit_log_upgrade) -> rabbit_log_upgrade_lager_event;
 make_internal_sink_name(Category) ->
     lager_util:make_internal_sink_name(Category).
 

src/rabbit_variable_queue.erl

@@ -2715,7 +2715,7 @@ transform_store(Store, TransformFun) ->
     rabbit_msg_store:transform_dir(rabbit_mnesia:dir(), Store, TransformFun).
 
 move_messages_to_vhost_store() ->
-    rabbit_log:info("Moving messages to per-vhost message store"),
+    log_upgrade("Moving messages to per-vhost message store"),
     Queues = list_persistent_queues(),
     %% Move the queue index for each persistent queue to the new store
     lists:foreach(
@@ -2736,7 +2736,7 @@ move_messages_to_vhost_store() ->
         {rabbit_variable_queue, migrate_queue, [OldStore, NewStoreSup]},
         QueuesWithTerms),
 
-    rabbit_log:info("Message store migration finished"),
+    log_upgrade("Message store migration finished"),
     delete_old_store(OldStore),
 
     ok = rabbit_queue_index:stop(),
@@ -2752,7 +2752,7 @@ in_batches(Size, BatchNum, MFA, List) ->
         true  -> {List, []};
         false -> lists:split(Size, List)
     end,
-    rabbit_log:info("Migrating batch ~p of ~p queues ~n", [BatchNum, Size]),
+    log_upgrade("Migrating batch ~p of ~p queues ~n", [BatchNum, Size]),
     {M, F, A} = MFA,
     Keys = [ rpc:async_call(node(), M, F, [El | A]) || El <- Batch ],
     lists:foreach(fun(Key) ->
@@ -2762,12 +2762,13 @@ in_batches(Size, BatchNum, MFA, List) ->
         end
     end,
     Keys),
-    rabbit_log:info("Batch ~p of ~p queues migrated ~n", [BatchNum, Size]),
+    log_upgrade("Batch ~p of ~p queues migrated ~n", [BatchNum, Size]),
     in_batches(Size, BatchNum + 1, MFA, Tail).
 
 migrate_queue({QueueName, RecoveryTerm}, OldStore, NewStoreSup) ->
-    rabbit_log:info("Migrating messages in queue ~s in vhost ~s to per-vhost message store~n",
-      [QueueName#resource.name, QueueName#resource.virtual_host]),
+    log_upgrade_verbose(
+        "Migrating messages in queue ~s in vhost ~s to per-vhost message store~n",
+        [QueueName#resource.name, QueueName#resource.virtual_host]),
     OldStoreClient = get_global_store_client(OldStore),
     NewStoreClient = get_per_vhost_store_client(QueueName, NewStoreSup),
     %% WARNING: During scan_queue_segments queue index state is being recovered
@@ -2790,7 +2791,7 @@ migrate_queue({QueueName, RecoveryTerm}, OldStore, NewStoreSup) ->
     NewRecoveryTerm = lists:keyreplace(persistent_ref, 1, RecoveryTerm,
                                        {persistent_ref, NewClientRef}),
     rabbit_queue_index:update_recovery_term(QueueName, NewRecoveryTerm),
-    rabbit_log:info("Queue migration finished ~p", [QueueName]),
+    log_upgrade_verbose("Queue migration finished ~p", [QueueName]),
     {QueueName, NewClientRef}.
 
 migrate_message(MsgId, OldC, NewC) ->
@@ -2859,3 +2860,14 @@ delete_old_store(OldStore) ->
     rabbit_file:recursive_delete(
         [filename:join([rabbit_mnesia:dir(), ?TRANSIENT_MSG_STORE])]).
 
+log_upgrade(Msg) ->
+    log_upgrade(Msg, []).
+
+log_upgrade(Msg, Args) ->
+    rabbit_log:info("message_store upgrades: " ++ Msg, Args).
+
+log_upgrade_verbose(Msg) ->
+    log_upgrade_verbose(Msg, []).
+
+log_upgrade_verbose(Msg, Args) ->
+    rabbit_log_upgrade:info(Msg, Args).