Include connection name into trace message headers

2 files changed, 17 insertions, 11 deletions

diff --git a/src/rabbit_channel.erl b/src/rabbit_channel.erl
index 1b4ba7f1c7..ac989d6246 100644
--- a/src/rabbit_channel.erl
+++ b/src/rabbit_channel.erl
@@ -670,7 +670,8 @@ handle_method(#'basic.publish'{exchange    = ExchangeNameBin,
                                    tx              = Tx,
                                    confirm_enabled = ConfirmEnabled,
                                    trace_state     = TraceState,
-                                   user = #user{username = Username}}) ->
+                                   user = #user{username = Username},
+                                   conn_name       = ConnName}) ->
     check_msg_size(Content),
     ExchangeName = rabbit_misc:r(VHostPath, exchange, ExchangeNameBin),
     check_write_permitted(ExchangeName, State),
@@ -691,7 +692,7 @@ handle_method(#'basic.publish'{exchange    = ExchangeNameBin,
         end,
     case rabbit_basic:message(ExchangeName, RoutingKey, DecodedContent) of
         {ok, Message} ->
-            rabbit_trace:tap_in(Message, Username, TraceState),
+            rabbit_trace:tap_in(Message, ConnName, Username, TraceState),
             Delivery = rabbit_basic:delivery(
                          Mandatory, DoConfirm, Message, MsgSeqNo),
             QNames = rabbit_exchange:route(Exchange, Delivery),
@@ -1367,7 +1368,8 @@ record_sent(ConsumerTag, AckRequired,
             State = #ch{unacked_message_q = UAMQ,
                         next_tag          = DeliveryTag,
                         trace_state       = TraceState,
-                        user              = #user{username = Username}}) ->
+                        user              = #user{username = Username},
+                        conn_name         = ConnName}) ->
     ?INCR_STATS([{queue_stats, QName, 1}], case {ConsumerTag, AckRequired} of
                                                {none,  true} -> get;
                                                {none, false} -> get_no_ack;
@@ -1378,7 +1380,7 @@ record_sent(ConsumerTag, AckRequired,
         true  -> ?INCR_STATS([{queue_stats, QName, 1}], redeliver, State);
         false -> ok
     end,
-    rabbit_trace:tap_out(Msg, Username, TraceState),
+    rabbit_trace:tap_out(Msg, ConnName, Username, TraceState),
     UAMQ1 = case AckRequired of
                 true  -> queue:in({DeliveryTag, ConsumerTag, {QPid, MsgId}},
                                   UAMQ);
diff --git a/src/rabbit_trace.erl b/src/rabbit_trace.erl
index afb149d015..8959eefb94 100644
--- a/src/rabbit_trace.erl
+++ b/src/rabbit_trace.erl
@@ -16,7 +16,7 @@
 
 -module(rabbit_trace).
 
--export([init/1, enabled/1, tap_in/3, tap_out/3, start/1, stop/1]).
+-export([init/1, enabled/1, tap_in/4, tap_out/4, start/1, stop/1]).
 
 -include("rabbit.hrl").
 -include("rabbit_framing.hrl").
@@ -32,8 +32,10 @@
 
 -spec(init/1 :: (rabbit_types:vhost()) -> state()).
 -spec(enabled/1 :: (rabbit_types:vhost()) -> boolean()).
--spec(tap_in/3 :: (rabbit_types:basic_message(), rabbit_types:username(), state()) -> 'ok').
--spec(tap_out/3 :: (rabbit_amqqueue:qmsg(), rabbit_types:username(), state()) -> 'ok').
+-spec(tap_in/4 :: (rabbit_types:basic_message(), string(),
+                   rabbit_types:username(), state()) -> 'ok').
+-spec(tap_out/4 :: (rabbit_amqqueue:qmsg(), string(),
+                    rabbit_types:username(), state()) -> 'ok').
 
 -spec(start/1 :: (rabbit_types:vhost()) -> 'ok').
 -spec(stop/1 :: (rabbit_types:vhost()) -> 'ok').
@@ -54,21 +56,23 @@ enabled(VHost) ->
     {ok, VHosts} = application:get_env(rabbit, ?TRACE_VHOSTS),
     lists:member(VHost, VHosts).
 
-tap_in(_Msg, _Username, none) -> ok;
+tap_in(_Msg, _ConnName, _Username, none) -> ok;
 tap_in(Msg = #basic_message{exchange_name = #resource{name = XName,
                                                       virtual_host = VHost}},
-       Username, TraceX) ->
+       ConnName, Username, TraceX) ->
     trace(TraceX, Msg, <<"publish">>, XName,
           [{<<"vhost">>, longstr, VHost},
+           {<<"connection">>, longstr, ConnName},
            {<<"user">>, longstr, Username}]).
 
-tap_out(_Msg, _Username, none) -> ok;
+tap_out(_Msg, _ConnName, _Username, none) -> ok;
 tap_out({#resource{name = QName, virtual_host = VHost}, _QPid, _QMsgId, Redelivered, Msg},
-        Username, TraceX) ->
+        ConnName, Username, TraceX) ->
     RedeliveredNum = case Redelivered of true -> 1; false -> 0 end,
     trace(TraceX, Msg, <<"deliver">>, QName,
           [{<<"redelivered">>, signedint, RedeliveredNum},
            {<<"vhost">>, longstr, VHost},
+           {<<"connection">>, longstr, ConnName},
            {<<"user">>, longstr, Username}]).
 
 %%----------------------------------------------------------------------------