rbd-mirror: add asok commands to get image replyer status and flush

Signed-off-by: Mykola Golub <mgolub@mirantis.com>

diff --git a/src/tools/rbd_mirror/ImageReplayer.cc b/src/tools/rbd_mirror/ImageReplayer.cc
index 47c863f07e97642fb0c9d050bdf786489b9bf85b..255fae2c9e078c652286a0c18b3419d66bbe6a03 100644 (file)
--- a/src/tools/rbd_mirror/ImageReplayer.cc
+++ b/src/tools/rbd_mirror/ImageReplayer.cc
@@ -1,6 +1,7 @@
 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
 // vim: ts=8 sw=2 smarttab
 
+#include "common/Formatter.h"
 #include "common/debug.h"
 #include "common/errno.h"
 #include "include/stringify.h"
@@ -61,8 +62,102 @@ struct C_ReplayCommitted : public Context {
   }
 };
 
+class ImageReplayerAdminSocketCommand {
+public:
+  virtual ~ImageReplayerAdminSocketCommand() {}
+  virtual bool call(Formatter *f, stringstream *ss) = 0;
+};
+
+class StatusCommand : public ImageReplayerAdminSocketCommand {
+public:
+  explicit StatusCommand(ImageReplayer *replayer) : replayer(replayer) {}
+
+  bool call(Formatter *f, stringstream *ss) {
+    if (f) {
+      f->open_object_section("status");
+      f->dump_stream("state") << replayer->get_state();
+      f->close_section();
+      f->flush(*ss);
+    } else {
+      *ss << "state: " << replayer->get_state();
+    }
+    return true;
+  }
+
+private:
+  ImageReplayer *replayer;
+};
+
+class FlushCommand : public ImageReplayerAdminSocketCommand {
+public:
+  explicit FlushCommand(ImageReplayer *replayer) : replayer(replayer) {}
+
+  bool call(Formatter *f, stringstream *ss) {
+    int r = replayer->flush();
+    if (r < 0) {
+      *ss << "flush: " << cpp_strerror(r);
+      return false;
+    }
+    return true;
+  }
+
+private:
+  ImageReplayer *replayer;
+};
+
 } // anonymous namespace
 
+class ImageReplayerAdminSocketHook : public AdminSocketHook {
+public:
+  ImageReplayerAdminSocketHook(CephContext *cct, ImageReplayer *replayer) :
+    admin_socket(cct->get_admin_socket()) {
+    std::string command;
+    int r;
+
+    command = "rbd mirror status " + stringify(*replayer);
+    r = admin_socket->register_command(command, command, this,
+                                      "get status for rbd mirror " +
+                                      stringify(*replayer));
+    if (r == 0) {
+      commands[command] = new StatusCommand(replayer);
+    }
+
+    command = "rbd mirror flush " + stringify(*replayer);
+    r = admin_socket->register_command(command, command, this,
+                                      "flush rbd mirror " +
+                                      stringify(*replayer));
+    if (r == 0) {
+      commands[command] = new FlushCommand(replayer);
+    }
+  }
+
+  ~ImageReplayerAdminSocketHook() {
+    for (Commands::const_iterator i = commands.begin(); i != commands.end();
+        ++i) {
+      (void)admin_socket->unregister_command(i->first);
+      delete i->second;
+    }
+  }
+
+  bool call(std::string command, cmdmap_t& cmdmap, std::string format,
+           bufferlist& out) {
+    Commands::const_iterator i = commands.find(command);
+    assert(i != commands.end());
+    Formatter *f = Formatter::create(format);
+    stringstream ss;
+    bool r = i->second->call(f, &ss);
+    delete f;
+    out.append(ss);
+    return r;
+  }
+
+private:
+  typedef std::map<std::string, ImageReplayerAdminSocketCommand*> Commands;
+
+  AdminSocket *admin_socket;
+  Commands commands;
+};
+
 ImageReplayer::ImageReplayer(RadosRef local, RadosRef remote,
                             const std::string &client_id,
                             int64_t remote_pool_id,
@@ -81,6 +176,9 @@ ImageReplayer::ImageReplayer(RadosRef local, RadosRef remote,
   m_remote_journaler(nullptr),
   m_replay_handler(nullptr)
 {
+  CephContext *cct = static_cast<CephContext *>(m_local->cct());
+
+  m_asok_hook = new ImageReplayerAdminSocketHook(cct, this);
 }
 
 ImageReplayer::~ImageReplayer()
@@ -89,6 +187,8 @@ ImageReplayer::~ImageReplayer()
   assert(m_local_replay == nullptr);
   assert(m_remote_journaler == nullptr);
   assert(m_replay_handler == nullptr);
+
+  delete m_asok_hook;
 }
 
 int ImageReplayer::start(const BootstrapParams *bootstrap_params)
@@ -732,6 +832,34 @@ void ImageReplayer::shut_down_journal_replay()
   }
 }
 
+std::ostream &operator<<(std::ostream &os, const ImageReplayer::State &state)
+{
+  switch (state) {
+  case ImageReplayer::STATE_UNINITIALIZED:
+    os << "Uninitialized";
+    break;
+  case ImageReplayer::STATE_STARTING:
+    os << "Starting";
+    break;
+  case ImageReplayer::STATE_REPLAYING:
+    os << "Replaying";
+    break;
+  case ImageReplayer::STATE_FLUSHING_REPLAY:
+    os << "FlushingReplay";
+    break;
+  case ImageReplayer::STATE_STOPPING:
+    os << "Stopping";
+    break;
+  case ImageReplayer::STATE_STOPPED:
+    os << "Stopped";
+    break;
+  default:
+    os << "Unknown(" << state << ")";
+    break;
+  }
+  return os;
+}
+
 std::ostream &operator<<(std::ostream &os, const ImageReplayer &replayer)
 {
   os << "ImageReplayer[" << replayer.m_remote_pool_id << "/"

diff --git a/src/tools/rbd_mirror/ImageReplayer.h b/src/tools/rbd_mirror/ImageReplayer.h
index a504f10009cb98de6ff7f8fded1080c22ba86253..c37c5f03cdcecc3dff9a0bfeb38d67b3686c9b15 100644 (file)
--- a/src/tools/rbd_mirror/ImageReplayer.h
+++ b/src/tools/rbd_mirror/ImageReplayer.h
@@ -36,6 +36,8 @@ template <typename> class Replay;
 namespace rbd {
 namespace mirror {
 
+class ImageReplayerAdminSocketHook;
+
 /**
  * Replays changes from a remote cluster for a single image.
  */
@@ -68,6 +70,8 @@ public:
   ImageReplayer(const ImageReplayer&) = delete;
   ImageReplayer& operator=(const ImageReplayer&) = delete;
 
+  State get_state() { return m_state; }
+
   int start(const BootstrapParams *bootstrap_params = nullptr);
   void stop();
 
@@ -107,6 +111,7 @@ private:
   librbd::journal::Replay<librbd::ImageCtx> *m_local_replay;
   ::journal::Journaler *m_remote_journaler;
   ::journal::ReplayHandler *m_replay_handler;
+  ImageReplayerAdminSocketHook *m_asok_hook;
 };
 
 } // namespace mirror