Allow the console backend to redirect output to a remote node

To support tracing to a console that is not the erlang node's *direct*
console introduce 2 additional options to the console backend:
`group_leader` and `id`. Group leader allows to use io:put_chars to send
the console output to a remote pid (eg. the group leader on a node
created by nodetool) and the ID allows multiple console backends to
exist simultaneously (for example {lager_console_backend, NodeName}).

Additionally, to support ephemeral console backends with associated
traces, if the console handler explicitly removes itself from the event
handler, remove any traces for that id.

To facilitate this, a new function has been added to the lager module:
`lager:clear_trace_by_destination/1`.

src/lager.erl

@@ -32,7 +32,7 @@
         rotate_handler/1, rotate_handler/2, rotate_sink/1, rotate_all/0,
         trace/2, trace/3, trace_file/2, trace_file/3, trace_file/4, trace_console/1, trace_console/2,
         install_trace/2, remove_trace/1, trace_func/3,
-        list_all_sinks/0, clear_all_traces/0, stop_trace/1, stop_trace/3, status/0,
+        list_all_sinks/0, clear_all_traces/0, clear_trace_by_destination/1, 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, set_loghwm/2, set_loghwm/3, set_loghwm/4,
         safe_format/3, safe_format_chop/3, unsafe_format/2, dispatch_log/5, dispatch_log/7, dispatch_log/9,
@@ -336,6 +336,12 @@ clear_traces_by_sink(Sinks) ->
                   end,
                   Sinks).
 
+clear_trace_by_destination(ID) ->
+    Sinks = lists:sort(list_all_sinks()),
+    Traces = find_traces(Sinks),
+    [ stop_trace_int({Filter, Level, Destination}, Sink) || {Sink, {Filter, Level, Destination}} <- Traces, Destination == ID].
+
+
 clear_all_traces() ->
     Handlers = lager_config:global_get(handlers, []),
     clear_traces_by_sink(list_all_sinks()),

src/lager_console_backend.erl

@@ -34,7 +34,8 @@
         code_change/3]).
 
 -record(state, {level :: {'mask', integer()},
-                out = user :: user | standard_error,
+                out = user :: user | standard_error | pid(),
+                id :: atom() | {atom(), any()},
                 formatter :: atom(),
                 format_config :: any(),
                 colors=[] :: list()}).
@@ -93,17 +94,27 @@ init(Options) when is_list(Options) ->
             ?INT_LOG(warning, Msg, []),
             {error, {fatal, old_shell}};
         {true, L} ->
-            [UseErr, Formatter, Config] = [ get_option(K, Options, Default) || {K, Default} <- [
+            [UseErr, GroupLeader, ID, Formatter, Config] = [ get_option(K, Options, Default) || {K, Default} <- [
                                                                                    {use_stderr, false},
+                                                                                   {group_leader, false},
+                                                                                   {id, ?MODULE},
                                                                                    {formatter, lager_default_formatter},
                                                                                    {formatter_config, ?DEFAULT_FORMAT_CONFIG}
                                                                                                ]
                                           ],
             Out = case UseErr of
-                     false -> user;
+                     false ->
+                          case GroupLeader of
+                              false -> user;
+                              GLPid when is_pid(GLPid) ->
+                                  ?INT_LOG(notice, "logging to remote process ~p on ~p", [GLPid, node(GLPid)]),
+                                  erlang:monitor_node(node(GLPid), true),
+                                  GLPid
+                          end;
                      true -> standard_error
                   end,
             {ok, #state{level=L,
+                    id=ID,
                     out=Out,
                     formatter=Formatter,
                     format_config=Config,
@@ -134,6 +145,10 @@ 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([{group_leader, L}|T]) when is_pid(L) ->
+    validate_options(T);
+validate_options([{id, {?MODULE, _}}|T]) ->
+    validate_options(T);
 validate_options([H|_]) ->
     throw({error, {fatal, {bad_console_config, H}}}).
 
@@ -159,8 +174,8 @@ handle_call(_Request, State) ->
 
 %% @private
 handle_event({log, Message},
-    #state{level=L,out=Out,formatter=Formatter,format_config=FormatConfig,colors=Colors} = State) ->
-    case lager_util:is_loggable(Message, L, ?MODULE) of
+    #state{level=L,out=Out,formatter=Formatter,format_config=FormatConfig,colors=Colors,id=ID} = State) ->
+    case lager_util:is_loggable(Message, L, ID) of
         true ->
             io:put_chars(Out, Formatter:format(Message,FormatConfig,Colors)),
             {ok, State};
@@ -171,10 +186,23 @@ handle_event(_Event, State) ->
     {ok, State}.
 
 %% @private
+handle_info({nodedown, Node}, State=#state{out=Out}) when is_pid(Out) ->
+    ?INT_LOG(notice, "Remote node ~p is down ~p", [Node, node(Out)]),
+    case node(Out) of
+        Node ->
+            ?INT_LOG(notice, "Removing handler", []),
+            remove_handler;
+        _ ->
+            {ok, State}
+    end;
 handle_info(_Info, State) ->
     {ok, State}.
 
 %% @private
+terminate(remove_handler, _State=#state{id=ID}) ->
+    %% have to do this asynchronously because we're in the event handlr
+    spawn(fun() -> lager:clear_trace_by_destination(ID) end),
+    ok;
 terminate(_Reason, _State) ->
     ok.