ceph/src/tools/cephfs_mirror/Mirror.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 CEPHFS_MIRROR_H
   5 #define CEPHFS_MIRROR_H
   6
   7 #include <map>
   8 #include <set>
   9 #include <vector>
  10
  11 #include "common/ceph_mutex.h"
  12 #include "common/WorkQueue.h"
  13 #include "mds/FSMap.h"
  14 #include "ClusterWatcher.h"
  15 #include "FSMirror.h"
  16 #include "ServiceDaemon.h"
  17 #include "Types.h"
  18
  19 class Messenger;
  20 class MonClient;
  21 class ContextWQ;
  22
  23 namespace cephfs {
  24 namespace mirror {
  25
  26 // this wraps up ClusterWatcher and FSMirrors to implement mirroring
  27 // for ceph filesystems.
  28
  29 class Mirror {
  30 public:
  31   Mirror(CephContext *cct, const std::vector<const char*> &args,
  32          MonClient *monc, Messenger *msgr);
  33   ~Mirror();
  34
  35   int init(std::string &reason);
  36   void shutdown();
  37   void run();
  38
  39   void handle_signal(int signum);
  40
  41 private:
  42   static constexpr std::string_view MIRRORING_MODULE = "mirroring";
  43
  44   struct C_EnableMirroring;
  45   struct C_DisableMirroring;
  46   struct C_PeerUpdate;
  47   struct C_RestartMirroring;
  48
  49   struct ClusterListener : ClusterWatcher::Listener {
  50     Mirror *mirror;
  51
  52     ClusterListener(Mirror *mirror)
  53       : mirror(mirror) {
  54     }
  55
  56     void handle_mirroring_enabled(const FilesystemSpec &spec) override {
  57       mirror->mirroring_enabled(spec.filesystem, spec.pool_id);
  58     }
  59
  60     void handle_mirroring_disabled(const Filesystem &filesystem) override {
  61       mirror->mirroring_disabled(filesystem);
  62     }
  63
  64     void handle_peers_added(const Filesystem &filesystem, const Peer &peer) override {
  65       mirror->peer_added(filesystem, peer);
  66     }
  67
  68     void handle_peers_removed(const Filesystem &filesystem, const Peer &peer) override {
  69       mirror->peer_removed(filesystem, peer);
  70     }
  71   };
  72
  73   struct MirrorAction {
  74     MirrorAction(uint64_t pool_id) :
  75       pool_id(pool_id) {
  76     }
  77
  78     uint64_t pool_id; // for restarting blocklisted mirror instance
  79     bool action_in_progress = false;
  80     bool restarting = false;
  81     std::list<Context *> action_ctxs;
  82     std::unique_ptr<FSMirror> fs_mirror;
  83   };
  84
  85   ceph::mutex m_lock = ceph::make_mutex("cephfs::mirror::Mirror");
  86   ceph::condition_variable m_cond;
  87
  88   CephContext *m_cct;
  89   std::vector<const char *> m_args;
  90   MonClient *m_monc;
  91   Messenger *m_msgr;
  92   ClusterListener m_listener;
  93
  94   ThreadPool *m_thread_pool = nullptr;
  95   ContextWQ *m_work_queue = nullptr;
  96   SafeTimer *m_timer = nullptr;
  97   ceph::mutex *m_timer_lock = nullptr;
  98   Context *m_timer_task = nullptr;
  99
 100   bool m_stopping = false;
 101   std::unique_ptr<ClusterWatcher> m_cluster_watcher;
 102   std::map<Filesystem, MirrorAction> m_mirror_actions;
 103
 104   RadosRef m_local;
 105   std::unique_ptr<ServiceDaemon> m_service_daemon;
 106
 107   int init_mon_client();
 108
 109   // called via listener
 110   void mirroring_enabled(const Filesystem &filesystem, uint64_t local_pool_id);
 111   void mirroring_disabled(const Filesystem &filesystem);
 112   void peer_added(const Filesystem &filesystem, const Peer &peer);
 113   void peer_removed(const Filesystem &filesystem, const Peer &peer);
 114
 115   // mirror enable callback
 116   void enable_mirroring(const Filesystem &filesystem, uint64_t local_pool_id,
 117                         Context *on_finish, bool is_restart=false);
 118   void handle_enable_mirroring(const Filesystem &filesystem, int r);
 119   void handle_enable_mirroring(const Filesystem &filesystem, const Peers &peers, int r);
 120
 121   // mirror disable callback
 122   void disable_mirroring(const Filesystem &filesystem, Context *on_finish);
 123   void handle_disable_mirroring(const Filesystem &filesystem, int r);
 124
 125   // peer update callback
 126   void add_peer(const Filesystem &filesystem, const Peer &peer);
 127   void remove_peer(const Filesystem &filesystem, const Peer &peer);
 128
 129   void schedule_mirror_update_task();
 130   void update_fs_mirrors();
 131
 132   void reopen_logs();
 133
 134   void _set_restarting(const Filesystem &filesystem) {
 135     auto &mirror_action = m_mirror_actions.at(filesystem);
 136     mirror_action.restarting = true;
 137   }
 138
 139   void _unset_restarting(const Filesystem &filesystem) {
 140     auto &mirror_action = m_mirror_actions.at(filesystem);
 141     mirror_action.restarting = false;
 142   }
 143
 144   bool _is_restarting(const Filesystem &filesystem) {
 145     auto &mirror_action = m_mirror_actions.at(filesystem);
 146     return mirror_action.restarting;
 147   }
 148 };
 149
 150 } // namespace mirror
 151 } // namespace cephfs
 152
 153 #endif // CEPHFS_MIRROR_H