BlkKin (LTTng + Zipkin) tracing patchset intro

index 36312cf..9ca7bc3 100644
@@ -285,6 +285,17 @@ if test "x$enable_coverage" != xno; then
 fi
 AC_SUBST(GCOV_PREFIX_STRIP, `echo $(pwd)/src | tr -dc / | wc -c`)
 
+# blkin (lttng+zipkin) tracing?
+AC_ARG_WITH([blkin],
+	    [AS_HELP_STRING([--with-blkin], [blkin (lttng + zipkin) tracing])],
+	    [],
+	    [with_blkin=no])
+have_blkin=no
+AS_IF([test "x$with_blkin" == "xyes"],
+    [PKG_CHECK_MODULES([BLKIN], [blkin], [have_blkin=yes])])
+AM_CONDITIONAL(WITH_BLKIN, test "x$have_blkin" == xyes)
+AM_COND_IF([WITH_BLKIN], [AC_DEFINE([WITH_BLKIN], [1], [Defined if using BlkKin])])
+
 # radosgw?
 AC_ARG_WITH([radosgw],
             [AS_HELP_STRING([--with-radosgw], [build RADOS gateway])],

index ea498d8..13256ff 100755
@@ -5,6 +5,7 @@ usage() {
 do_autogen.sh: make a ceph build by running autogen, etc.
 
 -h:                              this help message
+-b                               blkin tracing
 -d <level>                       debug build
                                  level 0: no debug
                                  level 1: -g

@@ -30,9 +31,11 @@ debug_level=0
 verbose=0
 profile=0
 CONFIGURE_FLAGS=""
-while getopts  "d:e:hHTPjpnvO:" flag
+while getopts  "bd:e:hHTPjpnvO:" flag
 do
     case $flag in
+    b) CONFIGURE_FLAGS="$CONFIGURE_FLAGS --with-blkin";;
+
     d) debug_level=$OPTARG;;
 
     O) CFLAGS="${CFLAGS} -O$OPTARG";;

88 lines not on this page · raw message

index e1ba4a8..0e5dbe7 100644
@@ -167,6 +167,12 @@ if ENABLE_COVERAGE
 EXTRALIBS += -lgcov
 endif # ENABLE_COVERAGE
 
+if WITH_BLKIN
+AM_CPPFLAGS += -DWITH_BLKIN
+AM_CXXFLAGS += $(BLKIN_CFLAGS)
+EXTRALIBS += $(BLKIN_LIBS)
+endif
+
 # Libosd always needs osdc and os
 LIBOSD += $(LIBOSDC) $(LIBOS)
 

index edec05e..243a616 100644
@@ -38,10 +38,20 @@ include tools/Makefile.am
 
 ceph_mon_SOURCES = ceph_mon.cc
 ceph_mon_LDADD = $(LIBMON) $(LIBOS) $(CEPH_GLOBAL) $(LIBCOMMON)
+if WITH_BLKIN
+ceph_mon_CPPFLAGS = -DWITH_BLKIN $(AM_CPPFLAGS)
+ceph_mon_CXXFLAGS = $(BLKIN_CFLAGS) $(AM_CXXFLAGS)
+ceph_mon_LDADD += $(BLKIN_LIBS)
+endif
 bin_PROGRAMS += ceph-mon
 
 ceph_osd_SOURCES = ceph_osd.cc
 ceph_osd_LDADD = $(LIBOSD) $(CEPH_GLOBAL) $(LIBCOMMON)
+if WITH_BLKIN
+ceph_osd_CPPFLAGS = -DWITH_BLKIN $(AM_CPPFLAGS)
+ceph_osd_CXXFLAGS = $(BLKIN_CFLAGS) $(AM_CXXFLAGS)
+ceph_osd_LDADD += $(BLKIN_LIBS)
+endif
 bin_PROGRAMS += ceph-osd
 
 ceph_mds_SOURCES = ceph_mds.cc

index 23e9167..a2a59c0 100644
@@ -9,6 +9,13 @@ librados_la_CXXFLAGS = ${AM_CXXFLAGS}
 
 LIBRADOS_DEPS += libcls_lock_client.la $(LIBOSDC) $(LIBCOMMON)
 librados_la_LIBADD = $(LIBRADOS_DEPS) $(PTHREAD_LIBS) $(CRYPTO_LIBS) $(EXTRALIBS)
+
+if WITH_BLKIN
+librados_la_CPPFLAGS = -DWITH_BLKIN $(AM_CPPFLAGS)
+librados_la_CXXFLAGS += $(BLKIN_CXXFLAGS)
+librados_la_LIBADD += $(BLKIN_LIBS)
+endif
+
 librados_la_LDFLAGS = ${AM_LDFLAGS} -version-info 2:0:0
 if LINUX
 librados_la_LDFLAGS += -export-symbols-regex '^rados_.*'

index 68c6503..e08cbde 100644
@@ -15,6 +15,12 @@ libmon_la_SOURCES = \
 	mon/DataHealthService.cc \
 	mon/ConfigKeyService.cc
 libmon_la_LIBADD = $(LIBAUTH) $(LIBCOMMON) $(LIBOS)
+
+if WITH_BLKIN
+libmon_la_CXXFLAGS = $(BLKIN_CFLAGS) $(AM_CXXFLAGS)
+libmon_la_LIBADD += $(BLKIN_LIBS)
+endif
+
 noinst_LTLIBRARIES += libmon.la
 
 noinst_HEADERS += \

@@ -43,4 +49,3 @@ noinst_HEADERS += \
 	mon/QuorumService.h \
 	mon/Session.h \
 	mon/mon_types.h
-

