ceph/src/tools/rbd_mirror/PoolReplayer.h

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3
   4 #ifndef CEPH_RBD_MIRROR_POOL_REPLAYER_H
   5 #define CEPH_RBD_MIRROR_POOL_REPLAYER_H
   6
   7 #include "common/AsyncOpTracker.h"
   8 #include "common/Cond.h"
   9 #include "common/Mutex.h"
  10 #include "common/WorkQueue.h"
  11 #include "include/rados/librados.hpp"
  12
  13 #include "ClusterWatcher.h"
  14 #include "LeaderWatcher.h"
  15 #include "PoolWatcher.h"
  16 #include "ImageDeleter.h"
  17 #include "types.h"
  18 #include "tools/rbd_mirror/service_daemon/Types.h"
  19
  20 #include <set>
  21 #include <map>
  22 #include <memory>
  23 #include <atomic>
  24 #include <string>
  25
  26 class AdminSocketHook;
  27
  28 namespace librbd { class ImageCtx; }
  29
  30 namespace rbd {
  31 namespace mirror {
  32
  33 template <typename> class InstanceReplayer;
  34 template <typename> class InstanceWatcher;
  35 template <typename> class ServiceDaemon;
  36 template <typename> struct Threads;
  37
  38 /**
  39  * Controls mirroring for a single remote cluster.
  40  */
  41 class PoolReplayer {
  42 public:
  43   PoolReplayer(Threads<librbd::ImageCtx> *threads,
  44                ServiceDaemon<librbd::ImageCtx>* service_daemon,
  45                ImageDeleter<>* image_deleter,
  46                int64_t local_pool_id, const peer_t &peer,
  47                const std::vector<const char*> &args);
  48   ~PoolReplayer();
  49   PoolReplayer(const PoolReplayer&) = delete;
  50   PoolReplayer& operator=(const PoolReplayer&) = delete;
  51
  52   bool is_blacklisted() const;
  53   bool is_leader() const;
  54   bool is_running() const;
  55
  56   void init();
  57   void shut_down();
  58
  59   void run();
  60
  61   void print_status(Formatter *f, stringstream *ss);
  62   void start();
  63   void stop(bool manual);
  64   void restart();
  65   void flush();
  66   void release_leader();
  67
  68 private:
  69   struct PoolWatcherListener : public PoolWatcher<>::Listener {
  70     PoolReplayer *pool_replayer;
  71     bool local;
  72
  73     PoolWatcherListener(PoolReplayer *pool_replayer, bool local)
  74       : pool_replayer(pool_replayer), local(local) {
  75     }
  76
  77     void handle_update(const std::string &mirror_uuid,
  78                        ImageIds &&added_image_ids,
  79                        ImageIds &&removed_image_ids) override {
  80       pool_replayer->handle_update((local ? "" : mirror_uuid),
  81                                    std::move(added_image_ids),
  82                                    std::move(removed_image_ids));
  83     }
  84   };
  85
  86   void handle_update(const std::string &mirror_uuid,
  87                      ImageIds &&added_image_ids,
  88                      ImageIds &&removed_image_ids);
  89
  90   int init_rados(const std::string &cluster_name,
  91                  const std::string &client_name,
  92                  const std::string &description, RadosRef *rados_ref);
  93
  94   void handle_post_acquire_leader(Context *on_finish);
  95   void handle_pre_release_leader(Context *on_finish);
  96
  97   void init_local_pool_watcher(Context *on_finish);
  98   void handle_init_local_pool_watcher(int r, Context *on_finish);
  99
 100   void init_remote_pool_watcher(Context *on_finish);
 101
 102   void shut_down_pool_watchers(Context *on_finish);
 103   void handle_shut_down_pool_watchers(int r, Context *on_finish);
 104
 105   void wait_for_update_ops(Context *on_finish);
 106   void handle_wait_for_update_ops(int r, Context *on_finish);
 107
 108   void handle_update_leader(const std::string &leader_instance_id);
 109
 110   Threads<librbd::ImageCtx> *m_threads;
 111   ServiceDaemon<librbd::ImageCtx>* m_service_daemon;
 112   ImageDeleter<>* m_image_deleter;
 113   int64_t m_local_pool_id = -1;
 114   peer_t m_peer;
 115   std::vector<const char*> m_args;
 116
 117   mutable Mutex m_lock;
 118   Cond m_cond;
 119   std::atomic<bool> m_stopping = { false };
 120   bool m_manual_stop = false;
 121   bool m_blacklisted = false;
 122
 123   RadosRef m_local_rados;
 124   RadosRef m_remote_rados;
 125
 126   librados::IoCtx m_local_io_ctx;
 127   librados::IoCtx m_remote_io_ctx;
 128
 129   PoolWatcherListener m_local_pool_watcher_listener;
 130   std::unique_ptr<PoolWatcher<> > m_local_pool_watcher;
 131
 132   PoolWatcherListener m_remote_pool_watcher_listener;
 133   std::unique_ptr<PoolWatcher<> > m_remote_pool_watcher;
 134
 135   std::unique_ptr<InstanceReplayer<librbd::ImageCtx>> m_instance_replayer;
 136
 137   std::string m_asok_hook_name;
 138   AdminSocketHook *m_asok_hook = nullptr;
 139
 140   service_daemon::CalloutId m_callout_id = service_daemon::CALLOUT_ID_NONE;
 141
 142   class PoolReplayerThread : public Thread {
 143     PoolReplayer *m_pool_replayer;
 144   public:
 145     PoolReplayerThread(PoolReplayer *pool_replayer)
 146       : m_pool_replayer(pool_replayer) {
 147     }
 148     void *entry() override {
 149       m_pool_replayer->run();
 150       return 0;
 151     }
 152   } m_pool_replayer_thread;
 153
 154   class LeaderListener : public LeaderWatcher<>::Listener {
 155   public:
 156     LeaderListener(PoolReplayer *pool_replayer)
 157       : m_pool_replayer(pool_replayer) {
 158     }
 159
 160   protected:
 161     void post_acquire_handler(Context *on_finish) override {
 162       m_pool_replayer->handle_post_acquire_leader(on_finish);
 163     }
 164
 165     void pre_release_handler(Context *on_finish) override {
 166       m_pool_replayer->handle_pre_release_leader(on_finish);
 167     }
 168
 169     void update_leader_handler(
 170       const std::string &leader_instance_id) override {
 171       m_pool_replayer->handle_update_leader(leader_instance_id);
 172     }
 173
 174   private:
 175     PoolReplayer *m_pool_replayer;
 176   } m_leader_listener;
 177
 178   std::unique_ptr<LeaderWatcher<> > m_leader_watcher;
 179   std::unique_ptr<InstanceWatcher<librbd::ImageCtx> > m_instance_watcher;
 180   AsyncOpTracker m_update_op_tracker;
 181 };
 182
 183 } // namespace mirror
 184 } // namespace rbd
 185
 186 #endif // CEPH_RBD_MIRROR_POOL_REPLAYER_H