blkin: Messenger integration

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

diff --git a/src/common/common_init.cc b/src/common/common_init.cc
index 161af4e5a40276fde235d3da8148e2b13116aa70..d796e933c4b95e92e041580e3521410219b21cf8 100644 (file)
--- a/src/common/common_init.cc
+++ b/src/common/common_init.cc
@@ -25,6 +25,7 @@
 #include "common/valgrind.h"
 #include "common/version.h"
 #include "common/strtol.h"
+#include "common/zipkin_trace.h"
 #include "include/color.h"
 
 #include <errno.h>

diff --git a/src/include/ceph_features.h b/src/include/ceph_features.h
index 68a55d05d3acf40fec0174693bcb26fa43a4bbca..c5578161a93c9ee55d238370c6df95aa0144f8fb 100755 (executable)
--- a/src/include/ceph_features.h
+++ b/src/include/ceph_features.h
@@ -161,12 +161,22 @@ DEFINE_CEPH_FEATURE(58, 1, FS_FILE_LAYOUT_V2) // overlap
 DEFINE_CEPH_FEATURE(59, 1, FS_BTIME)
 DEFINE_CEPH_FEATURE(59, 1, FS_CHANGE_ATTR) // overlap
 DEFINE_CEPH_FEATURE(59, 1, MSG_ADDR2) // overlap
+DEFINE_CEPH_FEATURE(60, 1, BLKIN_TRACING)  // *do not share this bit*
 
 DEFINE_CEPH_FEATURE(61, 1, RESERVED2)          // unused, but slow down!
 DEFINE_CEPH_FEATURE(62, 1, RESERVED)           // do not use; used as a sentinal
 DEFINE_CEPH_FEATURE_DEPRECATED(63, 1, RESERVED_BROKEN, LUMINOUS) // client-facing
 
 
+/*
+ * conditionally include blkin in CEPH_FEATURES_ALL/SUPPORTED_DEFAULT
+ */
+#ifdef WITH_BLKIN
+#define CEPH_FEATURES_BLKIN CEPH_FEATURE_BLKIN_TRACING
+#else
+#define CEPH_FEATURES_BLKIN 0
+#endif
+
 /*
  * Features supported.  Should be everything above.
  */
@@ -227,6 +237,7 @@ DEFINE_CEPH_FEATURE_DEPRECATED(63, 1, RESERVED_BROKEN, LUMINOUS) // client-facin
         CEPH_FEATURE_SERVER_LUMINOUS |         \
         CEPH_FEATURE_RESEND_ON_SPLIT |         \
         CEPH_FEATURE_RADOS_BACKOFF |           \
+        CEPH_FEATURES_BLKIN | \
         0ULL)
 
 #define CEPH_FEATURES_SUPPORTED_DEFAULT  CEPH_FEATURES_ALL

diff --git a/src/msg/Message.cc b/src/msg/Message.cc
index bac97fc0da91aa5d86b87a996198aa80ed65eea1..bcc4ad0425a49a50baa17fd2b3cc1c0a48f77925 100644 (file)
--- 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 c37cd475a737c56b9d195ed0d2d3e763e4ee2fb2..b3a836c12d416d59bac916f794b57b5144ed43f3 100644 (file)
--- 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 ca00e0bb5caca291bf2e75a025f1de82cc2a8f40..6ab2862dc412a1948977d38da92142df97f38458 100644 (file)
--- 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 c4df379939afa078f74888561b846ec90c977633..a186ec3c875a3a4ee30630fad2261b6896621e11 100644 (file)
--- 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,8 +218,18 @@ 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.
    *