index a849a1c..c1c9514 100644
@@ -7,6 +7,12 @@ libmsg_la_SOURCES = \
 	msg/SimpleMessenger.cc \
 	msg/msg_types.cc
 
+if WITH_BLKIN
+libmsg_la_CPPFLAGS = -DWITH_BLKIN $(AM_CPPFLAGS)
+libmsg_la_CXXFLAGS = $(BLKIN_CFLAGS) $(AM_CXXFLAGS)
+libmsg_la_LIBADD = $(BLKIN_LIBS)
+endif
+
 noinst_HEADERS += \
 	msg/Accepter.h \
 	msg/DispatchQueue.h \

index 252c678..f69adae 100644
@@ -29,6 +29,12 @@ if WITH_LIBZFS
 libos_la_SOURCES += os/ZFSFileStoreBackend.cc
 endif
 
+if WITH_BLKIN
+libos_la_CPPFLAGS = -DWITH_BLKIN $(AM_CPPFLAGS)
+libos_la_CXXFLAGS = $(BLKIN_CFLAGS) $(AM_CXXFLAGS)
+libos_la_LIBADD = $(BLKIN_LIBS)
+endif
+
 noinst_LTLIBRARIES += libos.la
 
 noinst_HEADERS += \

index 8a2cb2b..b8d9ef1 100644
@@ -20,6 +20,13 @@ libosd_la_SOURCES = \
 	osd/ECUtil.cc \
 	objclass/class_api.cc
 libosd_la_LIBADD = $(LIBOSDC) $(LIBOS)
+
+if WITH_BLKIN
+libosd_la_CPPFLAGS = -DWITH_BLKIN $(AM_CPPFLAGS)
+libosd_la_CXXFLAGS = $(BLKIN_CFLAGS) $(AM_CXXFLAGS)
+libosd_la_LIBADD += $(BLKIN_LIBS)
+endif
+
 noinst_LTLIBRARIES += libosd.la
 
 noinst_HEADERS += \

index 3a8a216..aac506f 100644
@@ -4,6 +4,13 @@ libosdc_la_SOURCES = \
 	osdc/Filer.cc \
 	osdc/Striper.cc \
 	osdc/Journaler.cc
+
+if WITH_BLKIN
+libosdc_la_CPPFLAGS = -DWITH_BLKIN $(AM_CPPFLAGS)
+libosdc_la_CXXFLAGS = $(BLKIN_CFLAGS) $(AM_CXXFLAGS)
+libosdc_la_LIBADD = $(BLKIN_LIBS)
+endif
+
 noinst_LTLIBRARIES += libosdc.la
 
 noinst_HEADERS += \

index ddb2f91..63326af 100644
@@ -240,6 +240,19 @@ void OpTracker::_mark_event(TrackedOp *op, const string &evt,
 	  << ", request: " << *op->request << dendl;
 }
 
+#ifdef WITH_BLKIN
+void OpTracker::trace_event(TrackedOp *op, TrackedOpTraceRef t, const string &evt, TrackedOpEndpointRef ep)
+{
+  t->event(evt, ep);
+}
+
+void OpTracker::trace_keyval(TrackedOp *op, TrackedOpTraceRef t, const string &key,
+			     const string &val, TrackedOpEndpointRef ep)
+{
+  t->keyval(key, val, ep);
+}
+#endif
+
 void OpTracker::RemoveOnDelete::operator()(TrackedOp *op) {
   if (!tracker->tracking_enabled) {
     op->request->clear_data();

@@ -247,6 +260,7 @@ void OpTracker::RemoveOnDelete::operator()(TrackedOp *op) {
     return;
   }
   op->mark_event("done");
+  BLKIN_OP_TRACE_EVENT(op, osd, span_ended);
   tracker->unregister_inflight_op(op);
   // Do not delete op, unregister_inflight_op took control
 }

@@ -277,3 +291,116 @@ void TrackedOp::dump(utime_t now, Formatter *f) const
     f->close_section();
   }
 }
+
+#ifdef WITH_BLKIN
+bool TrackedOp::create_osd_trace(TrackedOpEndpointRef ep)
+{
+  string name = "OSD Handling op";
+  if (!request) {
+    return false;
+  }
+
+  TrackedOpTraceRef mt = request->get_master_trace();
+  if (!mt) {
+    return false;
+  }
+
+  osd_trace = ZTracer::create_ZTrace(name, mt, ep);
+  return true;
+}
+
+void TrackedOp::trace_osd(string event)
+{
+  if (!osd_trace) {
+    return;
+  }
+
+  tracker->trace_event(this, osd_trace, event, osd_trace->get_endpoint());
+}
+
+void TrackedOp::trace_osd(string key, string val)
+{
+  if (!osd_trace) {
+    return;
+  }
+
+  tracker->trace_keyval(this, osd_trace, key, val, osd_trace->get_endpoint());
+}
+
+
+bool TrackedOp::create_pg_trace(TrackedOpEndpointRef ep)
+{
+  string name = "PG";
+  if (!request) {
+    return false;
+  }
+
+  TrackedOpTraceRef mt = request->get_master_trace();
+  if (!mt) {
+    return false;
+  }
+
+  pg_trace = ZTracer::create_ZTrace(name, mt, ep);
+  return true;
+}
+
+void TrackedOp::trace_pg(string event)
+{
+  if (!pg_trace) {
+    return;
+  }
+
+  tracker->trace_event(this, osd_trace, event, pg_trace->get_endpoint());
+}
+
+void TrackedOp::get_pg_trace_info(struct blkin_trace_info *info)
+{
+  if (!pg_trace) {
+    return;
+  }
+
+  osd_trace->get_trace_info(info);
+}
+
+bool TrackedOp::create_journal_trace(TrackedOpEndpointRef ep)
+{
+  string name = "Journal access";
+
+  if (!osd_trace) {
+    return false;
+  }
+
+  journal_trace = ZTracer::create_ZTrace(name, osd_trace, ep);
+  return true;
+}
+
+void TrackedOp::trace_journal(string event)
+{
+  if (!journal_trace) {
+    return;
+  }
+
+  tracker->trace_event(this, journal_trace, event, journal_trace->get_endpoint());
+}
+
+bool TrackedOp::create_filestore_trace(TrackedOpEndpointRef ep)
+{
+  string name = "Filestore access";
+
+  if (!osd_trace) {
+    return false;
+  }
+
+  filestore_trace = ZTracer::create_ZTrace(name, osd_trace, ep);
+  return true;
+}
+
+void TrackedOp::trace_filestore(string event)
+{
+  if (!filestore_trace) {
+    return;
+  }
+
+  tracker->trace_event(this, filestore_trace, event, filestore_trace->get_endpoint());
+}
+#endif // WITH_BLKIN

