objecter: add an admin socket command to get in-flight requests

Fixes: #1881
Signed-off-by: Josh Durgin <josh.durgin@dreamhost.com>

2 files changed, 182 insertions, 0 deletions

diff --git a/src/osdc/Objecter.cc b/src/osdc/Objecter.cc
index 43b41da1cbc..eb70aea2aa1 100644
--- a/src/osdc/Objecter.cc
+++ b/src/osdc/Objecter.cc
@@ -199,6 +199,16 @@ void Objecter::init()
     cct->get_perfcounters_collection()->add(logger);
   }
 
+  m_request_state_hook = new RequestStateHook(this);
+  AdminSocket* admin_socket = cct->get_admin_socket();
+  int ret = admin_socket->register_command("objecter_requests",
+					   m_request_state_hook,
+					   "show in-progress osd requests");
+  if (ret < 0) {
+    lderr(cct) << "error registering admin socket command: "
+	       << cpp_strerror(-ret) << dendl;
+  }
+
   schedule_tick();
   maybe_request_map();
 }
@@ -216,6 +226,13 @@ void Objecter::shutdown()
     tick_event = NULL;
   }
 
+  if (m_request_state_hook) {
+    AdminSocket* admin_socket = cct->get_admin_socket();
+    admin_socket->unregister_command("objecter_requests");
+    delete m_request_state_hook;
+    m_request_state_hook = NULL;
+  }
+
   if (logger) {
     cct->get_perfcounters_collection()->remove(logger);
     delete logger;
@@ -1845,3 +1862,145 @@ void Objecter::dump_active()
   }
 }
 
+void Objecter::dump_requests(Formatter& fmt) const
+{
+  assert(client_lock.is_locked());
+
+  fmt.open_object_section("requests");
+  dump_ops(fmt);
+  dump_linger_ops(fmt);
+  dump_pool_ops(fmt);
+  dump_pool_stat_ops(fmt);
+  dump_statfs_ops(fmt);
+  fmt.close_section(); // requests object
+}
+
+void Objecter::dump_ops(Formatter& fmt) const
+{
+  fmt.open_array_section("ops");
+  for (hash_map<tid_t,Op*>::const_iterator p = ops.begin();
+       p != ops.end();
+       ++p) {
+    Op *op = p->second;
+    fmt.open_object_section("op");
+    fmt.dump_unsigned("tid", op->tid);
+    fmt.dump_stream("pg") << op->pgid;
+    fmt.dump_int("osd", op->session ? op->session->osd : -1);
+    fmt.dump_stream("last_sent") << op->stamp;
+    fmt.dump_int("attempts", op->attempts);
+    fmt.dump_stream("object_id") << op->oid;
+    fmt.dump_stream("object_locator") << op->oloc;
+    fmt.dump_stream("snapid") << op->snapid;
+    fmt.dump_stream("snap_context") << op->snapc;
+    fmt.dump_stream("mtime") << op->mtime;
+
+    fmt.open_array_section("osd_ops");
+    for (vector<OSDOp>::const_iterator it = op->ops.begin();
+	 it != op->ops.end();
+	 ++it) {
+      fmt.dump_stream("osd_op") << *it;
+    }
+    fmt.close_section(); // osd_ops array
+
+    fmt.close_section(); // op object
+  }
+  fmt.close_section(); // ops array
+}
+
+void Objecter::dump_linger_ops(Formatter& fmt) const
+{
+  fmt.open_array_section("linger_ops");
+  for (map<uint64_t, LingerOp*>::const_iterator p = linger_ops.begin();
+       p != linger_ops.end();
+       ++p) {
+    LingerOp *op = p->second;
+    fmt.open_object_section("linger_op");
+    fmt.dump_unsigned("linger_id", op->linger_id);
+    fmt.dump_stream("pg") << op->pgid;
+    fmt.dump_int("osd", op->session ? op->session->osd : -1);
+    fmt.dump_stream("object_id") << op->oid;
+    fmt.dump_stream("object_locator") << op->oloc;
+    fmt.dump_stream("snapid") << op->snap;
+    fmt.dump_stream("registering") << op->snap;
+    fmt.dump_stream("registered") << op->snap;
+    fmt.close_section(); // linger_op object
+  }
+  fmt.close_section(); // linger_ops array
+}
+
+void Objecter::dump_pool_ops(Formatter& fmt) const
+{
+  fmt.open_array_section("pool_ops");
+  for (map<tid_t, PoolOp*>::const_iterator p = pool_ops.begin();
+       p != pool_ops.end();
+       ++p) {
+    PoolOp *op = p->second;
+    fmt.open_object_section("pool_op");
+    fmt.dump_unsigned("tid", op->tid);
+    fmt.dump_int("pool", op->pool);
+    fmt.dump_string("name", op->name);
+    fmt.dump_int("operation_type", op->pool_op);
+    fmt.dump_unsigned("auid", op->auid);
+    fmt.dump_unsigned("crush_rule", op->crush_rule);
+    fmt.dump_stream("snapid") << op->snapid;
+    fmt.dump_stream("last_sent") << op->last_submit;
+    fmt.close_section(); // pool_op object
+  }
+  fmt.close_section(); // pool_ops array
+}
+
+void Objecter::dump_pool_stat_ops(Formatter& fmt) const
+{
+  fmt.open_array_section("pool_stat_ops");
+  for (map<tid_t, PoolStatOp*>::const_iterator p = poolstat_ops.begin();
+       p != poolstat_ops.end();
+       ++p) {
+    PoolStatOp *op = p->second;
+    fmt.open_object_section("pool_stat_op");
+    fmt.dump_unsigned("tid", op->tid);
+    fmt.dump_stream("last_sent") << op->last_submit;
+
+    fmt.open_array_section("pools");
+    for (list<string>::const_iterator it = op->pools.begin();
+	 it != op->pools.end();
+	 ++it) {
+      fmt.dump_string("pool", *it);
+    }
+    fmt.close_section(); // pool_op object
+
+    fmt.close_section(); // pool_stat_op object
+  }
+  fmt.close_section(); // pool_stat_ops array
+}
+
+void Objecter::dump_statfs_ops(Formatter& fmt) const
+{
+  fmt.open_array_section("statfs_ops");
+  for (map<tid_t, StatfsOp*>::const_iterator p = statfs_ops.begin();
+       p != statfs_ops.end();
+       ++p) {
+    StatfsOp *op = p->second;
+    fmt.open_object_section("statfs_op");
+    fmt.dump_unsigned("tid", op->tid);
+    fmt.dump_stream("last_sent") << op->last_submit;
+    fmt.close_section(); // pool_stat_op object
+  }
+  fmt.close_section(); // pool_stat_ops array
+}
+
+Objecter::RequestStateHook::RequestStateHook(Objecter *objecter) :
+  m_objecter(objecter)
+{
+}
+
+bool Objecter::RequestStateHook::call(std::string command, bufferlist& out)
+{
+  stringstream ss;
+  JSONFormatter formatter(true);
+  m_objecter->client_lock.Lock();
+  m_objecter->dump_requests(formatter);
+  m_objecter->client_lock.Unlock();
+  formatter.flush(ss);
+  out.append(ss);
+  return true;
+}
diff --git a/src/osdc/Objecter.h b/src/osdc/Objecter.h
index 61b6e8f7ccb..d8ebd8e07cb 100644
--- a/src/osdc/Objecter.h
+++ b/src/osdc/Objecter.h
@@ -22,6 +22,7 @@
 #include "osd/OSDMap.h"
 #include "messages/MOSDOp.h"
 
