Trace activity on a per-vhost basis

2 files changed, 37 insertions, 27 deletions

diff --git a/src/rabbit_control.erl b/src/rabbit_control.erl
index 941c59999a..d0fb8da4f8 100644
--- a/src/rabbit_control.erl
+++ b/src/rabbit_control.erl
@@ -124,9 +124,9 @@ Available commands:
   list_bindings  [-p <VHostPath>] 
   list_connections [<ConnectionInfoItem> ...]
 
-  set_env <VariableName> <ErlangTerm>
-  get_env <VariableName>
-  unset_env <VariableName>
+  set_env <VariableTerm> <ErlangTerm>
+  get_env <VariableTerm>
+  unset_env <VariableTerm>
 
 Quiet output mode is selected with the \"-q\" flag. Informational messages
 are suppressed when quiet mode is in effect.
@@ -267,19 +267,17 @@ action(list_connections, Node, Args, Inform) ->
                                [ArgAtoms]),
                       ArgAtoms);
 
-action(set_env, Node, [VariableName, ErlangTermStr], Inform) ->
-    Inform("Setting variable ~p for node ~p to ~s", [VariableName, Node, ErlangTermStr]),
-    {ok, Tokens, _} = erl_scan:string(ErlangTermStr ++ "."),
-    {ok, Term} = erl_parse:parse_term(Tokens),
-    rpc_call(Node, application, set_env, [rabbit, list_to_atom(VariableName), Term]);
+action(set_env, Node, [VarStr, TermStr], Inform) ->
+    Inform("Setting control variable ~s for node ~p to ~s", [VarStr, Node, TermStr]),
+    rpc_call(Node, application, set_env, [rabbit, parse_term(VarStr), parse_term(TermStr)]);
 
-action(get_env, Node, [VariableName], Inform) ->
-    Inform("Getting variable ~p for node ~p", [VariableName, Node]),
-    io:format("~p~n", [rpc_call(Node, application, get_env, [rabbit, list_to_atom(VariableName)])]);
+action(get_env, Node, [VarStr], Inform) ->
+    Inform("Getting control variable ~s for node ~p", [VarStr, Node]),
+    io:format("~p~n", [rpc_call(Node, application, get_env, [rabbit, parse_term(VarStr)])]);
 
-action(unset_env, Node, [VariableName], Inform) ->
-    Inform("Clearing variable ~p for node ~p", [VariableName, Node]),
-    rpc_call(Node, application, unset_env, [rabbit, list_to_atom(VariableName)]);
+action(unset_env, Node, [VarStr], Inform) ->
+    Inform("Clearing control variable ~s for node ~p", [VarStr, Node]),
+    rpc_call(Node, application, unset_env, [rabbit, parse_term(VarStr)]);
 
 action(Command, Node, Args, Inform) ->
     {VHost, RemainingArgs} = parse_vhost_flag(Args),
@@ -318,6 +316,11 @@ default_if_empty(List, Default) when is_list(List) ->
         [list_to_atom(X) || X <- List]
     end.
 
+parse_term(Str) ->
+    {ok, Tokens, _} = erl_scan:string(Str ++ "."),
+    {ok, Term} = erl_parse:parse_term(Tokens),
+    Term.
+
 display_info_list(Results, InfoItemKeys) when is_list(Results) ->
     lists:foreach(fun (Result) -> display_row([format_info_item(Result, X) ||
                                                   X <- InfoItemKeys])
diff --git a/src/rabbit_log.erl b/src/rabbit_log.erl
index 0cf44c8d23..b581d1ee10 100644
--- a/src/rabbit_log.erl
+++ b/src/rabbit_log.erl
@@ -100,15 +100,18 @@ error(Fmt) ->
 error(Fmt, Args) when is_list(Args) ->
     gen_server:cast(?SERVER, {error, Fmt, Args}).
 
-tap_trace_in(Message = #basic_message{exchange_name = XName},
+tap_trace_in(Message = #basic_message{exchange_name = #resource{virtual_host = VHostBin,
+                                                                name = XNameBin}},
              QPids) ->
-    check_trace(fun (TraceExchangeBin) ->
+    check_trace(VHostBin,
+                fun (TraceExchangeBin) ->
                         QInfos = [rabbit_amqqueue:info(#amqqueue{pid = P}, [name]) || P <- QPids],
                         QNames = [N || [{name, #resource{name = N}}] <- QInfos],
                         maybe_inject(TraceExchangeBin,
-                                     XName,
+                                     VHostBin,
+                                     XNameBin,
                                      <<"publish">>,
-                                     XName,
+                                     XNameBin,
                                      [{<<"queue_names">>,
                                        longstr,
                                        list_to_binary(rabbit_misc:intersperse(",", QNames))},
@@ -117,20 +120,23 @@ tap_trace_in(Message = #basic_message{exchange_name = XName},
                                        message_to_table(Message)}])
                 end).
 
-tap_trace_out(Message = #basic_message{exchange_name = XName},
-              QName) ->
-    check_trace(fun (TraceExchangeBin) ->
+tap_trace_out(Message = #basic_message{exchange_name = #resource{virtual_host = VHostBin,
+                                                                 name = XNameBin}},
+              #resource{name = QNameBin}) ->
+    check_trace(VHostBin,
+                fun (TraceExchangeBin) ->
                         maybe_inject(TraceExchangeBin,
-                                     XName,
+                                     VHostBin,
+                                     XNameBin,
                                      <<"deliver">>,
-                                     QName,
+                                     QNameBin,
                                      [{<<"message">>,
                                        table,
                                        message_to_table(Message)}])
                 end).
 
-check_trace(F) ->
-    case catch case application:get_env(rabbit, trace_exchange) of
+check_trace(VHostBin, F) ->
+    case catch case application:get_env(rabbit, {trace_exchange, VHostBin}) of
                    undefined ->
                        ok;
                    {ok, TraceExchangeBin} ->
@@ -143,9 +149,10 @@ check_trace(F) ->
     end.
 
 maybe_inject(TraceExchangeBin,
-             #resource{virtual_host = VHostBin, name = OriginalExchangeBin},
+             VHostBin,
+             OriginalExchangeBin,
              RKPrefix,
-             #resource{name = RKSuffix},
+             RKSuffix,
              Table) ->
     if
         TraceExchangeBin =:= OriginalExchangeBin ->