79 lines not on this page · raw message

index 0693c09..c8de308 100644
@@ -106,6 +106,7 @@ Pipe::Pipe(SimpleMessenger *r, int st, Connection *con)
   msgr->timeout = msgr->cct->_conf->ms_tcp_read_timeout * 1000; //convert to ms
   if (msgr->timeout == 0)
     msgr->timeout = -1;
+  BLKIN_PIPE_ENDPOINT();
 }
 
 Pipe::~Pipe()

@@ -115,6 +116,33 @@ Pipe::~Pipe()
   delete delay_thread;
 }
 
+#ifdef WITH_BLKIN
+void Pipe::set_endpoint()
+{
+  string type;
+  entity_inst_t inst = msgr->get_myinst();
+
+  if (inst.name.is_client()) {
+    type = "MON";
+  } else if (inst.name.is_mds()) {
+    type = "MDS";
+  } else if (inst.name.is_osd()) {
+    type = "OSD";
+  } else if (inst.name.is_client()) {
+    type = "CLIENT";
+  } else {
+    type = "UNKNOWN";
+  }
+
+  string host;
+  int port;
+
+  inst.addr.to_string(host, port);
+
+  pipe_endpoint = ZTracer::create_ZTraceEndpoint(host, port, "Messenger-" + type);
+}
+#endif // WITH_BLKIN
+
 void Pipe::handle_ack(uint64_t seq)
 {
   lsubdout(msgr->cct, ms, 15) << "reader got ack seq " << seq << dendl;

@@ -1479,6 +1507,7 @@ void Pipe::reader()
 	  fault(true);
 	continue;
       }
+      BLKIN_MSG_TRACE_EVENT(m, message_read);
 
       if (state == STATE_CLOSED ||
 	  state == STATE_CONNECTING) {

@@ -1526,8 +1555,9 @@ void Pipe::reader()
       } else {
 	in_q->enqueue(m, m->get_priority(), conn_id);
       }
-    } 
-    
+      BLKIN_MSG_TRACE_EVENT(m, messenger_end);
+    }
+
     else if (tag == CEPH_MSGR_TAG_CLOSE) {
       ldout(msgr->cct,20) << "reader got CLOSE" << dendl;
       pipe_lock.Lock();

@@ -1695,6 +1725,9 @@ void Pipe::writer()
 
 	pipe_lock.Unlock();
 
+	BLKIN_MSG_TRACE_EVENT(m, writer_sending);
+	BLKIN_MSG_TRACE_EVENT_IF(m->trace_end_after_span, m, Span ended);
+
         ldout(msgr->cct,20) << "writer sending " << m->get_seq() << " " << m << dendl;
 	int rc = write_message(header, footer, blist);
 

index 2070fe5..14cb7be 100644
@@ -112,6 +112,7 @@ int SimpleMessenger::_send_message(Message *m, const entity_inst_t& dest,
 
   lock.Lock();
   Pipe *pipe = _lookup_pipe(dest.addr);
+  BLKIN_MSG_TRACE_EVENT(m, "submitting message");
   submit_message(m, (pipe ? pipe->connection_state.get() : NULL),
                  dest.addr, dest.name.type(), lazy);
   lock.Unlock();

index b02db76..ba36b8d 100644
@@ -132,7 +132,7 @@ bool entity_addr_t::parse(const char *s, const char **end)
 
 
 
-ostream& operator<<(ostream& out, const sockaddr_storage &ss)
+int ss_to_string(const sockaddr_storage &ss, string &host, int &port)
 {
   char buf[NI_MAXHOST] = { 0 };
   char serv[NI_MAXSERV] = { 0 };

@@ -147,8 +147,25 @@ ostream& operator<<(ostream& out, const sockaddr_storage &ss)
   getnameinfo((struct sockaddr *)&ss, hostlen, buf, sizeof(buf),
 	      serv, sizeof(serv),
 	      NI_NUMERICHOST | NI_NUMERICSERV);
+  host = buf;
+  port = atoi(serv);
+  return 0;
+}
+
+int entity_addr_t::to_string(string &host, int &port)
+{
+  return ss_to_string(addr, host, port);
+}
+
+
+ostream& operator<<(ostream& out, const sockaddr_storage &ss)
+{
+  string host;
+  int port;
+
+  ss_to_string(ss, host, port);
   if (ss.ss_family == AF_INET6)
-    return out << '[' << buf << "]:" << serv;
+    return out << '[' << host << "]:" << port;
   return out //<< ss.ss_family << ":"
-	     << buf << ':' << serv;
+	     << host << ':' << port;
 }

index 4e84b4d..f16366c 100644
@@ -418,6 +418,8 @@ int main(int argc, const char **argv)
     global_init_chdir(g_ceph_context);
   }
 
+  BLKIN_ZTRACE_INIT();
+
   MonitorDBStore *store = new MonitorDBStore(g_conf->mon_data);
 
   Monitor::StoreConverter converter(g_conf->mon_data, store);

index 029ef28..efd5b2e 100644
@@ -148,6 +148,9 @@ int main(int argc, const char **argv)
     return 0;
   }
 
