amqp: log time elapsed between receiving a message and replying

Change-Id: I1e64587cb76fe449fd0d60f78e830dc603574af5

1 files changed, 7 insertions, 2 deletions

diff --git a/oslo_messaging/_drivers/amqpdriver.py b/oslo_messaging/_drivers/amqpdriver.py
index 4ec37e8..d899713 100644
--- a/oslo_messaging/_drivers/amqpdriver.py
+++ b/oslo_messaging/_drivers/amqpdriver.py
@@ -21,6 +21,7 @@ import time
 import uuid
 
 import cachetools
+from oslo_utils import timeutils
 from six import moves
 
 import oslo_messaging
@@ -46,6 +47,8 @@ class AMQPIncomingMessage(base.RpcIncomingMessage):
         self.msg_id = msg_id
         self.reply_q = reply_q
         self._obsolete_reply_queues = obsolete_reply_queues
+        self.stopwatch = timeutils.StopWatch()
+        self.stopwatch.start()
 
     def _send_reply(self, conn, reply=None, failure=None, log_failure=True):
         if not self._obsolete_reply_queues.reply_q_valid(self.reply_q,
@@ -63,10 +66,12 @@ class AMQPIncomingMessage(base.RpcIncomingMessage):
         unique_id = msg[rpc_amqp.UNIQUE_ID]
 
         LOG.debug("sending reply msg_id: %(msg_id)s "
-                  "reply queue: %(reply_q)s", {
+                  "reply queue: %(reply_q)s "
+                  "time elapsed: %(elapsed)ss", {
                       'msg_id': self.msg_id,
                       'unique_id': unique_id,
-                      'reply_q': self.reply_q})
+                      'reply_q': self.reply_q,
+                      'elapsed': self.stopwatch.elapsed()})
         conn.direct_send(self.reply_q, rpc_common.serialize_msg(msg))
 
     def reply(self, reply=None, failure=None, log_failure=True):