blkin: Messenger integration

Signed-off-by: Casey Bodley <cbodley@redhat.com>

4 files changed, 84 insertions, 2 deletions

diff --git a/src/msg/Message.cc b/src/msg/Message.cc
index bac97fc0da9..bcc4ad0425a 100644
--- a/src/msg/Message.cc
+++ b/src/msg/Message.cc
@@ -842,6 +842,43 @@ Message *decode_message(CephContext *cct, int crcflags,
 }
 
 
+WRITE_RAW_ENCODER(blkin_trace_info)
+
+void Message::encode_trace(bufferlist &bl, uint64_t features) const
+{
+#ifdef WITH_BLKIN
+  if (features & CEPH_FEATURE_BLKIN_TRACING)
+    ::encode(*trace.get_info(), bl);
+#endif
+}
+
+void Message::decode_trace(bufferlist::iterator &p, bool create)
+{
+#ifdef WITH_BLKIN
+  if (!connection)
+    return;
+
+  const auto endpoint = connection->get_messenger()->get_trace_endpoint();
+  blkin_trace_info info = {};
+
+  // only decode a trace if both sides of the connection agree
+  if (connection->has_feature(CEPH_FEATURE_BLKIN_TRACING))
+    ::decode(info, p);
+
+  if (info.trace_id) {
+    trace.init(get_type_name(), endpoint, &info, true);
+    trace.event("decoded trace");
+  } else if (create) { // create a trace even if we didn't get one on the wire
+    trace.init(get_type_name(), endpoint);
+    trace.event("created trace");
+  }
+  trace.keyval("tid", get_tid());
+  trace.keyval("entity type", get_source().type_str());
+  trace.keyval("entity num", get_source().num());
+#endif
+}
+
+
 // This routine is not used for ordinary messages, but only when encapsulating a message
 // for forwarding and routing.  It's also used in a backward compatibility test, which only
 // effectively tests backward compability for those functions.  To avoid backward compatibility
diff --git a/src/msg/Message.h b/src/msg/Message.h
index c37cd475a73..b3a836c12d4 100644
--- a/src/msg/Message.h
+++ b/src/msg/Message.h
@@ -26,6 +26,7 @@
 #include "include/types.h"
 #include "include/buffer.h"
 #include "common/Throttle.h"
+#include "common/zipkin_trace.h"
 #include "msg_types.h"
 
 #include "common/RefCountedObj.h"
@@ -240,6 +241,11 @@ protected:
   bi::list_member_hook<> dispatch_q;
 
 public:
+  // zipkin tracing
+  ZTracer::Trace trace;
+  void encode_trace(bufferlist &bl, uint64_t features) const;
+  void decode_trace(bufferlist::iterator &p, bool create = false);
+
   class CompletionHook : public Context {
   protected:
     Message *m;
@@ -297,6 +303,7 @@ protected:
     if (byte_throttler)
       byte_throttler->put(payload.length() + middle.length() + data.length());
     release_message_throttle();
+    trace.event("message destructed");
     /* call completion hooks (if any) */
     if (completion_hook)
       completion_hook->complete(0);
diff --git a/src/msg/Messenger.cc b/src/msg/Messenger.cc
index ca00e0bb5ca..6ab2862dc41 100644
--- a/src/msg/Messenger.cc
+++ b/src/msg/Messenger.cc
@@ -2,7 +2,9 @@
 // vim: ts=8 sw=2 smarttab
 
 #include <random>
+#include <netdb.h>
 #include "include/Spinlock.h"
+
 #include "include/types.h"
 #include "Messenger.h"
 
@@ -48,6 +50,27 @@ Messenger *Messenger::create(CephContext *cct, const string &type,
   return nullptr;
 }
 
+void Messenger::set_endpoint_addr(const entity_addr_t& a,
+                                  const entity_name_t &name)
+{
+  size_t hostlen;
+  if (a.get_family() == AF_INET)
+    hostlen = sizeof(struct sockaddr_in);
+  else if (a.get_family() == AF_INET6)
+    hostlen = sizeof(struct sockaddr_in6);
+  else
+    hostlen = 0;
+
+  if (hostlen) {
+    char buf[NI_MAXHOST] = { 0 };
+    getnameinfo(a.get_sockaddr(), hostlen, buf, sizeof(buf),
+                NULL, 0, NI_NUMERICHOST);
+
+    trace_endpoint.copy_ip(buf);
+  }
+  trace_endpoint.set_port(a.get_port());
+}
+
 /*
  * Pre-calculate desired software CRC settings.  CRC computation may
  * be disabled by default for some transports (e.g., those with strong
diff --git a/src/msg/Messenger.h b/src/msg/Messenger.h
index c4df379939a..a186ec3c875 100644
--- a/src/msg/Messenger.h
+++ b/src/msg/Messenger.h
@@ -41,6 +41,10 @@ class Messenger {
 private:
   list<Dispatcher*> dispatchers;
   list <Dispatcher*> fast_dispatchers;
+  ZTracer::Endpoint trace_endpoint;
+
+  void set_endpoint_addr(const entity_addr_t& a,
+                         const entity_name_t &name);
 
 protected:
   /// the "name" of the local daemon. eg client.99
@@ -136,7 +140,8 @@ public:
    * or use the create() function.
    */
   Messenger(CephContext *cct_, entity_name_t w)
-    : my_inst(),
+    : trace_endpoint("0.0.0.0", 0, "Messenger"),
+      my_inst(),
       default_send_priority(CEPH_MSG_PRIO_DEFAULT), started(false),
       magic(0),
       socket_priority(-1),
@@ -213,9 +218,19 @@ protected:
   /**
    * set messenger's address
    */
-  virtual void set_myaddr(const entity_addr_t& a) { my_inst.addr = a; }
+  virtual void set_myaddr(const entity_addr_t& a) {
+    my_inst.addr = a;
+    set_endpoint_addr(a, my_inst.name);
+  }
 public:
   /**
+   * @return the zipkin trace endpoint
+   */
+  const ZTracer::Endpoint* get_trace_endpoint() const {
+    return &trace_endpoint;
+  }
+
+  /**
    * Retrieve the Messenger's name.
    *
    * @return A const reference to the name this Messenger