+#include "common/admin_socket.h"
 #include "common/Timer.h"
 
 #include <list>
@@ -424,6 +425,15 @@ class Objecter {
   void schedule_tick();
   void tick();
 
+  class RequestStateHook : public AdminSocketHook {
+    Objecter *m_objecter;
+  public:
+    RequestStateHook(Objecter *objecter);
+    bool call(std::string command, bufferlist& out);
+  };
+
+  RequestStateHook *m_request_state_hook;
+
 public:
   /*** track pending operations ***/
   // read
@@ -774,10 +784,13 @@ public:
     last_seen_pgmap_version(0),
     client_lock(l), timer(t),
     logger(NULL), tick_event(NULL),
+    m_request_state_hook(NULL),
     num_homeless_ops(0),
     op_throttler(cct->_conf->objecter_inflight_op_bytes)
   { }
   ~Objecter() {
+    assert(!tick_event);
+    assert(!m_request_state_hook);
     assert(!logger);
   }
 
@@ -813,7 +826,17 @@ private:
   bool is_active() {
     return !(ops.empty() && linger_ops.empty() && poolstat_ops.empty() && statfs_ops.empty());
   }
+
+  /**
+   * Output in-flight requests
+   */
   void dump_active();
+  void dump_requests(Formatter& fmt) const;
+  void dump_ops(Formatter& fmt) const;
+  void dump_linger_ops(Formatter& fmt) const;
+  void dump_pool_ops(Formatter& fmt) const;
+  void dump_pool_stat_ops(Formatter& fmt) const;
+  void dump_statfs_ops(Formatter& fmt) const;
 
   int get_client_incarnation() const { return client_inc; }
   void set_client_incarnation(int inc) { client_inc = inc; }