ceph/src/librbd/api/Mirror.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 "librbd/api/Mirror.h"
   5 #include "include/rados/librados.hpp"
   6 #include "include/stringify.h"
   7 #include "common/ceph_json.h"
   8 #include "common/dout.h"
   9 #include "common/errno.h"
  10 #include "cls/rbd/cls_rbd_client.h"
  11 #include "librbd/ImageCtx.h"
  12 #include "librbd/ImageState.h"
  13 #include "librbd/Journal.h"
  14 #include "librbd/MirroringWatcher.h"
  15 #include "librbd/Operations.h"
  16 #include "librbd/Utils.h"
  17 #include "librbd/api/Image.h"
  18 #include "librbd/api/Namespace.h"
  19 #include "librbd/mirror/DemoteRequest.h"
  20 #include "librbd/mirror/DisableRequest.h"
  21 #include "librbd/mirror/EnableRequest.h"
  22 #include "librbd/mirror/GetInfoRequest.h"
  23 #include "librbd/mirror/GetStatusRequest.h"
  24 #include "librbd/mirror/GetUuidRequest.h"
  25 #include "librbd/mirror/PromoteRequest.h"
  26 #include "librbd/mirror/Types.h"
  27 #include "librbd/MirroringWatcher.h"
  28 #include "librbd/mirror/snapshot/CreatePrimaryRequest.h"
  29 #include "librbd/mirror/snapshot/ImageMeta.h"
  30 #include "librbd/mirror/snapshot/UnlinkPeerRequest.h"
  31 #include "librbd/mirror/snapshot/Utils.h"
  32 #include <boost/algorithm/string/trim.hpp>
  33 #include <boost/algorithm/string/replace.hpp>
  34 #include <boost/scope_exit.hpp>
  35 #include "json_spirit/json_spirit.h"
  36
  37 #include <algorithm>
  38 #include <bitset>
  39
  40 #define dout_subsys ceph_subsys_rbd
  41 #undef dout_prefix
  42 #define dout_prefix *_dout << "librbd::api::Mirror: " << __func__ << ": "
  43
  44 namespace librbd {
  45 namespace api {
  46
  47 namespace {
  48
  49 int get_config_key(librados::Rados& rados, const std::string& key,
  50                    std::string* value) {
  51   std::string cmd =
  52     "{"
  53       "\"prefix\": \"config-key get\", "
  54       "\"key\": \"" + key + "\""
  55     "}";
  56
  57   bufferlist in_bl;
  58   bufferlist out_bl;
  59
  60   int r = rados.mon_command(cmd, in_bl, &out_bl, nullptr);
  61   if (r == -EINVAL) {
  62     return -EOPNOTSUPP;
  63   } else if (r < 0 && r != -ENOENT) {
  64     return r;
  65   }
  66
  67   *value = out_bl.to_str();
  68   return 0;
  69 }
  70
  71 int set_config_key(librados::Rados& rados, const std::string& key,
  72                    const std::string& value) {
  73   std::string cmd;
  74   if (value.empty()) {
  75     cmd = "{"
  76             "\"prefix\": \"config-key rm\", "
  77             "\"key\": \"" + key + "\""
  78           "}";
  79   } else {
  80     cmd = "{"
  81             "\"prefix\": \"config-key set\", "
  82             "\"key\": \"" + key + "\", "
  83             "\"val\": \"" + value + "\""
  84           "}";
  85   }
  86   bufferlist in_bl;
  87   bufferlist out_bl;
  88
  89   int r = rados.mon_command(cmd, in_bl, &out_bl, nullptr);
  90   if (r == -EINVAL) {
  91     return -EOPNOTSUPP;
  92   } else if (r < 0) {
  93     return r;
  94   }
  95
  96   return 0;
  97 }
  98
  99 std::string get_peer_config_key_name(int64_t pool_id,
 100                                      const std::string& peer_uuid) {
 101   return RBD_MIRROR_PEER_CONFIG_KEY_PREFIX + stringify(pool_id) + "/" +
 102            peer_uuid;
 103 }
 104
 105 int remove_peer_config_key(librados::IoCtx& io_ctx,
 106                            const std::string& peer_uuid) {
 107   int64_t pool_id = io_ctx.get_id();
 108   auto key = get_peer_config_key_name(pool_id, peer_uuid);
 109
 110   librados::Rados rados(io_ctx);
 111   int r = set_config_key(rados, key, "");
 112   if (r < 0 && r != -ENOENT && r != -EPERM) {
 113     return r;
 114   }
 115   return 0;
 116 }
 117
 118 int create_bootstrap_user(CephContext* cct, librados::Rados& rados,
 119                           std::string* peer_client_id, std::string* cephx_key) {
 120   ldout(cct, 20) << dendl;
 121
 122   // retrieve peer CephX user from config-key
 123   int r = get_config_key(rados, RBD_MIRROR_PEER_CLIENT_ID_CONFIG_KEY,
 124                          peer_client_id);
 125   if (r == -EACCES) {
 126       ldout(cct, 5) << "insufficient permissions to get peer-client-id "
 127                     << "config-key" << dendl;
 128       return r;
 129   } else if (r < 0 && r != -ENOENT) {
 130     lderr(cct) << "failed to retrieve peer client id key: "
 131                << cpp_strerror(r) << dendl;
 132     return r;
 133   } else if (r == -ENOENT || peer_client_id->empty()) {
 134     ldout(cct, 20) << "creating new peer-client-id config-key" << dendl;
 135
 136     *peer_client_id = "rbd-mirror-peer";
 137     r = set_config_key(rados, RBD_MIRROR_PEER_CLIENT_ID_CONFIG_KEY,
 138                        *peer_client_id);
 139     if (r == -EACCES) {
 140       ldout(cct, 5) << "insufficient permissions to update peer-client-id "
 141                     << "config-key" << dendl;
 142       return r;
 143     } else if (r < 0) {
 144       lderr(cct) << "failed to update peer client id key: "
 145                  << cpp_strerror(r) << dendl;
 146       return r;
 147     }
 148   }
 149   ldout(cct, 20) << "peer_client_id=" << *peer_client_id << dendl;
 150
 151   // create peer client user
 152   std::string cmd =
 153     R"({)" \
 154     R"(  "prefix": "auth get-or-create",)" \
 155     R"(  "entity": "client.)" + *peer_client_id + R"(",)" \
 156     R"(  "caps": [)" \
 157     R"(    "mon", "profile rbd-mirror-peer",)" \
 158     R"(    "osd", "profile rbd"],)" \
 159     R"(  "format": "json")" \
 160     R"(})";
 161
 162   bufferlist in_bl;
 163   bufferlist out_bl;
 164
 165   r = rados.mon_command(cmd, in_bl, &out_bl, nullptr);
 166   if (r == -EINVAL) {
 167     ldout(cct, 5) << "caps mismatch for existing user" << dendl;
 168     return -EEXIST;
 169   } else if (r == -EACCES) {
 170     ldout(cct, 5) << "insufficient permissions to create user" << dendl;
 171     return r;
 172   } else if (r < 0) {
 173     lderr(cct) << "failed to create or update RBD mirroring bootstrap user: "
 174                << cpp_strerror(r) << dendl;
 175     return r;
 176   }
 177
 178   // extract key from response
 179   bool json_valid = false;
 180   json_spirit::mValue json_root;
 181   if(json_spirit::read(out_bl.to_str(), json_root)) {
 182     try {
 183       auto& json_obj = json_root.get_array()[0].get_obj();
 184       *cephx_key = json_obj["key"].get_str();
 185       json_valid = true;
 186     } catch (std::runtime_error&) {
 187     }
 188   }
 189
 190   if (!json_valid) {
 191     lderr(cct) << "invalid auth keyring JSON received" << dendl;
 192     return -EBADMSG;
 193   }
 194
 195   return 0;
 196 }
 197
 198 int create_bootstrap_peer(CephContext* cct, librados::IoCtx& io_ctx,
 199                           mirror_peer_direction_t direction,
 200                           const std::string& site_name, const std::string& fsid,
 201                           const std::string& client_id, const std::string& key,
 202                           const std::string& mon_host,
 203                           const std::string& cluster1,
 204                           const std::string& cluster2) {
 205   ldout(cct, 20) << dendl;
 206
 207   std::string peer_uuid;
 208   std::vector<mirror_peer_site_t> peers;
 209   int r = Mirror<>::peer_site_list(io_ctx, &peers);
 210   if (r < 0 && r != -ENOENT) {
 211     lderr(cct) << "failed to list mirror peers: " << cpp_strerror(r) << dendl;
 212     return r;
 213   }
 214
 215   if (peers.empty()) {
 216     r = Mirror<>::peer_site_add(io_ctx, &peer_uuid, direction, site_name,
 217                                 "client." + client_id);
 218     if (r < 0) {
 219       lderr(cct) << "failed to add " << cluster1 << " peer to "
 220                  << cluster2 << " " << "cluster: " << cpp_strerror(r) << dendl;
 221       return r;
 222     }
 223   } else if (peers[0].site_name != site_name &&
 224              peers[0].site_name != fsid) {
 225     // only support a single peer
 226     lderr(cct) << "multiple peers are not currently supported" << dendl;
 227     return -EINVAL;
 228   } else {
 229     peer_uuid = peers[0].uuid;
 230
 231     if (peers[0].site_name != site_name) {
 232       r = Mirror<>::peer_site_set_name(io_ctx, peer_uuid, site_name);
 233       if (r < 0) {
 234         // non-fatal attempt to update site name
 235         lderr(cct) << "failed to update peer site name" << dendl;
 236       }
 237     }
 238   }
 239
 240   Mirror<>::Attributes attributes {
 241     {"mon_host", mon_host},
 242     {"key", key}};
 243   r = Mirror<>::peer_site_set_attributes(io_ctx, peer_uuid, attributes);
 244   if (r < 0) {
 245     lderr(cct) << "failed to update " << cluster1 << " cluster connection "
 246                << "attributes in " << cluster2 << " cluster: "
 247                << cpp_strerror(r) << dendl;
 248     return r;
 249   }
 250
 251   return 0;
 252 }
 253
 254 int list_mirror_images(librados::IoCtx& io_ctx,
 255                        std::set<std::string>& mirror_image_ids) {
 256   CephContext *cct = reinterpret_cast<CephContext *>(io_ctx.cct());
 257
 258   std::string last_read = "";
 259   int max_read = 1024;
 260   int r;
 261   do {
 262     std::map<std::string, std::string> mirror_images;
 263     r =  cls_client::mirror_image_list(&io_ctx, last_read, max_read,
 264                                        &mirror_images);
 265     if (r < 0 && r != -ENOENT) {
 266       lderr(cct) << "error listing mirrored image directory: "
 267                  << cpp_strerror(r) << dendl;
 268       return r;
 269     }
 270     for (auto it = mirror_images.begin(); it != mirror_images.end(); ++it) {
 271       mirror_image_ids.insert(it->first);
 272     }
 273     if (!mirror_images.empty()) {
 274       last_read = mirror_images.rbegin()->first;
 275     }
 276     r = mirror_images.size();
 277   } while (r == max_read);
 278
 279   return 0;
 280 }
 281
 282 struct C_ImageGetInfo : public Context {
 283   mirror_image_info_t *mirror_image_info;
 284   mirror_image_mode_t *mirror_image_mode;
 285   Context *on_finish;
 286
 287   cls::rbd::MirrorImage mirror_image;
 288   mirror::PromotionState promotion_state = mirror::PROMOTION_STATE_PRIMARY;
 289   std::string primary_mirror_uuid;
 290
 291   C_ImageGetInfo(mirror_image_info_t *mirror_image_info,
 292                  mirror_image_mode_t *mirror_image_mode,  Context *on_finish)
 293     : mirror_image_info(mirror_image_info),
 294       mirror_image_mode(mirror_image_mode), on_finish(on_finish) {
 295   }
 296
 297   void finish(int r) override {
 298     if (r < 0 && r != -ENOENT) {
 299       on_finish->complete(r);
 300       return;
 301     }
 302
 303     if (mirror_image_info != nullptr) {
 304       mirror_image_info->global_id = mirror_image.global_image_id;
 305       mirror_image_info->state = static_cast<rbd_mirror_image_state_t>(
 306         mirror_image.state);
 307       mirror_image_info->primary = (
 308         promotion_state == mirror::PROMOTION_STATE_PRIMARY);
 309     }
 310
 311     if (mirror_image_mode != nullptr) {
 312       *mirror_image_mode =
 313         static_cast<rbd_mirror_image_mode_t>(mirror_image.mode);
 314     }
 315
 316     on_finish->complete(0);
 317   }
 318 };
 319
 320 struct C_ImageGetGlobalStatus : public C_ImageGetInfo {
 321   std::string image_name;
 322   mirror_image_global_status_t *mirror_image_global_status;
 323
 324   cls::rbd::MirrorImageStatus mirror_image_status_internal;
 325
 326   C_ImageGetGlobalStatus(
 327       const std::string &image_name,
 328       mirror_image_global_status_t *mirror_image_global_status,
 329       Context *on_finish)
 330     : C_ImageGetInfo(&mirror_image_global_status->info, nullptr, on_finish),
 331       image_name(image_name),
 332       mirror_image_global_status(mirror_image_global_status) {
 333   }
 334
 335   void finish(int r) override {
 336     if (r < 0 && r != -ENOENT) {
 337       on_finish->complete(r);
 338       return;
 339     }
 340
 341     mirror_image_global_status->name = image_name;
 342     mirror_image_global_status->site_statuses.clear();
 343     mirror_image_global_status->site_statuses.reserve(
 344       mirror_image_status_internal.mirror_image_site_statuses.size());
 345     for (auto& site_status :
 346            mirror_image_status_internal.mirror_image_site_statuses) {
 347       mirror_image_global_status->site_statuses.push_back({
 348         site_status.mirror_uuid,
 349         static_cast<mirror_image_status_state_t>(site_status.state),
 350         site_status.description, site_status.last_update.sec(),
 351         site_status.up});
 352     }
 353     C_ImageGetInfo::finish(0);
 354   }
 355 };
 356
 357 template <typename I>
 358 struct C_ImageSnapshotCreate : public Context {
 359   I *ictx;
 360   uint64_t *snap_id;
 361   Context *on_finish;
 362
 363   cls::rbd::MirrorImage mirror_image;
 364   mirror::PromotionState promotion_state;
 365   std::string primary_mirror_uuid;
 366
 367   C_ImageSnapshotCreate(I *ictx, uint64_t *snap_id, Context *on_finish)
 368     : ictx(ictx), snap_id(snap_id), on_finish(on_finish) {
 369   }
 370
 371   void finish(int r) override {
 372     if (r < 0 && r != -ENOENT) {
 373       on_finish->complete(r);
 374       return;
 375     }
 376
 377     if (mirror_image.mode != cls::rbd::MIRROR_IMAGE_MODE_SNAPSHOT ||
 378         mirror_image.state != cls::rbd::MIRROR_IMAGE_STATE_ENABLED) {
 379       lderr(ictx->cct) << "snapshot based mirroring is not enabled" << dendl;
 380       on_finish->complete(-EINVAL);
 381       return;
 382     }
 383
 384     auto req = mirror::snapshot::CreatePrimaryRequest<I>::create(
 385       ictx, mirror_image.global_image_id, CEPH_NOSNAP, 0U, snap_id, on_finish);
 386     req->send();
 387   }
 388 };
 389
 390 } // anonymous namespace
 391
 392 template <typename I>
 393 int Mirror<I>::image_enable(I *ictx, mirror_image_mode_t mode,
 394                             bool relax_same_pool_parent_check) {
 395   CephContext *cct = ictx->cct;
 396   ldout(cct, 20) << "ictx=" << ictx << " mode=" << mode
 397                  << " relax_same_pool_parent_check="
 398                  << relax_same_pool_parent_check <<  dendl;
 399
 400   int r = ictx->state->refresh_if_required();
 401   if (r < 0) {
 402     return r;
 403   }
 404
 405   cls::rbd::MirrorMode mirror_mode;
 406   r = cls_client::mirror_mode_get(&ictx->md_ctx, &mirror_mode);
 407   if (r < 0) {
 408     lderr(cct) << "cannot enable mirroring: failed to retrieve mirror mode: "
 409                << cpp_strerror(r) << dendl;
 410     return r;
 411   }
 412
 413   if (mirror_mode != cls::rbd::MIRROR_MODE_IMAGE) {
 414     lderr(cct) << "cannot enable mirroring in the current pool mirroring mode"
 415                << dendl;
 416     return -EINVAL;
 417   }
 418
 419   // is mirroring not enabled for the parent?
 420   {
 421     std::shared_lock image_locker{ictx->image_lock};
 422     ImageCtx *parent = ictx->parent;
 423     if (parent) {
 424       if (parent->md_ctx.get_id() != ictx->md_ctx.get_id() ||
 425           !relax_same_pool_parent_check) {
 426         cls::rbd::MirrorImage mirror_image_internal;
 427         r = cls_client::mirror_image_get(&(parent->md_ctx), parent->id,
 428                                          &mirror_image_internal);
 429         if (r == -ENOENT) {
 430           lderr(cct) << "mirroring is not enabled for the parent" << dendl;
 431           return -EINVAL;
 432         }
 433       }
 434     }
 435   }
 436
 437   if (mode == RBD_MIRROR_IMAGE_MODE_JOURNAL &&
 438       !ictx->test_features(RBD_FEATURE_JOURNALING)) {
 439     uint64_t features = RBD_FEATURE_JOURNALING;
 440     if (!ictx->test_features(RBD_FEATURE_EXCLUSIVE_LOCK)) {
 441       features |= RBD_FEATURE_EXCLUSIVE_LOCK;
 442     }
 443     r = ictx->operations->update_features(features, true);
 444     if (r < 0) {
 445       lderr(cct) << "cannot enable journaling: " << cpp_strerror(r) << dendl;
 446       return r;
 447     }
 448   }
 449
 450   C_SaferCond ctx;
 451   auto req = mirror::EnableRequest<ImageCtx>::create(
 452     ictx, static_cast<cls::rbd::MirrorImageMode>(mode), "", false, &ctx);
 453   req->send();
 454
 455   r = ctx.wait();
 456   if (r < 0) {
 457     lderr(cct) << "cannot enable mirroring: " << cpp_strerror(r) << dendl;
 458     return r;
 459   }
 460
 461   return 0;
 462 }
 463
 464 template <typename I>
 465 int Mirror<I>::image_disable(I *ictx, bool force) {
 466   CephContext *cct = ictx->cct;
 467   ldout(cct, 20) << "ictx=" << ictx << dendl;
 468
 469   int r = ictx->state->refresh_if_required();
 470   if (r < 0) {
 471     return r;
 472   }
 473
 474   cls::rbd::MirrorMode mirror_mode;
 475   r = cls_client::mirror_mode_get(&ictx->md_ctx, &mirror_mode);
 476   if (r < 0) {
 477     lderr(cct) << "cannot disable mirroring: failed to retrieve pool "
 478       "mirroring mode: " << cpp_strerror(r) << dendl;
 479     return r;
 480   }
 481
 482   if (mirror_mode != cls::rbd::MIRROR_MODE_IMAGE) {
 483     lderr(cct) << "cannot disable mirroring in the current pool mirroring "
 484       "mode" << dendl;
 485     return -EINVAL;
 486   }
 487
 488   // is mirroring  enabled for the image?
 489   cls::rbd::MirrorImage mirror_image_internal;
 490   r = cls_client::mirror_image_get(&ictx->md_ctx, ictx->id,
 491                                    &mirror_image_internal);
 492   if (r == -ENOENT) {
 493     // mirroring is not enabled for this image
 494     ldout(cct, 20) << "ignoring disable command: mirroring is not enabled for "
 495                    << "this image" << dendl;
 496     return 0;
 497   } else if (r == -EOPNOTSUPP) {
 498     ldout(cct, 5) << "mirroring not supported by OSD" << dendl;
 499     return r;
 500   } else if (r < 0) {
 501     lderr(cct) << "failed to retrieve mirror image metadata: "
 502                << cpp_strerror(r) << dendl;
 503     return r;
 504   }
 505
 506   mirror_image_internal.state = cls::rbd::MIRROR_IMAGE_STATE_DISABLING;
 507   r = cls_client::mirror_image_set(&ictx->md_ctx, ictx->id,
 508                                    mirror_image_internal);
 509   if (r < 0) {
 510     lderr(cct) << "cannot disable mirroring: " << cpp_strerror(r) << dendl;
 511     return r;
 512   }
 513
 514   bool rollback = false;
 515   BOOST_SCOPE_EXIT_ALL(ictx, &mirror_image_internal, &rollback) {
 516     if (rollback) {
 517       // restore the mask bit for treating the non-primary feature as read-only
 518       ictx->image_lock.lock();
 519       ictx->read_only_mask |= IMAGE_READ_ONLY_FLAG_NON_PRIMARY;
 520       ictx->image_lock.unlock();
 521
 522       ictx->state->handle_update_notification();
 523
 524       // attempt to restore the image state
 525       CephContext *cct = ictx->cct;
 526       mirror_image_internal.state = cls::rbd::MIRROR_IMAGE_STATE_ENABLED;
 527       int r = cls_client::mirror_image_set(&ictx->md_ctx, ictx->id,
 528                                            mirror_image_internal);
 529       if (r < 0) {
 530         lderr(cct) << "failed to re-enable image mirroring: "
 531                    << cpp_strerror(r) << dendl;
 532       }
 533     }
 534   };
 535
 536   std::unique_lock image_locker{ictx->image_lock};
 537   map<librados::snap_t, SnapInfo> snap_info = ictx->snap_info;
 538   for (auto &info : snap_info) {
 539     cls::rbd::ParentImageSpec parent_spec{ictx->md_ctx.get_id(),
 540                                           ictx->md_ctx.get_namespace(),
 541                                           ictx->id, info.first};
 542     std::vector<librbd::linked_image_spec_t> child_images;
 543     r = Image<I>::list_children(ictx, parent_spec, &child_images);
 544     if (r < 0) {
 545       rollback = true;
 546       return r;
 547     }
 548
 549     if (child_images.empty()) {
 550       continue;
 551     }
 552
 553     librados::IoCtx child_io_ctx;
 554     int64_t child_pool_id = -1;
 555     for (auto &child_image : child_images){
 556       std::string pool = child_image.pool_name;
 557       if (child_pool_id == -1 ||
 558           child_pool_id != child_image.pool_id ||
 559           child_io_ctx.get_namespace() != child_image.pool_namespace) {
 560         r = util::create_ioctx(ictx->md_ctx, "child image",
 561                                child_image.pool_id,
 562                                child_image.pool_namespace,
 563                                &child_io_ctx);
 564         if (r < 0) {
 565           rollback = true;
 566           return r;
 567         }
 568
 569         child_pool_id = child_image.pool_id;
 570       }
 571
 572       cls::rbd::MirrorImage child_mirror_image_internal;
 573       r = cls_client::mirror_image_get(&child_io_ctx, child_image.image_id,
 574                                        &child_mirror_image_internal);
 575       if (r != -ENOENT) {
 576         rollback = true;
 577         lderr(cct) << "mirroring is enabled on one or more children "
 578                    << dendl;
 579         return -EBUSY;
 580       }
 581     }
 582   }
 583   image_locker.unlock();
 584
 585   if (mirror_image_internal.mode == cls::rbd::MIRROR_IMAGE_MODE_SNAPSHOT) {
 586     // don't let the non-primary feature bit prevent image updates
 587     ictx->image_lock.lock();
 588     ictx->read_only_mask &= ~IMAGE_READ_ONLY_FLAG_NON_PRIMARY;
 589     ictx->image_lock.unlock();
 590
 591     r = ictx->state->refresh();
 592     if (r < 0) {
 593       rollback = true;
 594       return r;
 595     }
 596
 597     // remove any snapshot-based mirroring image-meta from image
 598     std::string mirror_uuid;
 599     r = uuid_get(ictx->md_ctx, &mirror_uuid);
 600     if (r < 0) {
 601       rollback = true;
 602       return r;
 603     }
 604
 605     r = ictx->operations->metadata_remove(
 606       mirror::snapshot::util::get_image_meta_key(mirror_uuid));
 607     if (r < 0 && r != -ENOENT) {
 608       lderr(cct) << "cannot remove snapshot image-meta key: " << cpp_strerror(r)
 609                  << dendl;
 610       rollback = true;
 611       return r;
 612     }
 613   }
 614
 615   C_SaferCond ctx;
 616   auto req = mirror::DisableRequest<ImageCtx>::create(ictx, force, true,
 617                                                       &ctx);
 618   req->send();
 619
 620   r = ctx.wait();
 621   if (r < 0) {
 622     lderr(cct) << "cannot disable mirroring: " << cpp_strerror(r) << dendl;
 623     rollback = true;
 624     return r;
 625   }
 626
 627   if (mirror_image_internal.mode == cls::rbd::MIRROR_IMAGE_MODE_JOURNAL) {
 628     r = ictx->operations->update_features(RBD_FEATURE_JOURNALING, false);
 629     if (r < 0) {
 630       lderr(cct) << "cannot disable journaling: " << cpp_strerror(r) << dendl;
 631       // not fatal
 632     }
 633   }
 634
 635   return 0;
 636 }
 637
 638 template <typename I>
 639 int Mirror<I>::image_promote(I *ictx, bool force) {
 640   CephContext *cct = ictx->cct;
 641
 642   C_SaferCond ctx;
 643   Mirror<I>::image_promote(ictx, force, &ctx);
 644   int r = ctx.wait();
 645   if (r < 0) {
 646     lderr(cct) << "failed to promote image" << dendl;
 647     return r;
 648   }
 649
 650   return 0;
 651 }
 652
 653 template <typename I>
 654 void Mirror<I>::image_promote(I *ictx, bool force, Context *on_finish) {
 655   CephContext *cct = ictx->cct;
 656   ldout(cct, 20) << "ictx=" << ictx << ", "
 657                  << "force=" << force << dendl;
 658
 659   // don't let the non-primary feature bit prevent image updates
 660   ictx->image_lock.lock();
 661   ictx->read_only_mask &= ~IMAGE_READ_ONLY_FLAG_NON_PRIMARY;
 662   ictx->image_lock.unlock();
 663
 664   auto on_promote = new LambdaContext([ictx, on_finish](int r) {
 665       ictx->image_lock.lock();
 666       ictx->read_only_mask |= IMAGE_READ_ONLY_FLAG_NON_PRIMARY;
 667       ictx->image_lock.unlock();
 668
 669       ictx->state->handle_update_notification();
 670       on_finish->complete(r);
 671     });
 672
 673   auto on_refresh = new LambdaContext([ictx, force, on_promote](int r) {
 674       if (r < 0) {
 675         lderr(ictx->cct) << "refresh failed: " << cpp_strerror(r) << dendl;
 676         on_promote->complete(r);
 677         return;
 678       }
 679
 680       auto req = mirror::PromoteRequest<>::create(*ictx, force, on_promote);
 681       req->send();
 682     });
 683   ictx->state->refresh(on_refresh);
 684 }
 685
 686 template <typename I>
 687 int Mirror<I>::image_demote(I *ictx) {
 688   CephContext *cct = ictx->cct;
 689
 690   C_SaferCond ctx;
 691   Mirror<I>::image_demote(ictx, &ctx);
 692   int r = ctx.wait();
 693   if (r < 0) {
 694     lderr(cct) << "failed to demote image" << dendl;
 695     return r;
 696   }
 697
 698   return 0;
 699 }
 700
 701 template <typename I>
 702 void Mirror<I>::image_demote(I *ictx, Context *on_finish) {
 703   CephContext *cct = ictx->cct;
 704   ldout(cct, 20) << "ictx=" << ictx << dendl;
 705
 706   auto on_cleanup = new LambdaContext([ictx, on_finish](int r) {
 707       ictx->image_lock.lock();
 708       ictx->read_only_mask |= IMAGE_READ_ONLY_FLAG_NON_PRIMARY;
 709       ictx->image_lock.unlock();
 710
 711       ictx->state->handle_update_notification();
 712
 713       on_finish->complete(r);
 714     });
 715   auto on_refresh = new LambdaContext([ictx, on_cleanup](int r) {
 716       if (r < 0) {
 717         lderr(ictx->cct) << "refresh failed: " << cpp_strerror(r) << dendl;
 718         on_cleanup->complete(r);
 719         return;
 720       }
 721
 722       auto req = mirror::DemoteRequest<>::create(*ictx, on_cleanup);
 723       req->send();
 724     });
 725
 726   // ensure we can create a snapshot after setting the non-primary
 727   // feature bit
 728   ictx->image_lock.lock();
 729   ictx->read_only_mask &= ~IMAGE_READ_ONLY_FLAG_NON_PRIMARY;
 730   ictx->image_lock.unlock();
 731
 732   ictx->state->refresh(on_refresh);
 733 }
 734
 735 template <typename I>
 736 int Mirror<I>::image_resync(I *ictx) {
 737   CephContext *cct = ictx->cct;
 738   ldout(cct, 20) << "ictx=" << ictx << dendl;
 739
 740   int r = ictx->state->refresh_if_required();
 741   if (r < 0) {
 742     return r;
 743   }
 744
 745   cls::rbd::MirrorImage mirror_image;
 746   mirror::PromotionState promotion_state;
 747   std::string primary_mirror_uuid;
 748   C_SaferCond get_info_ctx;
 749   auto req = mirror::GetInfoRequest<I>::create(*ictx, &mirror_image,
 750                                                &promotion_state,
 751                                                &primary_mirror_uuid,
 752                                                &get_info_ctx);
 753   req->send();
 754
 755   r = get_info_ctx.wait();
 756   if (r < 0) {
 757     return r;
 758   }
 759
 760   if (promotion_state == mirror::PROMOTION_STATE_PRIMARY) {
 761     lderr(cct) << "image is primary, cannot resync to itself" << dendl;
 762     return -EINVAL;
 763   }
 764
 765   if (mirror_image.mode == cls::rbd::MIRROR_IMAGE_MODE_JOURNAL) {
 766     // flag the journal indicating that we want to rebuild the local image
 767     r = Journal<I>::request_resync(ictx);
 768     if (r < 0) {
 769       lderr(cct) << "failed to request resync: " << cpp_strerror(r) << dendl;
 770       return r;
 771     }
 772   } else if (mirror_image.mode == cls::rbd::MIRROR_IMAGE_MODE_SNAPSHOT) {
 773     std::string mirror_uuid;
 774     r = uuid_get(ictx->md_ctx, &mirror_uuid);
 775     if (r < 0) {
 776       return r;
 777     }
 778
 779     mirror::snapshot::ImageMeta image_meta(ictx, mirror_uuid);
 780
 781     C_SaferCond load_meta_ctx;
 782     image_meta.load(&load_meta_ctx);
 783     r = load_meta_ctx.wait();
 784     if (r < 0 && r != -ENOENT) {
 785       lderr(cct) << "failed to load mirror image-meta: " << cpp_strerror(r)
 786                  << dendl;
 787       return r;
 788     }
 789
 790     image_meta.resync_requested = true;
 791
 792     C_SaferCond save_meta_ctx;
 793     image_meta.save(&save_meta_ctx);
 794     r = save_meta_ctx.wait();
 795     if (r < 0) {
 796       lderr(cct) << "failed to request resync: " << cpp_strerror(r) << dendl;
 797       return r;
 798     }
 799   } else {
 800     lderr(cct) << "unknown mirror mode" << dendl;
 801     return -EINVAL;
 802   }
 803
 804   return 0;
 805 }
 806
 807 template <typename I>
 808 void Mirror<I>::image_get_info(I *ictx, mirror_image_info_t *mirror_image_info,
 809                                Context *on_finish) {
 810   CephContext *cct = ictx->cct;
 811   ldout(cct, 20) << "ictx=" << ictx << dendl;
 812
 813   auto on_refresh = new LambdaContext(
 814     [ictx, mirror_image_info, on_finish](int r) {
 815       if (r < 0) {
 816         lderr(ictx->cct) << "refresh failed: " << cpp_strerror(r) << dendl;
 817         on_finish->complete(r);
 818         return;
 819       }
 820
 821       auto ctx = new C_ImageGetInfo(mirror_image_info, nullptr, on_finish);
 822       auto req = mirror::GetInfoRequest<I>::create(*ictx, &ctx->mirror_image,
 823                                                    &ctx->promotion_state,
 824                                                    &ctx->primary_mirror_uuid,
 825                                                    ctx);
 826       req->send();
 827     });
 828
 829   if (ictx->state->is_refresh_required()) {
 830     ictx->state->refresh(on_refresh);
 831   } else {
 832     on_refresh->complete(0);
 833   }
 834 }
 835
 836 template <typename I>
 837 int Mirror<I>::image_get_info(I *ictx, mirror_image_info_t *mirror_image_info) {
 838   C_SaferCond ctx;
 839   image_get_info(ictx, mirror_image_info, &ctx);
 840
 841   int r = ctx.wait();
 842   if (r < 0) {
 843     return r;
 844   }
 845   return 0;
 846 }
 847
 848 template <typename I>
 849 void Mirror<I>::image_get_info(librados::IoCtx& io_ctx,
 850                                ContextWQ *op_work_queue,
 851                                const std::string &image_id,
 852                                mirror_image_info_t *mirror_image_info,
 853                                Context *on_finish) {
 854   auto cct = reinterpret_cast<CephContext *>(io_ctx.cct());
 855   ldout(cct, 20) << "pool_id=" << io_ctx.get_id() << ", image_id=" << image_id
 856                  << dendl;
 857
 858   auto ctx = new C_ImageGetInfo(mirror_image_info, nullptr, on_finish);
 859   auto req = mirror::GetInfoRequest<I>::create(io_ctx, op_work_queue, image_id,
 860                                                &ctx->mirror_image,
 861                                                &ctx->promotion_state,
 862                                                &ctx->primary_mirror_uuid, ctx);
 863   req->send();
 864 }
 865
 866 template <typename I>
 867 int Mirror<I>::image_get_info(librados::IoCtx& io_ctx,
 868                               ContextWQ *op_work_queue,
 869                               const std::string &image_id,
 870                               mirror_image_info_t *mirror_image_info) {
 871   C_SaferCond ctx;
 872   image_get_info(io_ctx, op_work_queue, image_id, mirror_image_info, &ctx);
 873
 874   int r = ctx.wait();
 875   if (r < 0) {
 876     return r;
 877   }
 878   return 0;
 879 }
 880
 881 template <typename I>
 882 void Mirror<I>::image_get_mode(I *ictx, mirror_image_mode_t *mode,
 883                                Context *on_finish) {
 884   CephContext *cct = ictx->cct;
 885   ldout(cct, 20) << "ictx=" << ictx << dendl;
 886
 887   auto ctx = new C_ImageGetInfo(nullptr, mode, on_finish);
 888   auto req = mirror::GetInfoRequest<I>::create(*ictx, &ctx->mirror_image,
 889                                                &ctx->promotion_state,
 890                                                &ctx->primary_mirror_uuid, ctx);
 891   req->send();
 892 }
 893
 894 template <typename I>
 895 int Mirror<I>::image_get_mode(I *ictx, mirror_image_mode_t *mode) {
 896   C_SaferCond ctx;
 897   image_get_mode(ictx, mode, &ctx);
 898
 899   int r = ctx.wait();
 900   if (r < 0) {
 901     return r;
 902   }
 903   return 0;
 904 }
 905
 906 template <typename I>
 907 void Mirror<I>::image_get_global_status(I *ictx,
 908                                         mirror_image_global_status_t *status,
 909                                         Context *on_finish) {
 910   CephContext *cct = ictx->cct;
 911   ldout(cct, 20) << "ictx=" << ictx << dendl;
 912
 913   auto ctx = new C_ImageGetGlobalStatus(ictx->name, status, on_finish);
 914   auto req = mirror::GetStatusRequest<I>::create(
 915     *ictx, &ctx->mirror_image_status_internal, &ctx->mirror_image,
 916     &ctx->promotion_state, ctx);
 917   req->send();
 918 }
 919
 920 template <typename I>
 921 int Mirror<I>::image_get_global_status(I *ictx,
 922                                        mirror_image_global_status_t *status) {
 923   C_SaferCond ctx;
 924   image_get_global_status(ictx, status, &ctx);
 925
 926   int r = ctx.wait();
 927   if (r < 0) {
 928     return r;
 929   }
 930   return 0;
 931 }
 932
 933 template <typename I>
 934 int Mirror<I>::image_get_instance_id(I *ictx, std::string *instance_id) {
 935   CephContext *cct = ictx->cct;
 936   ldout(cct, 20) << "ictx=" << ictx << dendl;
 937
 938   cls::rbd::MirrorImage mirror_image;
 939   int r = cls_client::mirror_image_get(&ictx->md_ctx, ictx->id, &mirror_image);
 940   if (r < 0 && r != -ENOENT) {
 941     lderr(cct) << "failed to retrieve mirroring state: " << cpp_strerror(r)
 942                << dendl;
 943     return r;
 944   } else if (mirror_image.state != cls::rbd::MIRROR_IMAGE_STATE_ENABLED) {
 945     lderr(cct) << "mirroring is not currently enabled" << dendl;
 946     return -EINVAL;
 947   }
 948
 949   entity_inst_t instance;
 950   r = cls_client::mirror_image_instance_get(&ictx->md_ctx,
 951                                             mirror_image.global_image_id,
 952                                             &instance);
 953   if (r < 0) {
 954     if (r != -ENOENT && r != -ESTALE) {
 955       lderr(cct) << "failed to get mirror image instance: " << cpp_strerror(r)
 956                  << dendl;
 957     }
 958     return r;
 959   }
 960
 961   *instance_id = stringify(instance.name.num());
 962   return 0;
 963 }
 964
 965 template <typename I>
 966 int Mirror<I>::site_name_get(librados::Rados& rados, std::string* name) {
 967   CephContext *cct = reinterpret_cast<CephContext *>(rados.cct());
 968   ldout(cct, 20) << dendl;
 969
 970   int r = get_config_key(rados, RBD_MIRROR_SITE_NAME_CONFIG_KEY, name);
 971   if (r == -EOPNOTSUPP) {
 972     return r;
 973   } else if (r == -ENOENT || name->empty()) {
 974     // default to the cluster fsid
 975     r = rados.cluster_fsid(name);
 976     if (r < 0) {
 977       lderr(cct) << "failed to retrieve cluster fsid: " << cpp_strerror(r)
 978                  << dendl;
 979     }
 980     return r;
 981   } else if (r < 0) {
 982     lderr(cct) << "failed to retrieve site name: " << cpp_strerror(r)
 983                << dendl;
 984     return r;
 985   }
 986
 987   return 0;
 988 }
 989
 990 template <typename I>
 991 int Mirror<I>::site_name_set(librados::Rados& rados, const std::string& name) {
 992   CephContext *cct = reinterpret_cast<CephContext *>(rados.cct());
 993
 994   std::string site_name{name};
 995   boost::algorithm::trim(site_name);
 996   ldout(cct, 20) << "site_name=" << site_name << dendl;
 997
 998   int r = set_config_key(rados, RBD_MIRROR_SITE_NAME_CONFIG_KEY, name);
 999   if (r == -EOPNOTSUPP) {
1000     return r;
1001   } else if (r < 0 && r != -ENOENT) {
1002     lderr(cct) << "failed to update site name: " << cpp_strerror(r)
1003                << dendl;
1004     return r;
1005   }
1006
1007   return 0;
1008 }
1009
1010 template <typename I>
1011 int Mirror<I>::mode_get(librados::IoCtx& io_ctx,
1012                         rbd_mirror_mode_t *mirror_mode) {
1013   CephContext *cct = reinterpret_cast<CephContext *>(io_ctx.cct());
1014   ldout(cct, 20) << dendl;
1015
1016   cls::rbd::MirrorMode mirror_mode_internal;
1017   int r = cls_client::mirror_mode_get(&io_ctx, &mirror_mode_internal);
1018   if (r < 0) {
1019     lderr(cct) << "failed to retrieve mirror mode: " << cpp_strerror(r)
1020                << dendl;
1021     return r;
1022   }
1023
1024   switch (mirror_mode_internal) {
1025   case cls::rbd::MIRROR_MODE_DISABLED:
1026   case cls::rbd::MIRROR_MODE_IMAGE:
1027   case cls::rbd::MIRROR_MODE_POOL:
1028     *mirror_mode = static_cast<rbd_mirror_mode_t>(mirror_mode_internal);
1029     break;
1030   default:
1031     lderr(cct) << "unknown mirror mode ("
1032                << static_cast<uint32_t>(mirror_mode_internal) << ")"
1033                << dendl;
1034     return -EINVAL;
1035   }
1036   return 0;
1037 }
1038
1039 template <typename I>
1040 int Mirror<I>::mode_set(librados::IoCtx& io_ctx,
1041                         rbd_mirror_mode_t mirror_mode) {
1042   CephContext *cct = reinterpret_cast<CephContext *>(io_ctx.cct());
1043   ldout(cct, 20) << dendl;
1044
1045   cls::rbd::MirrorMode next_mirror_mode;
1046   switch (mirror_mode) {
1047   case RBD_MIRROR_MODE_DISABLED:
1048   case RBD_MIRROR_MODE_IMAGE:
1049   case RBD_MIRROR_MODE_POOL:
1050     next_mirror_mode = static_cast<cls::rbd::MirrorMode>(mirror_mode);
1051     break;
1052   default:
1053     lderr(cct) << "unknown mirror mode ("
1054                << static_cast<uint32_t>(mirror_mode) << ")" << dendl;
1055     return -EINVAL;
1056   }
1057
1058   int r;
1059   if (next_mirror_mode == cls::rbd::MIRROR_MODE_DISABLED) {
1060     // fail early if pool still has peers registered and attempting to disable
1061     std::vector<cls::rbd::MirrorPeer> mirror_peers;
1062     r = cls_client::mirror_peer_list(&io_ctx, &mirror_peers);
1063     if (r < 0 && r != -ENOENT) {
1064       lderr(cct) << "failed to list peers: " << cpp_strerror(r) << dendl;
1065       return r;
1066     } else if (!mirror_peers.empty()) {
1067       lderr(cct) << "mirror peers still registered" << dendl;
1068       return -EBUSY;
1069     }
1070   }
1071
1072   cls::rbd::MirrorMode current_mirror_mode;
1073   r = cls_client::mirror_mode_get(&io_ctx, &current_mirror_mode);
1074   if (r < 0) {
1075     lderr(cct) << "failed to retrieve mirror mode: " << cpp_strerror(r)
1076                << dendl;
1077     return r;
1078   }
1079
1080   if (current_mirror_mode == next_mirror_mode) {
1081     return 0;
1082   } else if (current_mirror_mode == cls::rbd::MIRROR_MODE_DISABLED) {
1083     uuid_d uuid_gen;
1084     uuid_gen.generate_random();
1085     r = cls_client::mirror_uuid_set(&io_ctx, uuid_gen.to_string());
1086     if (r < 0) {
1087       lderr(cct) << "failed to allocate mirroring uuid: " << cpp_strerror(r)
1088                  << dendl;
1089       return r;
1090     }
1091   }
1092
1093   if (current_mirror_mode != cls::rbd::MIRROR_MODE_IMAGE) {
1094     r = cls_client::mirror_mode_set(&io_ctx, cls::rbd::MIRROR_MODE_IMAGE);
1095     if (r < 0) {
1096       lderr(cct) << "failed to set mirror mode to image: "
1097                  << cpp_strerror(r) << dendl;
1098       return r;
1099     }
1100
1101     r = MirroringWatcher<>::notify_mode_updated(io_ctx,
1102                                                 cls::rbd::MIRROR_MODE_IMAGE);
1103     if (r < 0) {
1104       lderr(cct) << "failed to send update notification: " << cpp_strerror(r)
1105                  << dendl;
1106     }
1107   }
1108
1109   if (next_mirror_mode == cls::rbd::MIRROR_MODE_IMAGE) {
1110     return 0;
1111   }
1112
1113   if (next_mirror_mode == cls::rbd::MIRROR_MODE_POOL) {
1114     map<string, string> images;
1115     r = Image<I>::list_images_v2(io_ctx, &images);
1116     if (r < 0) {
1117       lderr(cct) << "failed listing images: " << cpp_strerror(r) << dendl;
1118       return r;
1119     }
1120
1121     for (const auto& img_pair : images) {
1122       uint64_t features;
1123       uint64_t incompatible_features;
1124       r = cls_client::get_features(&io_ctx, util::header_name(img_pair.second),
1125                                    true, &features, &incompatible_features);
1126       if (r < 0) {
1127         lderr(cct) << "error getting features for image " << img_pair.first
1128                    << ": " << cpp_strerror(r) << dendl;
1129         return r;
1130       }
1131
1132       // Enable only journal based mirroring
1133
1134       if ((features & RBD_FEATURE_JOURNALING) != 0) {
1135         I *img_ctx = I::create("", img_pair.second, nullptr, io_ctx, false);
1136         r = img_ctx->state->open(0);
1137         if (r < 0) {
1138           lderr(cct) << "error opening image "<< img_pair.first << ": "
1139                      << cpp_strerror(r) << dendl;
1140           return r;
1141         }
1142
1143         r = image_enable(img_ctx, RBD_MIRROR_IMAGE_MODE_JOURNAL, true);
1144         int close_r = img_ctx->state->close();
1145         if (r < 0) {
1146           lderr(cct) << "error enabling mirroring for image "
1147                      << img_pair.first << ": " << cpp_strerror(r) << dendl;
1148           return r;
1149         } else if (close_r < 0) {
1150           lderr(cct) << "failed to close image " << img_pair.first << ": "
1151                      << cpp_strerror(close_r) << dendl;
1152           return close_r;
1153         }
1154       }
1155     }
1156   } else if (next_mirror_mode == cls::rbd::MIRROR_MODE_DISABLED) {
1157     while (true) {
1158       bool retry_busy = false;
1159       bool pending_busy = false;
1160
1161       std::set<std::string> image_ids;
1162       r = list_mirror_images(io_ctx, image_ids);
1163       if (r < 0) {
1164         lderr(cct) << "failed listing images: " << cpp_strerror(r) << dendl;
1165         return r;
1166       }
1167
1168       for (const auto& img_id : image_ids) {
1169         if (current_mirror_mode == cls::rbd::MIRROR_MODE_IMAGE) {
1170           cls::rbd::MirrorImage mirror_image;
1171           r = cls_client::mirror_image_get(&io_ctx, img_id, &mirror_image);
1172           if (r < 0 && r != -ENOENT) {
1173             lderr(cct) << "failed to retrieve mirroring state for image id "
1174                        << img_id << ": " << cpp_strerror(r) << dendl;
1175             return r;
1176           }
1177           if (mirror_image.state == cls::rbd::MIRROR_IMAGE_STATE_ENABLED) {
1178             lderr(cct) << "failed to disable mirror mode: there are still "
1179                        << "images with mirroring enabled" << dendl;
1180             return -EINVAL;
1181           }
1182         } else {
1183           I *img_ctx = I::create("", img_id, nullptr, io_ctx, false);
1184           r = img_ctx->state->open(0);
1185           if (r < 0) {
1186             lderr(cct) << "error opening image id "<< img_id << ": "
1187                        << cpp_strerror(r) << dendl;
1188             return r;
1189           }
1190
1191           r = image_disable(img_ctx, false);
1192           int close_r = img_ctx->state->close();
1193           if (r == -EBUSY) {
1194             pending_busy = true;
1195           } else if (r < 0) {
1196             lderr(cct) << "error disabling mirroring for image id " << img_id
1197                        << cpp_strerror(r) << dendl;
1198             return r;
1199           } else if (close_r < 0) {
1200             lderr(cct) << "failed to close image id " << img_id << ": "
1201                        << cpp_strerror(close_r) << dendl;
1202             return close_r;
1203           } else if (pending_busy) {
1204             // at least one mirrored image was successfully disabled, so we can
1205             // retry any failures caused by busy parent/child relationships
1206             retry_busy = true;
1207           }
1208         }
1209       }
1210
1211       if (!retry_busy && pending_busy) {
1212         lderr(cct) << "error disabling mirroring for one or more images"
1213                    << dendl;
1214         return -EBUSY;
1215       } else if (!retry_busy) {
1216         break;
1217       }
1218     }
1219   }
1220
1221   r = cls_client::mirror_mode_set(&io_ctx, next_mirror_mode);
1222   if (r < 0) {
1223     lderr(cct) << "failed to set mirror mode: " << cpp_strerror(r) << dendl;
1224     return r;
1225   }
1226
1227   r = MirroringWatcher<>::notify_mode_updated(io_ctx, next_mirror_mode);
1228   if (r < 0) {
1229     lderr(cct) << "failed to send update notification: " << cpp_strerror(r)
1230                << dendl;
1231   }
1232   return 0;
1233 }
1234
1235 template <typename I>
1236 int Mirror<I>::uuid_get(librados::IoCtx& io_ctx, std::string* mirror_uuid) {
1237   CephContext *cct = reinterpret_cast<CephContext *>(io_ctx.cct());
1238   ldout(cct, 20) << dendl;
1239
1240   C_SaferCond ctx;
1241   uuid_get(io_ctx, mirror_uuid, &ctx);
1242   int r = ctx.wait();
1243   if (r < 0) {
1244     if (r != -ENOENT) {
1245       lderr(cct) << "failed to retrieve mirroring uuid: " << cpp_strerror(r)
1246                  << dendl;
1247     }
1248     return r;
1249   }
1250
1251   return 0;
1252 }
1253
1254 template <typename I>
1255 void Mirror<I>::uuid_get(librados::IoCtx& io_ctx, std::string* mirror_uuid,
1256                          Context* on_finish) {
1257   CephContext *cct = reinterpret_cast<CephContext *>(io_ctx.cct());
1258   ldout(cct, 20) << dendl;
1259
1260   auto req = mirror::GetUuidRequest<I>::create(io_ctx, mirror_uuid, on_finish);
1261   req->send();
1262 }
1263
1264 template <typename I>
1265 int Mirror<I>::peer_bootstrap_create(librados::IoCtx& io_ctx,
1266                                      std::string* token) {
1267   CephContext *cct = reinterpret_cast<CephContext *>(io_ctx.cct());
1268   ldout(cct, 20) << dendl;
1269
1270   auto mirror_mode = cls::rbd::MIRROR_MODE_DISABLED;
1271   int r = cls_client::mirror_mode_get(&io_ctx, &mirror_mode);
1272   if (r < 0 && r != -ENOENT) {
1273     lderr(cct) << "failed to retrieve mirroring mode: " << cpp_strerror(r)
1274                << dendl;
1275     return r;
1276   } else if (mirror_mode == cls::rbd::MIRROR_MODE_DISABLED) {
1277     return -EINVAL;
1278   }
1279
1280   // retrieve the cluster fsid
1281   std::string fsid;
1282   librados::Rados rados(io_ctx);
1283   r = rados.cluster_fsid(&fsid);
1284   if (r < 0) {
1285     lderr(cct) << "failed to retrieve cluster fsid: " << cpp_strerror(r)
1286                << dendl;
1287     return r;
1288   }
1289
1290   std::string peer_client_id;
1291   std::string cephx_key;
1292   r = create_bootstrap_user(cct, rados, &peer_client_id, &cephx_key);
1293   if (r < 0) {
1294     return r;
1295   }
1296
1297   std::string mon_host = cct->_conf.get_val<std::string>("mon_host");
1298   ldout(cct, 20) << "mon_host=" << mon_host << dendl;
1299
1300   // format the token response
1301   bufferlist token_bl;
1302   token_bl.append(
1303     R"({)" \
1304       R"("fsid":")" + fsid + R"(",)" + \
1305       R"("client_id":")" + peer_client_id + R"(",)" + \
1306       R"("key":")" + cephx_key + R"(",)" + \
1307       R"("mon_host":")" + \
1308         boost::replace_all_copy(mon_host, "\"", "\\\"") + R"(")" + \
1309     R"(})");
1310   ldout(cct, 20) << "token=" << token_bl.to_str() << dendl;
1311
1312   bufferlist base64_bl;
1313   token_bl.encode_base64(base64_bl);
1314   *token = base64_bl.to_str();
1315
1316   return 0;
1317 }
1318
1319 template <typename I>
1320 int Mirror<I>::peer_bootstrap_import(librados::IoCtx& io_ctx,
1321                                      rbd_mirror_peer_direction_t direction,
1322                                      const std::string& token) {
1323   CephContext *cct = reinterpret_cast<CephContext *>(io_ctx.cct());
1324   ldout(cct, 20) << dendl;
1325
1326   if (direction != RBD_MIRROR_PEER_DIRECTION_RX &&
1327       direction != RBD_MIRROR_PEER_DIRECTION_RX_TX) {
1328     lderr(cct) << "invalid mirror peer direction" << dendl;
1329     return -EINVAL;
1330   }
1331
1332   bufferlist token_bl;
1333   try {
1334     bufferlist base64_bl;
1335     base64_bl.append(token);
1336     token_bl.decode_base64(base64_bl);
1337   } catch (buffer::error& err) {
1338     lderr(cct) << "failed to decode base64" << dendl;
1339     return -EINVAL;
1340   }
1341
1342   ldout(cct, 20) << "token=" << token_bl.to_str() << dendl;
1343
1344   bool json_valid = false;
1345   std::string expected_remote_fsid;
1346   std::string remote_client_id;
1347   std::string remote_key;
1348   std::string remote_mon_host;
1349
1350   json_spirit::mValue json_root;
1351   if(json_spirit::read(token_bl.to_str(), json_root)) {
1352     try {
1353       auto& json_obj = json_root.get_obj();
1354       expected_remote_fsid = json_obj["fsid"].get_str();
1355       remote_client_id = json_obj["client_id"].get_str();
1356       remote_key = json_obj["key"].get_str();
1357       remote_mon_host = json_obj["mon_host"].get_str();
1358       json_valid = true;
1359     } catch (std::runtime_error&) {
1360     }
1361   }
1362
1363   if (!json_valid) {
1364     lderr(cct) << "invalid bootstrap token JSON received" << dendl;
1365     return -EINVAL;
1366   }
1367
1368   // sanity check import process
1369   std::string local_fsid;
1370   librados::Rados rados(io_ctx);
1371   int r = rados.cluster_fsid(&local_fsid);
1372   if (r < 0) {
1373     lderr(cct) << "failed to retrieve cluster fsid: " << cpp_strerror(r)
1374                << dendl;
1375     return r;
1376   }
1377
1378   std::string local_site_name;
1379   r = site_name_get(rados, &local_site_name);
1380   if (r < 0) {
1381     lderr(cct) << "failed to retrieve cluster site name: " << cpp_strerror(r)
1382                << dendl;
1383     return r;
1384   }
1385
1386   // attempt to connect to remote cluster
1387   librados::Rados remote_rados;
1388   remote_rados.init(remote_client_id.c_str());
1389
1390   auto remote_cct = reinterpret_cast<CephContext*>(remote_rados.cct());
1391   remote_cct->_conf.set_val("mon_host", remote_mon_host);
1392   remote_cct->_conf.set_val("key", remote_key);
1393
1394   r = remote_rados.connect();
1395   if (r < 0) {
1396     lderr(cct) << "failed to connect to peer cluster: " << cpp_strerror(r)
1397                << dendl;
1398     return r;
1399   }
1400
1401   std::string remote_fsid;
1402   r = remote_rados.cluster_fsid(&remote_fsid);
1403   if (r < 0) {
1404     lderr(cct) << "failed to retrieve remote cluster fsid: "
1405                << cpp_strerror(r) << dendl;
1406     return r;
1407   } else if (local_fsid == remote_fsid) {
1408     lderr(cct) << "cannot import token for local cluster" << dendl;
1409     return -EINVAL;
1410   } else if (expected_remote_fsid != remote_fsid) {
1411     lderr(cct) << "unexpected remote cluster fsid" << dendl;
1412     return -EINVAL;
1413   }
1414
1415   std::string remote_site_name;
1416   r = site_name_get(remote_rados, &remote_site_name);
1417   if (r < 0) {
1418     lderr(cct) << "failed to retrieve remote cluster site name: "
1419                << cpp_strerror(r) << dendl;
1420     return r;
1421   } else if (local_site_name == remote_site_name) {
1422     lderr(cct) << "cannot import token for duplicate site name" << dendl;
1423     return -EINVAL;
1424   }
1425
1426   librados::IoCtx remote_io_ctx;
1427   r = remote_rados.ioctx_create(io_ctx.get_pool_name().c_str(), remote_io_ctx);
1428   if (r == -ENOENT) {
1429     ldout(cct, 10) << "remote pool does not exist" << dendl;
1430     return r;
1431   } else if (r < 0) {
1432     lderr(cct) << "failed to open remote pool '" << io_ctx.get_pool_name()
1433                << "': " << cpp_strerror(r) << dendl;
1434     return r;
1435   }
1436
1437   auto remote_mirror_mode = cls::rbd::MIRROR_MODE_DISABLED;
1438   r = cls_client::mirror_mode_get(&remote_io_ctx, &remote_mirror_mode);
1439   if (r < 0 && r != -ENOENT) {
1440     lderr(cct) << "failed to retrieve remote mirroring mode: "
1441                << cpp_strerror(r) << dendl;
1442     return r;
1443   } else if (remote_mirror_mode == cls::rbd::MIRROR_MODE_DISABLED) {
1444     return -ENOSYS;
1445   }
1446
1447   auto local_mirror_mode = cls::rbd::MIRROR_MODE_DISABLED;
1448   r = cls_client::mirror_mode_get(&io_ctx, &local_mirror_mode);
1449   if (r < 0 && r != -ENOENT) {
1450     lderr(cct) << "failed to retrieve local mirroring mode: " << cpp_strerror(r)
1451                << dendl;
1452     return r;
1453   } else if (local_mirror_mode == cls::rbd::MIRROR_MODE_DISABLED) {
1454     // copy mirror mode from remote peer
1455     r = mode_set(io_ctx, static_cast<rbd_mirror_mode_t>(remote_mirror_mode));
1456     if (r < 0) {
1457       return r;
1458     }
1459   }
1460
1461   if (direction == RBD_MIRROR_PEER_DIRECTION_RX_TX) {
1462     // create a local mirror peer user and export it to the remote cluster
1463     std::string local_client_id;
1464     std::string local_key;
1465     r = create_bootstrap_user(cct, rados, &local_client_id, &local_key);
1466     if (r < 0) {
1467       return r;
1468     }
1469
1470     std::string local_mon_host = cct->_conf.get_val<std::string>("mon_host");
1471
1472     // create local cluster peer in remote cluster
1473     r = create_bootstrap_peer(cct, remote_io_ctx,
1474                               RBD_MIRROR_PEER_DIRECTION_RX_TX, local_site_name,
1475                               local_fsid, local_client_id, local_key,
1476                               local_mon_host, "local", "remote");
1477     if (r < 0) {
1478       return r;
1479     }
1480   }
1481
1482   // create remote cluster peer in local cluster
1483   r = create_bootstrap_peer(cct, io_ctx, direction, remote_site_name,
1484                             remote_fsid, remote_client_id, remote_key,
1485                             remote_mon_host, "remote", "local");
1486   if (r < 0) {
1487     return r;
1488   }
1489
1490   return 0;
1491 }
1492
1493 template <typename I>
1494 int Mirror<I>::peer_site_add(librados::IoCtx& io_ctx, std::string *uuid,
1495                              mirror_peer_direction_t direction,
1496                              const std::string &site_name,
1497                              const std::string &client_name) {
1498   CephContext *cct = reinterpret_cast<CephContext *>(io_ctx.cct());
1499   ldout(cct, 20) << "name=" << site_name << ", "
1500                  << "client=" << client_name << dendl;
1501
1502   if (cct->_conf->cluster == site_name) {
1503     lderr(cct) << "cannot add self as remote peer" << dendl;
1504     return -EINVAL;
1505   }
1506
1507   if (direction == RBD_MIRROR_PEER_DIRECTION_TX) {
1508     return -EINVAL;
1509   }
1510
1511   int r;
1512   do {
1513     uuid_d uuid_gen;
1514     uuid_gen.generate_random();
1515
1516     *uuid = uuid_gen.to_string();
1517     r = cls_client::mirror_peer_add(
1518       &io_ctx, {*uuid, static_cast<cls::rbd::MirrorPeerDirection>(direction),
1519                 site_name, client_name, ""});
1520     if (r == -ESTALE) {
1521       ldout(cct, 5) << "duplicate UUID detected, retrying" << dendl;
1522     } else if (r < 0) {
1523       lderr(cct) << "failed to add mirror peer '" << site_name << "': "
1524                  << cpp_strerror(r) << dendl;
1525       return r;
1526     }
1527   } while (r == -ESTALE);
1528   return 0;
1529 }
1530
1531 template <typename I>
1532 int Mirror<I>::peer_site_remove(librados::IoCtx& io_ctx,
1533                                 const std::string &uuid) {
1534   CephContext *cct = reinterpret_cast<CephContext *>(io_ctx.cct());
1535   ldout(cct, 20) << "uuid=" << uuid << dendl;
1536
1537   int r = remove_peer_config_key(io_ctx, uuid);
1538   if (r < 0) {
1539     lderr(cct) << "failed to remove peer attributes '" << uuid << "': "
1540                << cpp_strerror(r) << dendl;
1541     return r;
1542   }
1543
1544   r = cls_client::mirror_peer_remove(&io_ctx, uuid);
1545   if (r < 0 && r != -ENOENT) {
1546     lderr(cct) << "failed to remove peer '" << uuid << "': "
1547                << cpp_strerror(r) << dendl;
1548     return r;
1549   }
1550
1551   vector<string> names;
1552   r = Namespace<I>::list(io_ctx, &names);
1553   if (r < 0) {
1554     return r;
1555   }
1556
1557   names.push_back("");
1558
1559   librados::IoCtx ns_io_ctx;
1560   ns_io_ctx.dup(io_ctx);
1561
1562   for (auto &name : names) {
1563     ns_io_ctx.set_namespace(name);
1564
1565     std::set<std::string> image_ids;
1566     r = list_mirror_images(ns_io_ctx, image_ids);
1567     if (r < 0) {
1568       lderr(cct) << "failed listing images in "
1569                  << (name.empty() ? "default" : name) << " namespace : "
1570                  << cpp_strerror(r) << dendl;
1571       return r;
1572     }
1573
1574     for (const auto& image_id : image_ids) {
1575       cls::rbd::MirrorImage mirror_image;
1576       r = cls_client::mirror_image_get(&ns_io_ctx, image_id, &mirror_image);
1577       if (r == -ENOENT) {
1578         continue;
1579       }
1580       if (r < 0) {
1581         lderr(cct) << "error getting mirror info for image " << image_id
1582                    << ": " << cpp_strerror(r) << dendl;
1583         return r;
1584       }
1585       if (mirror_image.mode != cls::rbd::MIRROR_IMAGE_MODE_SNAPSHOT) {
1586         continue;
1587       }
1588
1589       // Snapshot based mirroring. Unlink the peer from mirroring snapshots.
1590       // TODO: optimize.
1591
1592       I *img_ctx = I::create("", image_id, nullptr, ns_io_ctx, false);
1593       img_ctx->read_only_mask &= ~IMAGE_READ_ONLY_FLAG_NON_PRIMARY;
1594
1595       r = img_ctx->state->open(0);
1596       if (r == -ENOENT) {
1597         continue;
1598       }
1599       if (r < 0) {
1600         lderr(cct) << "error opening image " << image_id << ": "
1601                    << cpp_strerror(r) << dendl;
1602         return r;
1603       }
1604
1605       std::list<uint64_t> snap_ids;
1606       {
1607         std::shared_lock image_locker{img_ctx->image_lock};
1608         for (auto &it : img_ctx->snap_info) {
1609           auto info = boost::get<cls::rbd::MirrorSnapshotNamespace>(
1610             &it.second.snap_namespace);
1611           if (info && info->mirror_peer_uuids.count(uuid)) {
1612             snap_ids.push_back(it.first);
1613           }
1614         }
1615       }
1616       for (auto snap_id : snap_ids) {
1617         C_SaferCond cond;
1618         auto req = mirror::snapshot::UnlinkPeerRequest<I>::create(
1619           img_ctx, snap_id, uuid, &cond);
1620         req->send();
1621         r = cond.wait();
1622         if (r == -ENOENT) {
1623           r = 0;
1624         }
1625         if (r < 0) {
1626           break;
1627         }
1628       }
1629
1630       int close_r = img_ctx->state->close();
1631       if (r < 0) {
1632         lderr(cct) << "error unlinking peer for image " << image_id << ": "
1633                    << cpp_strerror(r) << dendl;
1634         return r;
1635       } else if (close_r < 0) {
1636         lderr(cct) << "failed to close image " << image_id << ": "
1637                    << cpp_strerror(close_r) << dendl;
1638         return close_r;
1639       }
1640     }
1641   }
1642
1643   return 0;
1644 }
1645
1646 template <typename I>
1647 int Mirror<I>::peer_site_list(librados::IoCtx& io_ctx,
1648                               std::vector<mirror_peer_site_t> *peers) {
1649   CephContext *cct = reinterpret_cast<CephContext *>(io_ctx.cct());
1650   ldout(cct, 20) << dendl;
1651
1652   std::vector<cls::rbd::MirrorPeer> mirror_peers;
1653   int r = cls_client::mirror_peer_list(&io_ctx, &mirror_peers);
1654   if (r < 0 && r != -ENOENT) {
1655     lderr(cct) << "failed to list peers: " << cpp_strerror(r) << dendl;
1656     return r;
1657   }
1658
1659   peers->clear();
1660   peers->reserve(mirror_peers.size());
1661   for (auto &mirror_peer : mirror_peers) {
1662     mirror_peer_site_t peer;
1663     peer.uuid = mirror_peer.uuid;
1664     peer.direction = static_cast<mirror_peer_direction_t>(
1665       mirror_peer.mirror_peer_direction);
1666     peer.site_name = mirror_peer.site_name;
1667     peer.mirror_uuid = mirror_peer.mirror_uuid;
1668     peer.client_name = mirror_peer.client_name;
1669     peer.last_seen = mirror_peer.last_seen.sec();
1670     peers->push_back(peer);
1671   }
1672   return 0;
1673 }
1674
1675 template <typename I>
1676 int Mirror<I>::peer_site_set_client(librados::IoCtx& io_ctx,
1677                                     const std::string &uuid,
1678                                     const std::string &client_name) {
1679   CephContext *cct = reinterpret_cast<CephContext *>(io_ctx.cct());
1680   ldout(cct, 20) << "uuid=" << uuid << ", "
1681                  << "client=" << client_name << dendl;
1682
1683   int r = cls_client::mirror_peer_set_client(&io_ctx, uuid, client_name);
1684   if (r < 0) {
1685     lderr(cct) << "failed to update client '" << uuid << "': "
1686                << cpp_strerror(r) << dendl;
1687     return r;
1688   }
1689   return 0;
1690 }
1691
1692 template <typename I>
1693 int Mirror<I>::peer_site_set_name(librados::IoCtx& io_ctx,
1694                                   const std::string &uuid,
1695                                   const std::string &site_name) {
1696   CephContext *cct = reinterpret_cast<CephContext *>(io_ctx.cct());
1697   ldout(cct, 20) << "uuid=" << uuid << ", "
1698                  << "name=" << site_name << dendl;
1699
1700   if (cct->_conf->cluster == site_name) {
1701     lderr(cct) << "cannot set self as remote peer" << dendl;
1702     return -EINVAL;
1703   }
1704
1705   int r = cls_client::mirror_peer_set_cluster(&io_ctx, uuid, site_name);
1706   if (r < 0) {
1707     lderr(cct) << "failed to update site '" << uuid << "': "
1708                << cpp_strerror(r) << dendl;
1709     return r;
1710   }
1711   return 0;
1712 }
1713
1714 template <typename I>
1715 int Mirror<I>::peer_site_set_direction(librados::IoCtx& io_ctx,
1716                                        const std::string &uuid,
1717                                        mirror_peer_direction_t direction) {
1718   cls::rbd::MirrorPeerDirection mirror_peer_direction = static_cast<
1719     cls::rbd::MirrorPeerDirection>(direction);
1720
1721   CephContext *cct = reinterpret_cast<CephContext *>(io_ctx.cct());
1722   ldout(cct, 20) << "uuid=" << uuid << ", "
1723                  << "direction=" << mirror_peer_direction << dendl;
1724
1725   int r = cls_client::mirror_peer_set_direction(&io_ctx, uuid,
1726                                                 mirror_peer_direction);
1727   if (r < 0) {
1728     lderr(cct) << "failed to update direction '" << uuid << "': "
1729                << cpp_strerror(r) << dendl;
1730     return r;
1731   }
1732   return 0;
1733 }
1734
1735 template <typename I>
1736 int Mirror<I>::peer_site_get_attributes(librados::IoCtx& io_ctx,
1737                                         const std::string &uuid,
1738                                         Attributes* attributes) {
1739   CephContext *cct = reinterpret_cast<CephContext *>(io_ctx.cct());
1740   ldout(cct, 20) << "uuid=" << uuid << dendl;
1741
1742   attributes->clear();
1743
1744   librados::Rados rados(io_ctx);
1745   std::string value;
1746   int r = get_config_key(rados, get_peer_config_key_name(io_ctx.get_id(), uuid),
1747                          &value);
1748   if (r == -ENOENT || value.empty()) {
1749     return -ENOENT;
1750   } else if (r < 0) {
1751     lderr(cct) << "failed to retrieve peer attributes: " << cpp_strerror(r)
1752                << dendl;
1753     return r;
1754   }
1755
1756   bool json_valid = false;
1757   json_spirit::mValue json_root;
1758   if(json_spirit::read(value, json_root)) {
1759     try {
1760       auto& json_obj = json_root.get_obj();
1761       for (auto& pairs : json_obj) {
1762         (*attributes)[pairs.first] = pairs.second.get_str();
1763       }
1764       json_valid = true;
1765     } catch (std::runtime_error&) {
1766     }
1767   }
1768
1769   if (!json_valid) {
1770     lderr(cct) << "invalid peer attributes JSON received" << dendl;
1771     return -EINVAL;
1772   }
1773   return 0;
1774 }
1775
1776 template <typename I>
1777 int Mirror<I>::peer_site_set_attributes(librados::IoCtx& io_ctx,
1778                                         const std::string &uuid,
1779                                         const Attributes& attributes) {
1780   CephContext *cct = reinterpret_cast<CephContext *>(io_ctx.cct());
1781   ldout(cct, 20) << "uuid=" << uuid << ", "
1782                  << "attributes=" << attributes << dendl;
1783
1784   std::vector<mirror_peer_site_t> mirror_peers;
1785   int r = peer_site_list(io_ctx, &mirror_peers);
1786   if (r < 0) {
1787     return r;
1788   }
1789
1790   if (std::find_if(mirror_peers.begin(), mirror_peers.end(),
1791                    [&uuid](const librbd::mirror_peer_site_t& peer) {
1792                      return uuid == peer.uuid;
1793                    }) == mirror_peers.end()) {
1794     ldout(cct, 5) << "mirror peer uuid " << uuid << " does not exist" << dendl;
1795     return -ENOENT;
1796   }
1797
1798   std::stringstream ss;
1799   ss << "{";
1800   for (auto& pair : attributes) {
1801     ss << "\\\"" << pair.first << "\\\": "
1802        << "\\\"" << pair.second << "\\\"";
1803     if (&pair != &(*attributes.rbegin())) {
1804       ss << ", ";
1805     }
1806   }
1807   ss << "}";
1808
1809   librados::Rados rados(io_ctx);
1810   r = set_config_key(rados, get_peer_config_key_name(io_ctx.get_id(), uuid),
1811                      ss.str());
1812   if (r < 0 && r != -ENOENT) {
1813     lderr(cct) << "failed to update peer attributes: " << cpp_strerror(r)
1814                << dendl;
1815     return r;
1816   }
1817
1818   return 0;
1819 }
1820
1821 template <typename I>
1822 int Mirror<I>::image_global_status_list(
1823     librados::IoCtx& io_ctx, const std::string &start_id, size_t max,
1824     IdToMirrorImageGlobalStatus *images) {
1825   CephContext *cct = reinterpret_cast<CephContext *>(io_ctx.cct());
1826   int r;
1827
1828   map<string, string> id_to_name;
1829   {
1830     map<string, string> name_to_id;
1831     r = Image<I>::list_images_v2(io_ctx, &name_to_id);
1832     if (r < 0) {
1833       return r;
1834     }
1835     for (auto it : name_to_id) {
1836       id_to_name[it.second] = it.first;
1837     }
1838   }
1839
1840   map<std::string, cls::rbd::MirrorImage> images_;
1841   map<std::string, cls::rbd::MirrorImageStatus> statuses_;
1842
1843   r = librbd::cls_client::mirror_image_status_list(&io_ctx, start_id, max,
1844                                                    &images_, &statuses_);
1845   if (r < 0 && r != -ENOENT) {
1846     lderr(cct) << "failed to list mirror image statuses: "
1847                << cpp_strerror(r) << dendl;
1848     return r;
1849   }
1850
1851   const std::string STATUS_NOT_FOUND("status not found");
1852   for (auto it = images_.begin(); it != images_.end(); ++it) {
1853     auto &image_id = it->first;
1854     auto &info = it->second;
1855     if (info.state == cls::rbd::MIRROR_IMAGE_STATE_DISABLED) {
1856       continue;
1857     }
1858
1859     auto &image_name = id_to_name[image_id];
1860     if (image_name.empty()) {
1861       lderr(cct) << "failed to find image name for image " << image_id << ", "
1862                  << "using image id as name" << dendl;
1863       image_name = image_id;
1864     }
1865
1866     mirror_image_global_status_t& global_status = (*images)[image_id];
1867     global_status.name = image_name;
1868     global_status.info = mirror_image_info_t{
1869         info.global_image_id,
1870         static_cast<mirror_image_state_t>(info.state),
1871         false}; // XXX: To set "primary" right would require an additional call.
1872
1873     auto s_it = statuses_.find(image_id);
1874     if (s_it != statuses_.end()) {
1875       auto& status = s_it->second;
1876
1877       global_status.site_statuses.reserve(
1878         status.mirror_image_site_statuses.size());
1879       for (auto& site_status : status.mirror_image_site_statuses) {
1880         global_status.site_statuses.push_back(mirror_image_site_status_t{
1881           site_status.mirror_uuid,
1882           static_cast<mirror_image_status_state_t>(site_status.state),
1883           site_status.state == cls::rbd::MIRROR_IMAGE_STATUS_STATE_UNKNOWN ?
1884             STATUS_NOT_FOUND : site_status.description,
1885           site_status.last_update.sec(), site_status.up});
1886       }
1887     } else {
1888       // older OSD that only returns local status
1889       global_status.site_statuses.push_back(mirror_image_site_status_t{
1890         cls::rbd::MirrorImageSiteStatus::LOCAL_MIRROR_UUID,
1891         MIRROR_IMAGE_STATUS_STATE_UNKNOWN, STATUS_NOT_FOUND, 0, false});
1892     }
1893   }
1894
1895   return 0;
1896 }
1897
1898 template <typename I>
1899 int Mirror<I>::image_status_summary(librados::IoCtx& io_ctx,
1900                                     MirrorImageStatusStates *states) {
1901   CephContext *cct = reinterpret_cast<CephContext *>(io_ctx.cct());
1902
1903   std::vector<cls::rbd::MirrorPeer> mirror_peers;
1904   int r = cls_client::mirror_peer_list(&io_ctx, &mirror_peers);
1905   if (r < 0 && r != -ENOENT) {
1906     lderr(cct) << "failed to list mirror peers: " << cpp_strerror(r) << dendl;
1907     return r;
1908   }
1909
1910   std::map<cls::rbd::MirrorImageStatusState, int> states_;
1911   r = cls_client::mirror_image_status_get_summary(&io_ctx, mirror_peers,
1912                                                   &states_);
1913   if (r < 0 && r != -ENOENT) {
1914     lderr(cct) << "failed to get mirror status summary: "
1915                << cpp_strerror(r) << dendl;
1916     return r;
1917   }
1918   for (auto &s : states_) {
1919     (*states)[static_cast<mirror_image_status_state_t>(s.first)] = s.second;
1920   }
1921   return 0;
1922 }
1923
1924 template <typename I>
1925 int Mirror<I>::image_instance_id_list(
1926     librados::IoCtx& io_ctx, const std::string &start_image_id, size_t max,
1927     std::map<std::string, std::string> *instance_ids) {
1928   CephContext *cct = reinterpret_cast<CephContext *>(io_ctx.cct());
1929   std::map<std::string, entity_inst_t> instances;
1930
1931   int r = librbd::cls_client::mirror_image_instance_list(
1932       &io_ctx, start_image_id, max, &instances);
1933   if (r < 0 && r != -ENOENT) {
1934     lderr(cct) << "failed to list mirror image instances: " << cpp_strerror(r)
1935                << dendl;
1936     return r;
1937   }
1938
1939   for (auto it : instances) {
1940     (*instance_ids)[it.first] = stringify(it.second.name.num());
1941   }
1942
1943   return 0;
1944 }
1945
1946 template <typename I>
1947 int Mirror<I>::image_info_list(
1948     librados::IoCtx& io_ctx, mirror_image_mode_t *mode_filter,
1949     const std::string &start_id, size_t max,
1950     std::map<std::string, std::pair<mirror_image_mode_t,
1951                                     mirror_image_info_t>> *entries) {
1952   CephContext *cct = reinterpret_cast<CephContext *>(io_ctx.cct());
1953   ldout(cct, 20) << "pool=" << io_ctx.get_pool_name() << ", mode_filter="
1954                  << (mode_filter ? stringify(*mode_filter) : "null")
1955                  << ", start_id=" << start_id << ", max=" << max << dendl;
1956
1957   std::string last_read = start_id;
1958   entries->clear();
1959
1960   while (entries->size() < max) {
1961     map<std::string, cls::rbd::MirrorImage> images;
1962     map<std::string, cls::rbd::MirrorImageStatus> statuses;
1963
1964     int r = librbd::cls_client::mirror_image_status_list(&io_ctx, last_read,
1965                                                          max, &images,
1966                                                          &statuses);
1967     if (r < 0 && r != -ENOENT) {
1968       lderr(cct) << "failed to list mirror image statuses: "
1969                  << cpp_strerror(r) << dendl;
1970       return r;
1971     }
1972
1973     if (images.empty()) {
1974       break;
1975     }
1976
1977     ThreadPool *thread_pool;
1978     ContextWQ *op_work_queue;
1979     ImageCtx::get_thread_pool_instance(cct, &thread_pool, &op_work_queue);
1980
1981     for (auto &it : images) {
1982       auto &image_id = it.first;
1983       auto &image = it.second;
1984       auto mode = static_cast<mirror_image_mode_t>(image.mode);
1985
1986       if ((mode_filter && mode != *mode_filter) ||
1987           image.state != cls::rbd::MIRROR_IMAGE_STATE_ENABLED) {
1988         continue;
1989       }
1990
1991       // need to call get_info for every image to retrieve promotion state
1992
1993       mirror_image_info_t info;
1994       r = image_get_info(io_ctx, op_work_queue, image_id, &info);
1995       if (r < 0) {
1996         continue;
1997       }
1998
1999       (*entries)[image_id] = std::make_pair(mode, info);
2000       if (entries->size() == max) {
2001         break;
2002       }
2003     }
2004
2005     last_read = images.rbegin()->first;
2006   }
2007
2008   return 0;
2009 }
2010
2011 template <typename I>
2012 int Mirror<I>::image_snapshot_create(I *ictx, uint32_t flags,
2013                                      uint64_t *snap_id) {
2014   C_SaferCond ctx;
2015   Mirror<I>::image_snapshot_create(ictx, flags, snap_id, &ctx);
2016
2017   return ctx.wait();
2018 }
2019
2020 template <typename I>
2021 void Mirror<I>::image_snapshot_create(I *ictx, uint32_t flags,
2022                                       uint64_t *snap_id, Context *on_finish) {
2023   CephContext *cct = ictx->cct;
2024   ldout(cct, 20) << "ictx=" << ictx << dendl;
2025
2026   if (flags != 0U) {
2027     lderr(cct) << "invalid snap create flags: " << std::bitset<32>(flags)
2028                << dendl;
2029     on_finish->complete(-EINVAL);
2030     return;
2031   }
2032
2033   auto on_refresh = new LambdaContext(
2034     [ictx, snap_id, on_finish](int r) {
2035       if (r < 0) {
2036         lderr(ictx->cct) << "refresh failed: " << cpp_strerror(r) << dendl;
2037         on_finish->complete(r);
2038         return;
2039       }
2040
2041       auto ctx = new C_ImageSnapshotCreate<I>(ictx, snap_id, on_finish);
2042       auto req = mirror::GetInfoRequest<I>::create(*ictx, &ctx->mirror_image,
2043                                                    &ctx->promotion_state,
2044                                                    &ctx->primary_mirror_uuid,
2045                                                    ctx);
2046       req->send();
2047     });
2048
2049   if (ictx->state->is_refresh_required()) {
2050     ictx->state->refresh(on_refresh);
2051   } else {
2052     on_refresh->complete(0);
2053   }
2054 }
2055
2056 } // namespace api
2057 } // namespace librbd
2058
2059 template class librbd::api::Mirror<librbd::ImageCtx>;