src/ceph/src/tools/rbd_mirror/ClusterWatcher.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3
   4 #include "ClusterWatcher.h"
   5 #include "common/debug.h"
   6 #include "common/errno.h"
   7 #include "cls/rbd/cls_rbd_client.h"
   8 #include "librbd/internal.h"
   9 #include "librbd/api/Mirror.h"
  10 #include "tools/rbd_mirror/ServiceDaemon.h"
  11
  12 #define dout_context g_ceph_context
  13 #define dout_subsys ceph_subsys_rbd_mirror
  14 #undef dout_prefix
  15 #define dout_prefix *_dout << "rbd::mirror::ClusterWatcher:" << this << " " \
  16                            << __func__ << ": "
  17
  18 using std::list;
  19 using std::map;
  20 using std::set;
  21 using std::string;
  22 using std::unique_ptr;
  23 using std::vector;
  24
  25 using librados::Rados;
  26 using librados::IoCtx;
  27
  28 namespace rbd {
  29 namespace mirror {
  30
  31 ClusterWatcher::ClusterWatcher(RadosRef cluster, Mutex &lock,
  32                                ServiceDaemon<librbd::ImageCtx>* service_daemon)
  33   : m_cluster(cluster), m_lock(lock), m_service_daemon(service_daemon)
  34 {
  35 }
  36
  37 const ClusterWatcher::PoolPeers& ClusterWatcher::get_pool_peers() const
  38 {
  39   assert(m_lock.is_locked());
  40   return m_pool_peers;
  41 }
  42
  43 void ClusterWatcher::refresh_pools()
  44 {
  45   dout(20) << "enter" << dendl;
  46
  47   PoolPeers pool_peers;
  48   PoolNames pool_names;
  49   read_pool_peers(&pool_peers, &pool_names);
  50
  51   Mutex::Locker l(m_lock);
  52   m_pool_peers = pool_peers;
  53   // TODO: perhaps use a workqueue instead, once we get notifications
  54   // about config changes for existing pools
  55 }
  56
  57 void ClusterWatcher::read_pool_peers(PoolPeers *pool_peers,
  58                                      PoolNames *pool_names)
  59 {
  60   list<pair<int64_t, string> > pools;
  61   int r = m_cluster->pool_list2(pools);
  62   if (r < 0) {
  63     derr << "error listing pools: " << cpp_strerror(r) << dendl;
  64     return;
  65   }
  66
  67   std::set<int64_t> service_pool_ids;
  68   for (auto& kv : pools) {
  69     int64_t pool_id = kv.first;
  70     auto& pool_name = kv.second;
  71     int64_t base_tier;
  72     r = m_cluster->pool_get_base_tier(pool_id, &base_tier);
  73     if (r == -ENOENT) {
  74       dout(10) << "pool " << pool_name << " no longer exists" << dendl;
  75       continue;
  76     } else if (r < 0) {
  77       derr << "Error retrieving base tier for pool " << pool_name << dendl;
  78       continue;
  79     }
  80     if (pool_id != base_tier) {
  81       // pool is a cache; skip it
  82       continue;
  83     }
  84
  85     IoCtx ioctx;
  86     r = m_cluster->ioctx_create2(pool_id, ioctx);
  87     if (r == -ENOENT) {
  88       dout(10) << "pool " << pool_id << " no longer exists" << dendl;
  89       continue;
  90     } else if (r < 0) {
  91       derr << "Error accessing pool " << pool_name << cpp_strerror(r) << dendl;
  92       continue;
  93     }
  94
  95     cls::rbd::MirrorMode mirror_mode_internal;
  96     r = librbd::cls_client::mirror_mode_get(&ioctx, &mirror_mode_internal);
  97     if (r == 0 && mirror_mode_internal == cls::rbd::MIRROR_MODE_DISABLED) {
  98       dout(10) << "mirroring is disabled for pool " << pool_name << dendl;
  99       continue;
 100     }
 101
 102     service_pool_ids.insert(pool_id);
 103     if (m_service_pools.find(pool_id) == m_service_pools.end()) {
 104       m_service_pools[pool_id] = {};
 105       m_service_daemon->add_pool(pool_id, pool_name);
 106     }
 107
 108     if (r == -EPERM) {
 109       dout(10) << "access denied querying pool " << pool_name << dendl;
 110       m_service_pools[pool_id] = m_service_daemon->add_or_update_callout(
 111         pool_id, m_service_pools[pool_id],
 112         service_daemon::CALLOUT_LEVEL_WARNING, "access denied");
 113       continue;
 114     } else if (r < 0) {
 115       derr << "could not tell whether mirroring was enabled for " << pool_name
 116            << " : " << cpp_strerror(r) << dendl;
 117       m_service_pools[pool_id] = m_service_daemon->add_or_update_callout(
 118         pool_id, m_service_pools[pool_id],
 119         service_daemon::CALLOUT_LEVEL_WARNING, "mirroring mode query failed");
 120       continue;
 121     }
 122
 123     vector<librbd::mirror_peer_t> configs;
 124     r = librbd::api::Mirror<>::peer_list(ioctx, &configs);
 125     if (r < 0) {
 126       derr << "error reading mirroring config for pool " << pool_name
 127            << cpp_strerror(r) << dendl;
 128       m_service_pools[pool_id] = m_service_daemon->add_or_update_callout(
 129         pool_id, m_service_pools[pool_id],
 130         service_daemon::CALLOUT_LEVEL_ERROR, "mirroring peer list failed");
 131       continue;
 132     }
 133
 134     if (m_service_pools[pool_id] != service_daemon::CALLOUT_ID_NONE) {
 135       m_service_daemon->remove_callout(pool_id, m_service_pools[pool_id]);
 136       m_service_pools[pool_id] = service_daemon::CALLOUT_ID_NONE;
 137     }
 138
 139     pool_peers->insert({pool_id, Peers{configs.begin(), configs.end()}});
 140     pool_names->insert(pool_name);
 141   }
 142
 143   for (auto it = m_service_pools.begin(); it != m_service_pools.end(); ) {
 144     auto current_it(it++);
 145     if (service_pool_ids.find(current_it->first) == service_pool_ids.end()) {
 146       m_service_daemon->remove_pool(current_it->first);
 147       m_service_pools.erase(current_it->first);
 148     }
 149   }
 150 }
 151
 152 } // namespace mirror
 153 } // namespace rbd