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   int r = m_cluster->wait_for_latest_osdmap();
  61   if (r < 0) {
  62     derr << "error waiting for OSD map: " << cpp_strerror(r) << dendl;
  63     return;
  64   }
  65
  66   list<pair<int64_t, string> > pools;
  67   r = m_cluster->pool_list2(pools);
  68   if (r < 0) {
  69     derr << "error listing pools: " << cpp_strerror(r) << dendl;
  70     return;
  71   }
  72
  73   std::set<int64_t> service_pool_ids;
  74   for (auto& kv : pools) {
  75     int64_t pool_id = kv.first;
  76     auto& pool_name = kv.second;
  77     int64_t base_tier;
  78     r = m_cluster->pool_get_base_tier(pool_id, &base_tier);
  79     if (r == -ENOENT) {
  80       dout(10) << "pool " << pool_name << " no longer exists" << dendl;
  81       continue;
  82     } else if (r < 0) {
  83       derr << "Error retrieving base tier for pool " << pool_name << dendl;
  84       continue;
  85     }
  86     if (pool_id != base_tier) {
  87       // pool is a cache; skip it
  88       continue;
  89     }
  90
  91     IoCtx ioctx;
  92     r = m_cluster->ioctx_create2(pool_id, ioctx);
  93     if (r == -ENOENT) {
  94       dout(10) << "pool " << pool_id << " no longer exists" << dendl;
  95       continue;
  96     } else if (r < 0) {
  97       derr << "Error accessing pool " << pool_name << cpp_strerror(r) << dendl;
  98       continue;
  99     }
 100
 101     cls::rbd::MirrorMode mirror_mode_internal;
 102     r = librbd::cls_client::mirror_mode_get(&ioctx, &mirror_mode_internal);
 103     if (r == 0 && mirror_mode_internal == cls::rbd::MIRROR_MODE_DISABLED) {
 104       dout(10) << "mirroring is disabled for pool " << pool_name << dendl;
 105       continue;
 106     }
 107
 108     service_pool_ids.insert(pool_id);
 109     if (m_service_pools.find(pool_id) == m_service_pools.end()) {
 110       m_service_pools[pool_id] = {};
 111       m_service_daemon->add_pool(pool_id, pool_name);
 112     }
 113
 114     if (r == -EPERM) {
 115       dout(10) << "access denied querying pool " << pool_name << dendl;
 116       m_service_pools[pool_id] = m_service_daemon->add_or_update_callout(
 117         pool_id, m_service_pools[pool_id],
 118         service_daemon::CALLOUT_LEVEL_WARNING, "access denied");
 119       continue;
 120     } else if (r < 0) {
 121       derr << "could not tell whether mirroring was enabled for " << pool_name
 122            << " : " << cpp_strerror(r) << dendl;
 123       m_service_pools[pool_id] = m_service_daemon->add_or_update_callout(
 124         pool_id, m_service_pools[pool_id],
 125         service_daemon::CALLOUT_LEVEL_WARNING, "mirroring mode query failed");
 126       continue;
 127     }
 128
 129     vector<librbd::mirror_peer_t> configs;
 130     r = librbd::api::Mirror<>::peer_list(ioctx, &configs);
 131     if (r < 0) {
 132       derr << "error reading mirroring config for pool " << pool_name
 133            << cpp_strerror(r) << dendl;
 134       m_service_pools[pool_id] = m_service_daemon->add_or_update_callout(
 135         pool_id, m_service_pools[pool_id],
 136         service_daemon::CALLOUT_LEVEL_ERROR, "mirroring peer list failed");
 137       continue;
 138     }
 139
 140     if (m_service_pools[pool_id] != service_daemon::CALLOUT_ID_NONE) {
 141       m_service_daemon->remove_callout(pool_id, m_service_pools[pool_id]);
 142       m_service_pools[pool_id] = service_daemon::CALLOUT_ID_NONE;
 143     }
 144
 145     pool_peers->insert({pool_id, Peers{configs.begin(), configs.end()}});
 146     pool_names->insert(pool_name);
 147   }
 148
 149   for (auto it = m_service_pools.begin(); it != m_service_pools.end(); ) {
 150     auto current_it(it++);
 151     if (service_pool_ids.find(current_it->first) == service_pool_ids.end()) {
 152       m_service_daemon->remove_pool(current_it->first);
 153       m_service_pools.erase(current_it->first);
 154     }
 155   }
 156 }
 157
 158 } // namespace mirror
 159 } // namespace rbd