+  global_init_daemonize(g_ceph_context, 0);
+  BLKIN_ZTRACE_INIT();
+
   // whoami
   char *end;
   const char *id = g_conf->name.get_id().c_str();

@@ -433,8 +436,7 @@ int main(int argc, const char **argv)
 
   ms_objecter->bind(g_conf->public_addr);
 
-  // Set up crypto, daemonize, etc.
-  global_init_daemonize(g_ceph_context, 0);
+  // Set up crypto, etc.
   common_init_finish(g_ceph_context);
 
   if (g_conf->filestore_update_to >= (int)store->get_target_version()) {

index c6bd616..5377272 100644
@@ -837,8 +837,11 @@ void FileJournal::queue_completions_thru(uint64_t seq)
     }
     if (next.finish)
       finisher->queue(next.finish);
-    if (next.tracked_op)
+    if (next.tracked_op) {
       next.tracked_op->mark_event("journaled_completion_queued");
+      BLKIN_OP_TRACE_EVENT(next.tracked_op, journal, journaled_completion_queued);
+      BLKIN_OP_TRACE_EVENT(next.tracked_op, journal, span_ended);
+    }
   }
   finisher_cond.Signal();
 }

@@ -897,8 +900,10 @@ int FileJournal::prepare_single_write(bufferlist& bl, off64_t& queue_pos, uint64
   }
   bl.append((const char*)&h, sizeof(h));
 
-  if (next_write.tracked_op)
+  if (next_write.tracked_op) {
     next_write.tracked_op->mark_event("write_thread_in_journal_buffer");
+    BLKIN_OP_TRACE_EVENT(next_write.tracked_op, journal, write_thread_in_journal_buffer);
+  }
 
   // pop from writeq
   pop_write();

