Include username into trace message headers

2 files changed, 19 insertions, 13 deletions

diff --git a/src/rabbit_channel.erl b/src/rabbit_channel.erl
index 159442707a..1b4ba7f1c7 100644
--- a/src/rabbit_channel.erl
+++ b/src/rabbit_channel.erl
@@ -669,7 +669,8 @@ handle_method(#'basic.publish'{exchange    = ExchangeNameBin,
               Content, State = #ch{virtual_host    = VHostPath,
                                    tx              = Tx,
                                    confirm_enabled = ConfirmEnabled,
-                                   trace_state     = TraceState}) ->
+                                   trace_state     = TraceState,
+                                   user = #user{username = Username}}) ->
     check_msg_size(Content),
     ExchangeName = rabbit_misc:r(VHostPath, exchange, ExchangeNameBin),
     check_write_permitted(ExchangeName, State),
@@ -690,7 +691,7 @@ handle_method(#'basic.publish'{exchange    = ExchangeNameBin,
         end,
     case rabbit_basic:message(ExchangeName, RoutingKey, DecodedContent) of
         {ok, Message} ->
-            rabbit_trace:tap_in(Message, TraceState),
+            rabbit_trace:tap_in(Message, Username, TraceState),
             Delivery = rabbit_basic:delivery(
                          Mandatory, DoConfirm, Message, MsgSeqNo),
             QNames = rabbit_exchange:route(Exchange, Delivery),
@@ -1365,7 +1366,8 @@ record_sent(ConsumerTag, AckRequired,
             Msg = {QName, QPid, MsgId, Redelivered, _Message},
             State = #ch{unacked_message_q = UAMQ,
                         next_tag          = DeliveryTag,
-                        trace_state       = TraceState}) ->
+                        trace_state       = TraceState,
+                        user              = #user{username = Username}}) ->
     ?INCR_STATS([{queue_stats, QName, 1}], case {ConsumerTag, AckRequired} of
                                                {none,  true} -> get;
                                                {none, false} -> get_no_ack;
@@ -1376,7 +1378,7 @@ record_sent(ConsumerTag, AckRequired,
         true  -> ?INCR_STATS([{queue_stats, QName, 1}], redeliver, State);
         false -> ok
     end,
-    rabbit_trace:tap_out(Msg, TraceState),
+    rabbit_trace:tap_out(Msg, 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 017d1ab242..afb149d015 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/2, tap_out/2, start/1, stop/1]).
+-export([init/1, enabled/1, tap_in/3, tap_out/3, start/1, stop/1]).
 
 -include("rabbit.hrl").
 -include("rabbit_framing.hrl").
@@ -32,8 +32,8 @@
 
 -spec(init/1 :: (rabbit_types:vhost()) -> state()).
 -spec(enabled/1 :: (rabbit_types:vhost()) -> boolean()).
--spec(tap_in/2 :: (rabbit_types:basic_message(), state()) -> 'ok').
--spec(tap_out/2 :: (rabbit_amqqueue:qmsg(), state()) -> 'ok').
+-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(start/1 :: (rabbit_types:vhost()) -> 'ok').
 -spec(stop/1 :: (rabbit_types:vhost()) -> 'ok').
@@ -54,18 +54,22 @@ enabled(VHost) ->
     {ok, VHosts} = application:get_env(rabbit, ?TRACE_VHOSTS),
     lists:member(VHost, VHosts).
 
-tap_in(_Msg, none) -> ok;
+tap_in(_Msg, _Username, none) -> ok;
 tap_in(Msg = #basic_message{exchange_name = #resource{name = XName,
-                                                      virtual_host = VHost}}, TraceX) ->
+                                                      virtual_host = VHost}},
+       Username, TraceX) ->
     trace(TraceX, Msg, <<"publish">>, XName,
-          [{<<"vhost">>, longstr, VHost}]).
+          [{<<"vhost">>, longstr, VHost},
+           {<<"user">>, longstr, Username}]).
 
-tap_out(_Msg, none) -> ok;
-tap_out({#resource{name = QName, virtual_host = VHost}, _QPid, _QMsgId, Redelivered, Msg}, TraceX) ->
+tap_out(_Msg, _Username, none) -> ok;
+tap_out({#resource{name = QName, virtual_host = VHost}, _QPid, _QMsgId, Redelivered, Msg},
+        Username, TraceX) ->
     RedeliveredNum = case Redelivered of true -> 1; false -> 0 end,
     trace(TraceX, Msg, <<"deliver">>, QName,
           [{<<"redelivered">>, signedint, RedeliveredNum},
-           {<<"vhost">>, longstr, VHost}]).
+           {<<"vhost">>, longstr, VHost},
+           {<<"user">>, longstr, Username}]).
 
 %%----------------------------------------------------------------------------