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 "include/stringify.h"
   6 #include "common/ceph_json.h"
   7 #include "common/debug.h"
   8 #include "common/errno.h"
   9 #include "cls/rbd/cls_rbd_client.h"
  10 #include "librbd/internal.h"
  11 #include "librbd/api/Mirror.h"
  12 #include "tools/rbd_mirror/ServiceDaemon.h"
  13 #include "json_spirit/json_spirit.h"
  14
  15 #define dout_context g_ceph_context
  16 #define dout_subsys ceph_subsys_rbd_mirror
  17 #undef dout_prefix
  18 #define dout_prefix *_dout << "rbd::mirror::ClusterWatcher:" << this << " " \
  19                            << __func__ << ": "
  20
  21 using std::list;
  22 using std::map;
  23 using std::pair;
  24 using std::set;
  25 using std::string;
  26 using std::unique_ptr;
  27 using std::vector;
  28
  29 using librados::Rados;
  30 using librados::IoCtx;
  31
  32 namespace rbd {
  33 namespace mirror {
  34
  35 ClusterWatcher::ClusterWatcher(RadosRef cluster, ceph::mutex &lock,
  36                                ServiceDaemon<librbd::ImageCtx>* service_daemon)
  37   : m_cluster(cluster), m_lock(lock), m_service_daemon(service_daemon)
  38 {
  39 }
  40
  41 const ClusterWatcher::PoolPeers& ClusterWatcher::get_pool_peers() const
  42 {
  43   ceph_assert(ceph_mutex_is_locked(m_lock));
  44   return m_pool_peers;
  45 }
  46
  47 std::string ClusterWatcher::get_site_name() const {
  48   ceph_assert(ceph_mutex_is_locked(m_lock));
  49   return m_site_name;
  50 }
  51
  52 void ClusterWatcher::refresh_pools()
  53 {
  54   dout(20) << "enter" << dendl;
  55
  56   PoolPeers pool_peers;
  57   read_pool_peers(&pool_peers);
  58
  59   std::string site_name;
  60   int r = read_site_name(&site_name);
  61
  62   std::lock_guard l{m_lock};
  63   m_pool_peers = pool_peers;
  64
  65   if (r >= 0) {
  66     m_site_name = site_name;
  67   }
  68
  69   // TODO: perhaps use a workqueue instead, once we get notifications
  70   // about config changes for existing pools
  71 }
  72
  73 void ClusterWatcher::read_pool_peers(PoolPeers *pool_peers)
  74 {
  75   int r = m_cluster->wait_for_latest_osdmap();
  76   if (r < 0) {
  77     derr << "error waiting for OSD map: " << cpp_strerror(r) << dendl;
  78     return;
  79   }
  80
  81   list<pair<int64_t, string> > pools;
  82   r = m_cluster->pool_list2(pools);
  83   if (r < 0) {
  84     derr << "error listing pools: " << cpp_strerror(r) << dendl;
  85     return;
  86   }
  87
  88   std::set<int64_t> service_pool_ids;
  89   for (auto& kv : pools) {
  90     int64_t pool_id = kv.first;
  91     auto& pool_name = kv.second;
  92     int64_t base_tier;
  93     r = m_cluster->pool_get_base_tier(pool_id, &base_tier);
  94     if (r == -ENOENT) {
  95       dout(10) << "pool " << pool_name << " no longer exists" << dendl;
  96       continue;
  97     } else if (r < 0) {
  98       derr << "Error retrieving base tier for pool " << pool_name << dendl;
  99       continue;
 100     }
 101     if (pool_id != base_tier) {
 102       // pool is a cache; skip it
 103       continue;
 104     }
 105
 106     IoCtx ioctx;
 107     r = m_cluster->ioctx_create2(pool_id, ioctx);
 108     if (r == -ENOENT) {
 109       dout(10) << "pool " << pool_id << " no longer exists" << dendl;
 110       continue;
 111     } else if (r < 0) {
 112       derr << "Error accessing pool " << pool_name << cpp_strerror(r) << dendl;
 113       continue;
 114     }
 115
 116     cls::rbd::MirrorMode mirror_mode_internal;
 117     r = librbd::cls_client::mirror_mode_get(&ioctx, &mirror_mode_internal);
 118     if (r == 0 && mirror_mode_internal == cls::rbd::MIRROR_MODE_DISABLED) {
 119       dout(10) << "mirroring is disabled for pool " << pool_name << dendl;
 120       continue;
 121     }
 122
 123     service_pool_ids.insert(pool_id);
 124     if (m_service_pools.find(pool_id) == m_service_pools.end()) {
 125       m_service_pools[pool_id] = {};
 126       m_service_daemon->add_pool(pool_id, pool_name);
 127     }
 128
 129     if (r == -EPERM) {
 130       dout(10) << "access denied querying pool " << pool_name << dendl;
 131       m_service_pools[pool_id] = m_service_daemon->add_or_update_callout(
 132         pool_id, m_service_pools[pool_id],
 133         service_daemon::CALLOUT_LEVEL_WARNING, "access denied");
 134       continue;
 135     } else if (r < 0) {
 136       derr << "could not tell whether mirroring was enabled for " << pool_name
 137            << " : " << cpp_strerror(r) << dendl;
 138       m_service_pools[pool_id] = m_service_daemon->add_or_update_callout(
 139         pool_id, m_service_pools[pool_id],
 140         service_daemon::CALLOUT_LEVEL_WARNING, "mirroring mode query failed");
 141       continue;
 142     }
 143
 144     vector<librbd::mirror_peer_site_t> configs;
 145     r = librbd::api::Mirror<>::peer_site_list(ioctx, &configs);
 146     if (r < 0) {
 147       derr << "error reading mirroring config for pool " << pool_name
 148            << cpp_strerror(r) << dendl;
 149       m_service_pools[pool_id] = m_service_daemon->add_or_update_callout(
 150         pool_id, m_service_pools[pool_id],
 151         service_daemon::CALLOUT_LEVEL_ERROR, "mirroring peer list failed");
 152       continue;
 153     }
 154
 155     std::vector<PeerSpec> peers;
 156     peers.reserve(configs.size());
 157     for (auto& peer : configs) {
 158       if (peer.direction != RBD_MIRROR_PEER_DIRECTION_TX) {
 159         peers.push_back(peer);
 160       }
 161     }
 162
 163     for (auto& peer : peers) {
 164       r = resolve_peer_site_config_keys(pool_id, pool_name, &peer);
 165       if (r < 0) {
 166         break;
 167       }
 168     }
 169
 170     if (m_service_pools[pool_id] != service_daemon::CALLOUT_ID_NONE) {
 171       m_service_daemon->remove_callout(pool_id, m_service_pools[pool_id]);
 172       m_service_pools[pool_id] = service_daemon::CALLOUT_ID_NONE;
 173     }
 174
 175     pool_peers->emplace(pool_id, Peers{peers.begin(), peers.end()});
 176   }
 177
 178   for (auto it = m_service_pools.begin(); it != m_service_pools.end(); ) {
 179     auto current_it(it++);
 180     if (service_pool_ids.find(current_it->first) == service_pool_ids.end()) {
 181       m_service_daemon->remove_pool(current_it->first);
 182       m_service_pools.erase(current_it->first);
 183     }
 184   }
 185 }
 186
 187 int ClusterWatcher::read_site_name(std::string* site_name) {
 188   dout(10) << dendl;
 189
 190   librbd::RBD rbd;
 191   return rbd.mirror_site_name_get(*m_cluster, site_name);
 192 }
 193
 194 int ClusterWatcher::resolve_peer_site_config_keys(int64_t pool_id,
 195                                                   const std::string& pool_name,
 196                                                   PeerSpec* peer) {
 197   dout(10) << "retrieving config-key: pool_id=" << pool_id << ", "
 198            << "pool_name=" << pool_name << ", "
 199            << "peer_uuid=" << peer->uuid << dendl;
 200
 201   std::string cmd =
 202     "{"
 203       "\"prefix\": \"config-key get\", "
 204       "\"key\": \"" RBD_MIRROR_PEER_CONFIG_KEY_PREFIX + stringify(pool_id) +
 205         "/" + peer->uuid + "\""
 206     "}";
 207
 208   bufferlist in_bl;
 209   bufferlist out_bl;
 210   int r = m_cluster->mon_command(cmd, in_bl, &out_bl, nullptr);
 211   if (r == -ENOENT || out_bl.length() == 0) {
 212     return 0;
 213   } else if (r < 0) {
 214     derr << "error reading mirroring peer config for pool " << pool_name << ": "
 215          << cpp_strerror(r) << dendl;
 216     m_service_pools[pool_id] = m_service_daemon->add_or_update_callout(
 217       pool_id, m_service_pools[pool_id],
 218       service_daemon::CALLOUT_LEVEL_WARNING,
 219       "mirroring peer config-key query failed");
 220     return r;
 221   }
 222
 223   bool json_valid = false;
 224   json_spirit::mValue json_root;
 225   if(json_spirit::read(out_bl.to_str(), json_root)) {
 226     try {
 227       auto& json_obj = json_root.get_obj();
 228       if (json_obj.count("mon_host")) {
 229         peer->mon_host = json_obj["mon_host"].get_str();
 230       }
 231       if (json_obj.count("key")) {
 232         peer->key = json_obj["key"].get_str();
 233       }
 234       json_valid = true;
 235     } catch (std::runtime_error&) {
 236     }
 237   }
 238
 239   if (!json_valid) {
 240     derr << "error parsing mirroring peer config for pool " << pool_name << ", "
 241          << "peer " << peer->uuid << dendl;
 242     m_service_pools[pool_id] = m_service_daemon->add_or_update_callout(
 243       pool_id, m_service_pools[pool_id],
 244       service_daemon::CALLOUT_LEVEL_WARNING,
 245       "mirroring peer config-key decode failed");
 246   }
 247
 248   return 0;
 249 }
 250
 251 } // namespace mirror
 252 } // namespace rbd