@@ -1428,8 +1433,11 @@ void FileJournal::submit_entry(uint64_t seq, bufferlist& e, int alignment,
   dout(30) << "XXX throttle take " << e.length() << dendl;
   throttle_ops.take(1);
   throttle_bytes.take(e.length());
-  if (osd_op)
+  if (osd_op) {
     osd_op->mark_event("commit_queued_for_journal_write");
+    BLKIN_OP_CREATE_TRACE(osd_op, journal, journal_endpoint);
+    BLKIN_OP_TRACE_EVENT(osd_op, journal, commit_queued_for_journal_write);
+  }
   if (logger) {
     logger->set(l_os_jq_max_ops, throttle_ops.get_max());
     logger->set(l_os_jq_max_bytes, throttle_bytes.get_max());

index 9d6252c..11b63df 100644
@@ -467,6 +467,7 @@ FileStore::FileStore(const std::string &base, const std::string &jdev, const cha
   m_filestore_max_inline_xattrs(0)
 {
   m_filestore_kill_at.set(g_conf->filestore_kill_at);
+  BLKIN_MSG_END(filestore, "", 0, "Filestore (" + basedir + "(" + name + ")" + ")");
 
   ostringstream oss;
   oss << basedir << "/current";

@@ -1613,6 +1614,9 @@ void FileStore::queue_op(OpSequencer *osr, Op *o)
   // so that regardless of which order the threads pick up the
   // sequencer, the op order will be preserved.
 
+  if (o->osd_op) {
+    BLKIN_OP_TRACE_EVENT(o->osd_op, osd, queueing_for_filestore);
+  }
   osr->queue(o);
 
   logger->inc(l_os_ops);

@@ -1624,6 +1628,9 @@ void FileStore::queue_op(OpSequencer *osr, Op *o)
 	  << "   (queue has " << op_queue_len << " ops and " << op_queue_bytes << " bytes)"
 	  << dendl;
   op_wq.queue(osr);
+  if (o->osd_op) {
+    BLKIN_OP_TRACE_EVENT(o->osd_op, osd, queued_for_filestore);
+  }
 }
 
 void FileStore::op_queue_reserve_throttle(Op *o, ThreadPool::TPHandle *handle)

@@ -1694,8 +1701,17 @@ void FileStore::_do_op(OpSequencer *osr, ThreadPool::TPHandle &handle)
   osr->apply_lock.Lock();
   Op *o = osr->peek_queue();
   apply_manager.op_apply_start(o->op);
+
+  if (o->osd_op) {
+    BLKIN_OP_CREATE_TRACE(o->osd_op, filestore, filestore_endpoint);
+    BLKIN_OP_TRACE_EVENT(o->osd_op, filestore, filestore_dequeued);
+  }
   dout(5) << "_do_op " << o << " seq " << o->op << " " << *osr << "/" << osr->parent << " start" << dendl;
   int r = _do_transactions(o->tls, o->op, &handle);
+
+  if (o->osd_op) {
+    BLKIN_OP_TRACE_EVENT(o->osd_op, filestore, filestore_finished);
+  }
   apply_manager.op_apply_finish(o->op);
   dout(10) << "_do_op " << o << " seq " << o->op << " r = " << r
 	   << ", finisher " << o->onreadable << " " << o->onreadable_sync << dendl;

@@ -1704,6 +1720,10 @@ void FileStore::_do_op(OpSequencer *osr, ThreadPool::TPHandle &handle)
 void FileStore::_finish_op(OpSequencer *osr)
 {
   Op *o = osr->dequeue();
+  if (o->osd_op) {
+    BLKIN_OP_TRACE_EVENT(o->osd_op, filestore, filestore_finishing_op);
+    BLKIN_OP_TRACE_EVENT(o->osd_op, filestore, span_ended);
+  }
   
   dout(10) << "_finish_op " << o << " seq " << o->op << " " << *osr << "/" << osr->parent << dendl;
   osr->apply_lock.Unlock();  // locked in _do_op

index 3c27288..7c0bb79 100644
@@ -724,8 +724,10 @@ struct SubWriteCommitted : public Context {
     : pg(pg), msg(msg), tid(tid),
       version(version), last_complete(last_complete) {}
   void finish(int) {
-    if (msg)
+    if (msg) {
       msg->mark_event("sub_op_committed");
+      BLKIN_OP_TRACE_EVENT(msg, pg, sub_op_committed);
+    }
     pg->sub_write_committed(tid, version, last_complete);
   }
 };

@@ -766,8 +768,10 @@ struct SubWriteApplied : public Context {
     eversion_t version)
     : pg(pg), msg(msg), tid(tid), version(version) {}
   void finish(int) {
-    if (msg)
+    if (msg) {
       msg->mark_event("sub_op_applied");
+      BLKIN_OP_TRACE_EVENT(msg, pg, sub_op_applied);
+    }
     pg->sub_write_applied(tid, version);
   }
 };

index 4240ba8..5b93d80 100644
@@ -21,6 +21,9 @@
 #include <signal.h>
 #include <ctype.h>
 #include <boost/scoped_ptr.hpp>
+#ifdef WITH_BLKIN
+#include <boost/lexical_cast.hpp>
+#endif
 
 #ifdef HAVE_SYS_PARAM_H
 #include <sys/param.h>

@@ -963,6 +966,8 @@ OSD::OSD(CephContext *cct_, ObjectStore *store_,
                                          cct->_conf->osd_op_log_threshold);
   op_tracker.set_history_size_and_duration(cct->_conf->osd_op_history_size,
                                            cct->_conf->osd_op_history_duration);
+
+  BLKIN_MSG_END(osd, "", 0, "osd." + boost::lexical_cast<string>(whoami));
 }
 
 OSD::~OSD()

@@ -5024,6 +5029,8 @@ void OSD::_dispatch(Message *m)
     {
       OpRequestRef op = op_tracker.create_request<OpRequest>(m);
       op->mark_event("waiting_for_osdmap");
+      BLKIN_OP_CREATE_TRACE(op, osd, osd_endpoint);
+      BLKIN_OP_TRACE_EVENT(op, osd, waiting_for_osdmap);
       // no map?  starting up?
       if (!osdmap) {
         dout(7) << "no OSDMap, not booted" << dendl;

@@ -7386,6 +7393,8 @@ void OSD::handle_op(OpRequestRef op)
     return;
   }
 
+  BLKIN_OP_TRACE_EVENT(op, osd, handling_op);
+
   // we don't need encoded payload anymore
   m->clear_payload();
 

@@ -7513,7 +7522,10 @@ void OSD::handle_op(OpRequestRef op)
     return;
   }
 
+  BLKIN_OP_CREATE_TRACE(op, pg, pg->pg_endpoint);
+  BLKIN_OP_TRACE_EVENT(op, pg, enqueuing_op);
   enqueue_op(pg, op);
+  BLKIN_OP_TRACE_EVENT(op, pg, enqueued_op);
 }
 
 template<typename T, int MSGTYPE>

@@ -7522,6 +7534,8 @@ void OSD::handle_replica_op(OpRequestRef op)
   T *m = static_cast<T *>(op->get_req());
   assert(m->get_header().type == MSGTYPE);
 
+  BLKIN_OP_TRACE_EVENT(op, osd, handling_replica_op);
+
   dout(10) << __func__ << " " << *m << " epoch " << m->map_epoch << dendl;
   if (m->map_epoch < up_epoch) {
     dout(3) << "replica op from before up" << dendl;

@@ -7553,7 +7567,9 @@ void OSD::handle_replica_op(OpRequestRef op)
   if (!pg) {
     return;
   }
+  BLKIN_OP_CREATE_TRACE(op, pg, pg->pg_endpoint);
   enqueue_op(pg, op);
+  BLKIN_OP_TRACE_EVENT(op, osd, enqueued_replica_op);
 }
 
 bool OSD::op_is_discardable(MOSDOp *op)

@@ -7655,7 +7671,9 @@ void OSD::OpWQ::_process(PGRef pg, ThreadPool::TPHandle &handle)
   delete f;
   *_dout << dendl;
 
+  BLKIN_OP_TRACE_EVENT(op, pg, dequeuing_op);
   osd->dequeue_op(pg, op, handle);
+  BLKIN_OP_TRACE_EVENT(op, pg, dequeued_op);
   pg->unlock();
 }
 

index 6deb099..263283c 100644
@@ -199,6 +199,8 @@ PG::PG(OSDService *o, OSDMapRef curmap,
 #ifdef PG_DEBUG_REFS
   osd->add_pgid(p, this);
 #endif
+  BLKIN_OSS(oss, "PG " << info.pgid);
+  BLKIN_MSG_END(pg, "", 0, oss.str());
 }
 
 PG::~PG()

index 5a9668f..2b5be9a 100644
@@ -128,6 +128,7 @@ bool ReplicatedBackend::handle_message(
   OpRequestRef op
   )
 {
+  BLKIN_OP_TRACE_EVENT(op, pg, handling_message);
   dout(10) << __func__ << ": " << op << dendl;
   switch (op->get_req()->get_type()) {
   case MSG_OSD_PG_PUSH:

@@ -570,8 +571,10 @@ void ReplicatedBackend::op_applied(
   InProgressOp *op)
 {
   dout(10) << __func__ << ": " << op->tid << dendl;
-  if (op->op)
+  if (op->op) {
     op->op->mark_event("op_applied");
+    BLKIN_OP_TRACE_EVENT(op->op, pg, op_applied);
+  }
 
   op->waiting_for_applied.erase(get_parent()->whoami_shard());
   parent->op_applied(op->v);

@@ -590,8 +593,10 @@ void ReplicatedBackend::op_commit(
   InProgressOp *op)
 {
   dout(10) << __func__ << ": " << op->tid << dendl;
-  if (op->op)
+  if (op->op) {
     op->op->mark_event("op_commit");
+    BLKIN_OP_TRACE_EVENT(op->op, pg, op_commit);
+  }
 
   op->waiting_for_commit.erase(get_parent()->whoami_shard());
 

@@ -640,12 +645,18 @@ void ReplicatedBackend::sub_op_modify_reply(OpRequestRef op)
     if (r->ack_type & CEPH_OSD_FLAG_ONDISK) {
       assert(ip_op.waiting_for_commit.count(from));
       ip_op.waiting_for_commit.erase(from);
-      if (ip_op.op)
+      if (ip_op.op) {
 	ip_op.op->mark_event("sub_op_commit_rec");
+	BLKIN_OP_TRACE_EVENT(ip_op.op, pg, sub_op_commit_rec);
+	BLKIN_MSG_TRACE_EVENT(op->get_req(), span_ended);
+      }
     } else {
       assert(ip_op.waiting_for_applied.count(from));
-      if (ip_op.op)
+      if (ip_op.op) {
 	ip_op.op->mark_event("sub_op_applied_rec");
+	BLKIN_OP_TRACE_EVENT(ip_op.op, pg, sub_op_applied_rec);
+	BLKIN_MSG_TRACE_EVENT(op->get_req(), span_ended);
+      }
     }
     ip_op.waiting_for_applied.erase(from);
 

index 94eec05..6988c6d 100644
@@ -1081,6 +1081,7 @@ void ReplicatedPG::do_request(
   OpRequestRef op,
   ThreadPool::TPHandle &handle)
 {
+  BLKIN_OP_TRACE_EVENT(op, pg, starting_request);
   if (!op_has_sufficient_caps(op)) {
     osd->reply_op_error(op, -EPERM);
     return;

@@ -1190,6 +1191,7 @@ bool ReplicatedPG::check_src_targ(const hobject_t& soid, const hobject_t& toid)
  */
 void ReplicatedPG::do_op(OpRequestRef op)
 {
+  BLKIN_OP_TRACE_EVENT(op, pg, do_op);
   MOSDOp *m = static_cast<MOSDOp*>(op->get_req());
   assert(m->get_header().type == CEPH_MSG_OSD_OP);
   if (op->includes_pg_op()) {

@@ -1200,6 +1202,7 @@ void ReplicatedPG::do_op(OpRequestRef op)
     return do_pg_op(op);
   }
 
+  BLKIN_OP_TRACE_KEYVAL(op, osd, "object", m->get_oid().name);
   if (get_osdmap()->is_blacklisted(m->get_source_addr())) {
     dout(10) << "do_op " << m->get_source_addr() << " is blacklisted" << dendl;
     osd->reply_op_error(op, -EBLACKLISTED);

@@ -1278,6 +1281,7 @@ void ReplicatedPG::do_op(OpRequestRef op)
 	if (m->wants_ack()) {
 	  if (already_ack(oldv)) {
 	    MOSDOpReply *reply = new MOSDOpReply(m, 0, get_osdmap()->get_epoch(), 0, false);
+	    BLKIN_MSG_INIT_TRACE(reply, op->get_osd_trace());
 	    reply->add_flags(CEPH_OSD_FLAG_ACK);
 	    reply->set_reply_versions(oldv, entry->user_version);
 	    osd->send_message_osd_client(reply, m->get_connection());

@@ -1713,6 +1717,8 @@ void ReplicatedPG::execute_ctx(OpContext *ctx)
   const hobject_t& soid = obc->obs.oi.soid;
   map<hobject_t,ObjectContextRef>& src_obc = ctx->src_obc;
 
+  BLKIN_OP_TRACE_EVENT(op, pg, executing_ctx);
+
   // this method must be idempotent since we may call it several times
   // before we finally apply the resulting transaction.
   delete ctx->op_t;

@@ -1804,6 +1810,7 @@ void ReplicatedPG::execute_ctx(OpContext *ctx)
   // prepare the reply
   ctx->reply = new MOSDOpReply(m, 0, get_osdmap()->get_epoch(), 0,
 			       successful_write);
+  BLKIN_MSG_INIT_TRACE(ctx->reply, op->get_osd_trace());
 
   // Write operations aren't allowed to return a data payload because
   // we can't do so reliably. If the client has to resend the request

@@ -1945,6 +1952,8 @@ void ReplicatedPG::do_sub_op(OpRequestRef op)
   assert(m->get_header().type == MSG_OSD_SUBOP);
   dout(15) << "do_sub_op " << *op->get_req() << dendl;
 
+  BLKIN_OP_TRACE_EVENT(op, pg, do_sub_op);
+
   OSDOp *first = NULL;
   if (m->ops.size() >= 1) {
     first = &m->ops[0];

@@ -3010,6 +3019,13 @@ int ReplicatedPG::do_osd_ops(OpContext *ctx, vector<OSDOp>& ops)
     case CEPH_OSD_OP_READ:
       ++ctx->num_read;
       {
+        BLKIN_OP_TRACE_KEYVAL(ctx->op, osd, "type", "read");
+        BLKIN_OSS(oss1, op.extent.offset);
+        BLKIN_OP_TRACE_KEYVAL(ctx->op, osd, "offset", oss1.str());
+        BLKIN_OSS(oss2, op.extent.length);
+        BLKIN_OP_TRACE_KEYVAL(ctx->op, osd, "length", oss2.str());
+      }
+      {
 	__u32 seq = oi.truncate_seq;
 	uint64_t size = oi.size;
 	bool trimmed_read = false;

@@ -6583,12 +6599,14 @@ void ReplicatedPG::eval_repop(RepGather *repop)
 	  repop->ctx->reply = NULL;
 	else {
 	  reply = new MOSDOpReply(m, 0, get_osdmap()->get_epoch(), 0, true);
+	  BLKIN_MSG_INIT_TRACE(reply, repop->ctx->op->get_osd_trace());
 	  reply->set_reply_versions(repop->ctx->at_version,
 	                            repop->ctx->user_at_version);
 	}
 	reply->add_flags(CEPH_OSD_FLAG_ACK | CEPH_OSD_FLAG_ONDISK);
 	dout(10) << " sending commit on " << *repop << " " << reply << dendl;
 	osd->send_message_osd_client(reply, m->get_connection());
+	BLKIN_MSG_TRACE_EVENT(m, replied_commit);
 	repop->sent_disk = true;
 	repop->ctx->op->mark_commit_sent();
       }

@@ -6605,6 +6623,7 @@ void ReplicatedPG::eval_repop(RepGather *repop)
 	     ++i) {
 	  MOSDOp *m = (MOSDOp*)(*i)->get_req();
 	  MOSDOpReply *reply = new MOSDOpReply(m, 0, get_osdmap()->get_epoch(), 0, true);
+	  BLKIN_MSG_INIT_TRACE(reply, repop->ctx->op->get_osd_trace());
 	  reply->set_reply_versions(repop->ctx->at_version,
 	                            repop->ctx->user_at_version);
 	  reply->add_flags(CEPH_OSD_FLAG_ACK);

@@ -6628,6 +6647,7 @@ void ReplicatedPG::eval_repop(RepGather *repop)
         assert(entity_name_t::TYPE_OSD != m->get_connection()->peer_type);
 	osd->send_message_osd_client(reply, m->get_connection());
 	repop->sent_ack = true;
+	BLKIN_MSG_TRACE_EVENT(m, replied_ack);
       }
 
       // note the write is now readable (for rlatency calc).  note

@@ -6660,6 +6680,8 @@ void ReplicatedPG::eval_repop(RepGather *repop)
       dout(0) << "   q front is " << *repop_queue.front() << dendl; 
       assert(repop_queue.front() == repop);
     }
+    BLKIN_OP_TRACE_EVENT(repop->ctx->op, pg, all_done);
+    BLKIN_MSG_TRACE_EVENT(m, span_ended);
     repop_queue.pop_front();
     remove_repop(repop);
   }

@@ -6679,6 +6701,8 @@ void ReplicatedPG::issue_repop(RepGather *repop, utime_t now)
           << " o " << soid
           << dendl;
 
+  BLKIN_OP_TRACE_EVENT(ctx->op, pg, issuing_repop);
+
   repop->v = ctx->at_version;
   if (ctx->at_version > eversion_t()) {
     for (set<pg_shard_t>::iterator i = actingbackfill.begin();

@@ -6750,6 +6774,7 @@ void ReplicatedBackend::issue_op(
   InProgressOp *op,
   ObjectStore::Transaction *op_t)
 {
+  BLKIN_OP_TRACE_EVENT(op->op, pg, issuing_replication);
   int acks_wanted = CEPH_OSD_FLAG_ACK | CEPH_OSD_FLAG_ONDISK;
 
   if (parent->get_actingbackfill_shards().size() > 1) {

@@ -6777,6 +6802,8 @@ void ReplicatedBackend::issue_op(
       get_osdmap()->get_epoch(),
       tid, at_version);
 
+    BLKIN_MSG_INIT_TRACE_IF(op->op->get_req(), wr, op->op->get_osd_trace());
+
     // ship resulting transaction, log entries, and pg_stats
     if (!parent->should_send_op(peer, soid)) {
       dout(10) << "issue_repop shipping empty opt to osd." << peer

@@ -7628,6 +7655,7 @@ void ReplicatedBackend::sub_op_modify(OpRequestRef op)
 void ReplicatedBackend::sub_op_modify_applied(RepModifyRef rm)
 {
   rm->op->mark_event("sub_op_applied");
+  BLKIN_OP_TRACE_EVENT(rm->op, pg, sub_op_applied);
   rm->applied = true;
 
   dout(10) << "sub_op_modify_applied on " << rm << " op "

@@ -7641,6 +7669,9 @@ void ReplicatedBackend::sub_op_modify_applied(RepModifyRef rm)
       m, parent->whoami_shard(),
       0, get_osdmap()->get_epoch(), CEPH_OSD_FLAG_ACK);
     ack->set_priority(CEPH_MSG_PRIO_HIGH); // this better match commit priority!
+    BLKIN_MSG_INIT_TRACE(ack, rm->op->get_osd_trace());
+    BLKIN_MSG_TRACE_EVENT(rm->op->get_req(), replied_apply);
+    BLKIN_MSG_TRACE_EVENT(rm->op->get_req(), span_ended);
     get_parent()->send_message_osd_cluster(
       rm->ackerosd, ack, get_osdmap()->get_epoch());
   }

@@ -7666,6 +7697,9 @@ void ReplicatedBackend::sub_op_modify_commit(RepModifyRef rm)
     0, get_osdmap()->get_epoch(), CEPH_OSD_FLAG_ONDISK);
   commit->set_last_complete_ondisk(rm->last_complete);
   commit->set_priority(CEPH_MSG_PRIO_HIGH); // this better match ack priority!
+  BLKIN_MSG_INIT_TRACE(commit, rm->op->get_osd_trace());
+  BLKIN_MSG_TRACE_EVENT(rm->op->get_req(), replied_commit);
+  BLKIN_MSG_TRACE_EVENT(rm->op->get_req(), span_ended);
   get_parent()->send_message_osd_cluster(
     rm->ackerosd, commit, get_osdmap()->get_epoch());
   

@@ -8345,6 +8379,7 @@ struct C_OnPushCommit : public Context {
   OpRequestRef op;
   C_OnPushCommit(ReplicatedPG *pg, OpRequestRef op) : pg(pg), op(op) {}
   void finish(int) {
+    BLKIN_OP_TRACE_EVENT(op, pg, committed);
     op->mark_event("committed");
     log_subop_stats(pg->osd->logger, op, l_osd_push_inb, l_osd_sop_push_lat);
   }

index 9da65b0..f421f5b 100644
@@ -1650,6 +1650,7 @@ void Objecter::send_op(Op *op)
   m->ops = op->ops;
   m->set_mtime(op->mtime);
   m->set_retry_attempt(op->attempts++);
+  BLKIN_MSG_INIT_TRACE_IF(op->trace, m, op->trace);
 
   if (op->replay_version != eversion_t())
     m->set_version(op->replay_version);  // we're replaying this op!

@@ -1724,6 +1725,7 @@ void Objecter::handle_osd_op_reply(MOSDOpReply *m)
     ldout(cct, 7) << "handle_osd_op_reply " << tid
 	    << (m->is_ondisk() ? " ondisk":(m->is_onnvram() ? " onnvram":" ack"))
 	    << " ... stray" << dendl;
+    BLKIN_MSG_TRACE_EVENT(m, span_ended);
     m->put();
     return;
   }

@@ -1736,6 +1738,10 @@ void Objecter::handle_osd_op_reply(MOSDOpReply *m)
 		<< dendl;
   Op *op = ops[tid];
 
+  BLKIN_MSG_TRACE_EVENT_IF(op->oncommit, m, oncommit_message);
+  BLKIN_MSG_TRACE_EVENT_IF(op->onack, m, onack_message);
+  BLKIN_MSG_TRACE_EVENT(m, span_ended);
+
   if (m->get_retry_attempt() >= 0) {
     if (m->get_retry_attempt() != (op->attempts - 1)) {
       ldout(cct, 7) << " ignoring reply from attempt " << m->get_retry_attempt()

@@ -1794,6 +1800,7 @@ void Objecter::handle_osd_op_reply(MOSDOpReply *m)
   assert(op->out_bl.size() == op->out_rval.size());
   assert(op->out_bl.size() == op->out_handler.size());
   vector<OSDOp>::iterator p = out_ops.begin();
+  BLKIN_OP_EVENT_IF(op->trace, op->trace, in_handle_osd_op_reply);
   for (unsigned i = 0;
        p != out_ops.end() && pb != op->out_bl.end();
        ++i, ++p, ++pb, ++pr, ++ph) {

105 lines not on this page · raw message

index 249c34f..caa0e18 100644
@@ -84,6 +84,7 @@ librados::RadosClient::RadosClient(CephContext *cct_)
     finisher(cct),
     max_watch_cookie(0)
 {
+  BLKIN_ZTRACE_INIT();
 }
 
 int64_t librados::RadosClient::lookup_pool(const char *name)

index 2358fb4..14199f3 100644
@@ -2900,6 +2900,20 @@ extern "C" int rados_aio_read(rados_ioctx_t io, const char *o,
 		       buf, len, off, ctx->snap_seq);
 }
 
+#ifdef WITH_BLKIN
+extern "C" int rados_aio_read_traced(rados_ioctx_t io, const char *o,
+				rados_completion_t completion,
+				char *buf, size_t len, uint64_t off,
+				struct blkin_trace_info *info)
+{
+	librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
+	object_t oid(o);
+	return ctx->aio_read_traced(oid,
+			(librados::AioCompletionImpl*)completion, buf, len,
+			off, ctx->snap_seq, info);
+}
+#endif
+
 extern "C" int rados_aio_write(rados_ioctx_t io, const char *o,
 				rados_completion_t completion,
 				const char *buf, size_t len, uint64_t off)

@@ -2912,6 +2926,22 @@ extern "C" int rados_aio_write(rados_ioctx_t io, const char *o,
 			bl, len, off);
 }
 
+#ifdef WITH_BLKIN
+extern "C" int rados_aio_write_traced(rados_ioctx_t io, const char *o,
+		rados_completion_t completion,
+		const char *buf, size_t len, uint64_t off,
+		struct blkin_trace_info *info)
+{
+	librados::IoCtxImpl *ctx = (librados::IoCtxImpl *)io;
+	object_t oid(o);
+	bufferlist bl;
+	bl.append(buf, len);
+	return ctx->aio_write_traced(oid,
+			(librados::AioCompletionImpl*)completion,bl, len, off,
+			info);
+}
+#endif
+
 extern "C" int rados_aio_append(rados_ioctx_t io, const char *o,
 				rados_completion_t completion,
 				const char *buf, size_t len)

index a40051f..c0ede9f 100644
@@ -1,5 +1,6 @@
-all: hello_world.cc
+all: hello_world.cc hello_blkin.c
+	gcc hello_blkin.c -o hello_blkin -lrados -lblkin-front
 	g++ -g -c hello_world.cc -o hello_world.o
 	g++ -g hello_world.o -lrados -o librados_hello_world
 clean:
-	rm hello_world.o librados_hello_world
\ No newline at end of file
+	rm hello_world.o librados_hello_world hello