ceph/src/tools/rbd/action/MirrorPool.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 "tools/rbd/ArgumentTypes.h"
   5 #include "tools/rbd/MirrorDaemonServiceInfo.h"
   6 #include "tools/rbd/Shell.h"
   7 #include "tools/rbd/Utils.h"
   8 #include "include/buffer.h"
   9 #include "include/Context.h"
  10 #include "include/stringify.h"
  11 #include "include/rbd/librbd.hpp"
  12 #include "common/ceph_json.h"
  13 #include "common/config.h"
  14 #include "common/debug.h"
  15 #include "common/errno.h"
  16 #include "common/Formatter.h"
  17 #include "common/TextTable.h"
  18 #include "common/Throttle.h"
  19 #include "global/global_context.h"
  20 #include <fstream>
  21 #include <functional>
  22 #include <iostream>
  23 #include <regex>
  24 #include <set>
  25 #include <boost/program_options.hpp>
  26 #include "include/ceph_assert.h"
  27
  28 #include <atomic>
  29
  30 #define dout_context g_ceph_context
  31 #define dout_subsys ceph_subsys_rbd
  32 #undef dout_prefix
  33 #define dout_prefix *_dout << "rbd::action::MirrorPool: "
  34
  35 namespace rbd {
  36 namespace action {
  37 namespace mirror_pool {
  38
  39 namespace at = argument_types;
  40 namespace po = boost::program_options;
  41
  42 static const std::string ALL_NAME("all");
  43 static const std::string SITE_NAME("site-name");
  44
  45 namespace {
  46
  47 void add_site_name_optional(po::options_description *options) {
  48   options->add_options()
  49     (SITE_NAME.c_str(), po::value<std::string>(), "local site name");
  50 }
  51
  52 int set_site_name(librados::Rados& rados, const std::string& site_name) {
  53   librbd::RBD rbd;
  54   int r = rbd.mirror_site_name_set(rados, site_name);
  55   if (r == -EOPNOTSUPP) {
  56     std::cerr << "rbd: cluster does not support site names" << std::endl;
  57     return r;
  58   } else if (r < 0) {
  59     std::cerr << "rbd: failed to set site name" << cpp_strerror(r)
  60               << std::endl;
  61     return r;
  62   }
  63
  64   return 0;
  65 }
  66
  67 struct MirrorPeerDirection {};
  68
  69 void validate(boost::any& v, const std::vector<std::string>& values,
  70               MirrorPeerDirection *target_type, int permit_tx) {
  71   po::validators::check_first_occurrence(v);
  72   const std::string &s = po::validators::get_single_string(values);
  73
  74   if (s == "rx-only") {
  75     v = boost::any(RBD_MIRROR_PEER_DIRECTION_RX);
  76   } else if (s == "rx-tx") {
  77     v = boost::any(RBD_MIRROR_PEER_DIRECTION_RX_TX);
  78   } else if (permit_tx != 0 && s == "tx-only") {
  79     v = boost::any(RBD_MIRROR_PEER_DIRECTION_TX);
  80   } else {
  81     throw po::validation_error(po::validation_error::invalid_option_value);
  82   }
  83 }
  84
  85 void add_direction_optional(po::options_description *options) {
  86   options->add_options()
  87     ("direction", po::value<MirrorPeerDirection>(),
  88      "mirroring direction (rx-only, rx-tx)\n"
  89      "[default: rx-tx]");
  90 }
  91
  92 int validate_mirroring_enabled(librados::IoCtx& io_ctx) {
  93   librbd::RBD rbd;
  94   rbd_mirror_mode_t mirror_mode;
  95   int r = rbd.mirror_mode_get(io_ctx, &mirror_mode);
  96   if (r < 0) {
  97     std::cerr << "rbd: failed to retrieve mirror mode: "
  98               << cpp_strerror(r) << std::endl;
  99     return r;
 100   }
 101
 102   if (mirror_mode == RBD_MIRROR_MODE_DISABLED) {
 103     std::cerr << "rbd: mirroring not enabled on the pool" << std::endl;
 104     return -EINVAL;
 105   }
 106   return 0;
 107 }
 108
 109 int validate_uuid(const std::string &uuid) {
 110   std::regex pattern("^[A-F0-9]{8}-[A-F0-9]{4}-[A-F0-9]{4}-[A-F0-9]{4}-[A-F0-9]{12}$",
 111                        std::regex::icase);
 112   std::smatch match;
 113   if (!std::regex_match(uuid, match, pattern)) {
 114     std::cerr << "rbd: invalid uuid '" << uuid << "'" << std::endl;
 115     return -EINVAL;
 116   }
 117   return 0;
 118 }
 119
 120 int read_key_file(std::string path, std::string* key) {
 121   std::ifstream key_file;
 122   key_file.open(path);
 123   if (key_file.fail()) {
 124     std::cerr << "rbd: failed to open " << path << std::endl;
 125     return -EINVAL;
 126   }
 127
 128   std::getline(key_file, *key);
 129   if (key_file.bad()) {
 130     std::cerr << "rbd: failed to read key from " << path << std::endl;
 131     return -EINVAL;
 132   }
 133
 134   key_file.close();
 135   return 0;
 136 }
 137
 138 void add_uuid_option(po::options_description *positional) {
 139   positional->add_options()
 140     ("uuid", po::value<std::string>(), "peer uuid");
 141 }
 142
 143 int get_uuid(const po::variables_map &vm, size_t arg_index,
 144              std::string *uuid) {
 145   *uuid = utils::get_positional_argument(vm, arg_index);
 146   if (uuid->empty()) {
 147     std::cerr << "rbd: must specify peer uuid" << std::endl;
 148     return -EINVAL;
 149   }
 150   return validate_uuid(*uuid);
 151 }
 152
 153 int get_remote_cluster_spec(const po::variables_map &vm,
 154                             const std::string &spec,
 155                             std::string *remote_client_name,
 156                             std::string *remote_cluster,
 157                             std::map<std::string, std::string>* attributes) {
 158   if (vm.count("remote-client-name")) {
 159     *remote_client_name = vm["remote-client-name"].as<std::string>();
 160   }
 161   if (vm.count("remote-cluster")) {
 162     *remote_cluster = vm["remote-cluster"].as<std::string>();
 163   }
 164   if (vm.count("remote-mon-host")) {
 165     (*attributes)["mon_host"] = vm["remote-mon-host"].as<std::string>();
 166   }
 167   if (vm.count("remote-key-file")) {
 168     std::string key;
 169     int r = read_key_file(vm["remote-key-file"].as<std::string>(), &key);
 170     if (r < 0) {
 171       return r;
 172     }
 173     (*attributes)["key"] = key;
 174   }
 175
 176   if (!spec.empty()) {
 177     std::regex pattern("^(?:(client\\.[^@]+)@)?([^/@]+)$");
 178     std::smatch match;
 179     if (!std::regex_match(spec, match, pattern)) {
 180       std::cerr << "rbd: invalid spec '" << spec << "'" << std::endl;
 181       return -EINVAL;
 182     }
 183     if (match[1].matched) {
 184       *remote_client_name = match[1];
 185     }
 186     *remote_cluster = match[2];
 187   }
 188
 189   if (remote_cluster->empty()) {
 190     std::cerr << "rbd: remote cluster was not specified" << std::endl;
 191     return -EINVAL;
 192   }
 193   return 0;
 194 }
 195
 196 int set_peer_config_key(librados::IoCtx& io_ctx, const std::string& peer_uuid,
 197                         std::map<std::string, std::string>&& attributes) {
 198   librbd::RBD rbd;
 199   int r = rbd.mirror_peer_site_set_attributes(io_ctx, peer_uuid, attributes);
 200   if (r == -EPERM) {
 201     std::cerr << "rbd: permission denied attempting to set peer "
 202               << "config-key secrets in the monitor" << std::endl;
 203     return r;
 204   } else if (r < 0) {
 205     std::cerr << "rbd: failed to update mirroring peer config: "
 206               << cpp_strerror(r) << std::endl;
 207     return r;
 208   }
 209   return 0;
 210 }
 211
 212 int get_peer_config_key(librados::IoCtx& io_ctx, const std::string& peer_uuid,
 213                         std::map<std::string, std::string>* attributes) {
 214   librbd::RBD rbd;
 215   int r = rbd.mirror_peer_site_get_attributes(io_ctx, peer_uuid, attributes);
 216   if (r == -ENOENT) {
 217     return r;
 218   } else if (r == -EPERM) {
 219     std::cerr << "rbd: permission denied attempting to access peer "
 220               << "config-key secrets from the monitor" << std::endl;
 221     return r;
 222   } else if (r == -EINVAL) {
 223     std::cerr << "rbd: corrupt mirroring peer config" << std::endl;
 224     return r;
 225   } else if (r < 0) {
 226     std::cerr << "rbd: error reading mirroring peer config: "
 227               << cpp_strerror(r) << std::endl;
 228     return r;
 229   }
 230
 231   return 0;
 232 }
 233
 234 int update_peer_config_key(librados::IoCtx& io_ctx,
 235                            const std::string& peer_uuid,
 236                            const std::string& key,
 237                            const std::string& value) {
 238   std::map<std::string, std::string> attributes;
 239   int r = get_peer_config_key(io_ctx, peer_uuid, &attributes);
 240   if (r == -ENOENT) {
 241     return set_peer_config_key(io_ctx, peer_uuid, {{key, value}});
 242   } else if (r < 0) {
 243     return r;
 244   }
 245
 246   if (value.empty()) {
 247     attributes.erase(key);
 248   } else {
 249     attributes[key] = value;
 250   }
 251   return set_peer_config_key(io_ctx, peer_uuid, std::move(attributes));
 252 }
 253
 254 int format_mirror_peers(librados::IoCtx& io_ctx,
 255                         at::Format::Formatter formatter,
 256                         const std::vector<librbd::mirror_peer_site_t> &peers,
 257                         bool config_key) {
 258   if (formatter != nullptr) {
 259     formatter->open_array_section("peers");
 260   } else {
 261     std::cout <<  "Peer Sites: ";
 262     if (peers.empty()) {
 263       std::cout << "none";
 264     }
 265     std::cout << std::endl;
 266   }
 267
 268   for (auto &peer : peers) {
 269     std::map<std::string, std::string> attributes;
 270     if (config_key) {
 271       int r = get_peer_config_key(io_ctx, peer.uuid, &attributes);
 272       if (r < 0 && r != -ENOENT) {
 273         return r;
 274       }
 275     }
 276
 277     std::string direction;
 278     switch (peer.direction) {
 279     case RBD_MIRROR_PEER_DIRECTION_RX:
 280       direction = "rx-only";
 281       break;
 282     case RBD_MIRROR_PEER_DIRECTION_TX:
 283       direction = "tx-only";
 284       break;
 285     case RBD_MIRROR_PEER_DIRECTION_RX_TX:
 286       direction = "rx-tx";
 287       break;
 288     default:
 289       direction = "unknown";
 290       break;
 291     }
 292
 293     if (formatter != nullptr) {
 294       formatter->open_object_section("peer");
 295       formatter->dump_string("uuid", peer.uuid);
 296       formatter->dump_string("direction", direction);
 297       formatter->dump_string("site_name", peer.site_name);
 298       formatter->dump_string("mirror_uuid", peer.mirror_uuid);
 299       formatter->dump_string("client_name", peer.client_name);
 300       for (auto& pair : attributes) {
 301         formatter->dump_string(pair.first.c_str(), pair.second);
 302       }
 303       formatter->close_section();
 304     } else {
 305       std::cout << std::endl
 306                 << "UUID: " << peer.uuid << std::endl
 307                 << "Name: " << peer.site_name << std::endl;
 308       if (peer.direction != RBD_MIRROR_PEER_DIRECTION_RX ||
 309           !peer.mirror_uuid.empty()) {
 310         std::cout << "Mirror UUID: " << peer.mirror_uuid << std::endl;
 311       }
 312       std::cout << "Direction: " << direction << std::endl;
 313       if (peer.direction != RBD_MIRROR_PEER_DIRECTION_TX ||
 314           !peer.client_name.empty()) {
 315         std::cout << "Client: " << peer.client_name << std::endl;
 316       }
 317       if (config_key) {
 318         std::cout << "Mon Host: " << attributes["mon_host"] << std::endl
 319                   << "Key: " << attributes["key"] << std::endl;
 320       }
 321       if (peer.site_name != peers.rbegin()->site_name) {
 322         std::cout << std::endl;
 323       }
 324     }
 325   }
 326
 327   if (formatter != nullptr) {
 328     formatter->close_section();
 329   }
 330   return 0;
 331 }
 332
 333 class ImageRequestBase {
 334 public:
 335   void send() {
 336     dout(20) << this << " " << __func__ << ": image_name=" << m_image_name
 337              << dendl;
 338
 339     auto ctx = new LambdaContext([this](int r) {
 340         handle_finalize(r);
 341       });
 342
 343     // will pause here until slots are available
 344     m_finalize_ctx = m_throttle.start_op(ctx);
 345
 346     open_image();
 347   }
 348
 349 protected:
 350   ImageRequestBase(librados::IoCtx &io_ctx, OrderedThrottle &throttle,
 351                    const std::string &image_name)
 352     : m_io_ctx(io_ctx), m_throttle(throttle), m_image_name(image_name) {
 353   }
 354   virtual ~ImageRequestBase() {
 355   }
 356
 357   virtual bool skip_get_info() const {
 358     return false;
 359   }
 360   virtual void get_info(librbd::Image &image, librbd::mirror_image_info_t *info,
 361                         librbd::RBD::AioCompletion *aio_comp) {
 362     image.aio_mirror_image_get_info(info, sizeof(librbd::mirror_image_info_t),
 363                                     aio_comp);
 364   }
 365
 366   virtual bool skip_action(const librbd::mirror_image_info_t &info) const {
 367     return false;
 368   }
 369   virtual void execute_action(librbd::Image &image,
 370                               librbd::RBD::AioCompletion *aio_comp) = 0;
 371   virtual void handle_execute_action(int r) {
 372     dout(20) << this << " " << __func__ << ": r=" << r << dendl;
 373
 374     if (r < 0 && r != -ENOENT) {
 375       std::cerr << "rbd: failed to " << get_action_type() << " image "
 376                 << m_image_name << ": " << cpp_strerror(r) << std::endl;
 377       m_ret_val = r;
 378     }
 379
 380     close_image();
 381   }
 382
 383   virtual void finalize_action() {
 384   }
 385   virtual std::string get_action_type() const = 0;
 386
 387 private:
 388   /**
 389    * @verbatim
 390    *
 391    * <start>
 392    *    |
 393    *    v
 394    * OPEN_IMAGE
 395    *    |
 396    *    v
 397    * GET_INFO
 398    *    |
 399    *    v
 400    * EXECUTE_ACTION
 401    *    |
 402    *    v
 403    * CLOSE_IMAGE
 404    *    |
 405    *    v
 406    * FINALIZE_ACTION
 407    *    |
 408    *    v
 409    * <finish>
 410    *
 411    * @endverbatim
 412    */
 413
 414   librados::IoCtx &m_io_ctx;
 415   OrderedThrottle &m_throttle;
 416   const std::string m_image_name;
 417
 418   librbd::Image m_image;
 419   Context *m_finalize_ctx = nullptr;
 420
 421   librbd::mirror_image_info_t m_mirror_image_info;
 422
 423   int m_ret_val = 0;
 424
 425   void open_image() {
 426     dout(20) << this << " " << __func__ << dendl;
 427
 428     librbd::RBD rbd;
 429     auto aio_completion = utils::create_aio_completion<
 430       ImageRequestBase, &ImageRequestBase::handle_open_image>(this);
 431     rbd.aio_open(m_io_ctx, m_image, m_image_name.c_str(), nullptr,
 432                  aio_completion);
 433   }
 434
 435   void handle_open_image(int r) {
 436     dout(20) << this << " " << __func__ << ": r=" << r << dendl;
 437
 438     if (r < 0) {
 439       std::cerr << "rbd: failed to open image "
 440                 << m_image_name << ": " << cpp_strerror(r) << std::endl;
 441       m_finalize_ctx->complete(r);
 442       return;
 443     }
 444
 445     get_info();
 446   }
 447
 448   void get_info() {
 449     if (skip_get_info()) {
 450       execute_action();
 451       return;
 452     }
 453     dout(20) << this << " " << __func__ << dendl;
 454
 455     auto aio_completion = utils::create_aio_completion<
 456       ImageRequestBase, &ImageRequestBase::handle_get_info>(this);
 457     get_info(m_image, &m_mirror_image_info, aio_completion);
 458   }
 459
 460   void handle_get_info(int r) {
 461     dout(20) << this << " " << __func__ << ": r=" << r << dendl;
 462
 463     if (r == -ENOENT) {
 464       close_image();
 465       return;
 466     } else if (r < 0) {
 467       std::cerr << "rbd: failed to retrieve mirror image info for "
 468                 << m_image_name << ": " << cpp_strerror(r) << std::endl;
 469       m_ret_val = r;
 470       close_image();
 471       return;
 472     }
 473
 474     execute_action();
 475   }
 476
 477   void execute_action() {
 478     if (skip_action(m_mirror_image_info)) {
 479       close_image();
 480       return;
 481     }
 482     dout(20) << this << " " << __func__ << dendl;
 483
 484     auto aio_completion = utils::create_aio_completion<
 485       ImageRequestBase, &ImageRequestBase::handle_execute_action>(this);
 486     execute_action(m_image, aio_completion);
 487   }
 488
 489   void close_image() {
 490     dout(20) << this << " " << __func__ << dendl;
 491
 492     auto aio_completion = utils::create_aio_completion<
 493       ImageRequestBase, &ImageRequestBase::handle_close_image>(this);
 494     m_image.aio_close(aio_completion);
 495   }
 496
 497   void handle_close_image(int r) {
 498     dout(20) << this << " " << __func__ << ": r=" << r << dendl;
 499
 500     if (r < 0) {
 501       std::cerr << "rbd: failed to close image "
 502                 << m_image_name << ": " << cpp_strerror(r) << std::endl;
 503     }
 504
 505     m_finalize_ctx->complete(r);
 506   }
 507
 508   void handle_finalize(int r) {
 509     dout(20) << this << " " << __func__ << ": r=" << r << dendl;
 510
 511     if (r == 0 && m_ret_val < 0) {
 512       r = m_ret_val;
 513     }
 514     if (r >= 0) {
 515       finalize_action();
 516     }
 517     m_throttle.end_op(r);
 518     delete this;
 519   }
 520
 521 };
 522
 523 class PromoteImageRequest : public ImageRequestBase {
 524 public:
 525   PromoteImageRequest(librados::IoCtx &io_ctx, OrderedThrottle &throttle,
 526                       const std::string &image_name, std::atomic<unsigned> *counter,
 527                       bool force)
 528     : ImageRequestBase(io_ctx, throttle, image_name), m_counter(counter),
 529       m_force(force) {
 530   }
 531
 532 protected:
 533   bool skip_action(const librbd::mirror_image_info_t &info) const override {
 534     return (info.state != RBD_MIRROR_IMAGE_ENABLED || info.primary);
 535   }
 536
 537   void execute_action(librbd::Image &image,
 538                       librbd::RBD::AioCompletion *aio_comp) override {
 539     image.aio_mirror_image_promote(m_force, aio_comp);
 540   }
 541
 542   void handle_execute_action(int r) override {
 543     if (r >= 0) {
 544       (*m_counter)++;
 545     }
 546     ImageRequestBase::handle_execute_action(r);
 547   }
 548
 549   std::string get_action_type() const override {
 550     return "promote";
 551   }
 552
 553 private:
 554   std::atomic<unsigned> *m_counter = nullptr;
 555   bool m_force;
 556 };
 557
 558 class DemoteImageRequest : public ImageRequestBase {
 559 public:
 560   DemoteImageRequest(librados::IoCtx &io_ctx, OrderedThrottle &throttle,
 561                      const std::string &image_name, std::atomic<unsigned> *counter)
 562     : ImageRequestBase(io_ctx, throttle, image_name), m_counter(counter) {
 563   }
 564
 565 protected:
 566   bool skip_action(const librbd::mirror_image_info_t &info) const override {
 567     return (info.state != RBD_MIRROR_IMAGE_ENABLED || !info.primary);
 568   }
 569
 570   void execute_action(librbd::Image &image,
 571                       librbd::RBD::AioCompletion *aio_comp) override {
 572     image.aio_mirror_image_demote(aio_comp);
 573   }
 574   void handle_execute_action(int r) override {
 575     if (r >= 0) {
 576       (*m_counter)++;
 577     }
 578     ImageRequestBase::handle_execute_action(r);
 579   }
 580
 581   std::string get_action_type() const override {
 582     return "demote";
 583   }
 584
 585 private:
 586   std::atomic<unsigned> *m_counter = nullptr;
 587 };
 588
 589 class StatusImageRequest : public ImageRequestBase {
 590 public:
 591   StatusImageRequest(
 592       librados::IoCtx &io_ctx, OrderedThrottle &throttle,
 593       const std::string &image_name,
 594       const std::map<std::string, std::string> &instance_ids,
 595       const std::vector<librbd::mirror_peer_site_t>& mirror_peers,
 596       const std::map<std::string, std::string> &peer_mirror_uuids_to_name,
 597       const MirrorDaemonServiceInfo &daemon_service_info,
 598       at::Format::Formatter formatter)
 599     : ImageRequestBase(io_ctx, throttle, image_name),
 600       m_instance_ids(instance_ids), m_mirror_peers(mirror_peers),
 601       m_peer_mirror_uuids_to_name(peer_mirror_uuids_to_name),
 602       m_daemon_service_info(daemon_service_info), m_formatter(formatter) {
 603   }
 604
 605 protected:
 606   bool skip_get_info() const override {
 607     return true;
 608   }
 609
 610   void execute_action(librbd::Image &image,
 611                       librbd::RBD::AioCompletion *aio_comp) override {
 612     image.get_id(&m_image_id);
 613     image.aio_mirror_image_get_global_status(
 614       &m_mirror_image_global_status, sizeof(m_mirror_image_global_status),
 615       aio_comp);
 616   }
 617
 618   void finalize_action() override {
 619     if (m_mirror_image_global_status.info.global_id.empty()) {
 620       return;
 621     }
 622
 623     utils::populate_unknown_mirror_image_site_statuses(
 624       m_mirror_peers, &m_mirror_image_global_status);
 625
 626     librbd::mirror_image_site_status_t local_status;
 627     int local_site_r = utils::get_local_mirror_image_status(
 628       m_mirror_image_global_status, &local_status);
 629     m_mirror_image_global_status.site_statuses.erase(
 630       std::remove_if(m_mirror_image_global_status.site_statuses.begin(),
 631                      m_mirror_image_global_status.site_statuses.end(),
 632                      [](auto& status) {
 633           return (status.mirror_uuid ==
 634                     RBD_MIRROR_IMAGE_STATUS_LOCAL_MIRROR_UUID);
 635         }),
 636       m_mirror_image_global_status.site_statuses.end());
 637
 638     std::string instance_id = (local_site_r >= 0 && local_status.up &&
 639                                m_instance_ids.count(m_image_id)) ?
 640         m_instance_ids.find(m_image_id)->second : "";
 641
 642     auto mirror_service = m_daemon_service_info.get_by_instance_id(instance_id);
 643     if (m_formatter != nullptr) {
 644       m_formatter->open_object_section("image");
 645       m_formatter->dump_string("name", m_mirror_image_global_status.name);
 646       m_formatter->dump_string(
 647         "global_id", m_mirror_image_global_status.info.global_id);
 648       if (local_site_r >= 0) {
 649         m_formatter->dump_string("state", utils::mirror_image_site_status_state(
 650           local_status));
 651         m_formatter->dump_string("description", local_status.description);
 652         if (mirror_service != nullptr) {
 653           mirror_service->dump_image(m_formatter);
 654         }
 655         m_formatter->dump_string("last_update", utils::timestr(
 656           local_status.last_update));
 657       }
 658       if (!m_mirror_image_global_status.site_statuses.empty()) {
 659         m_formatter->open_array_section("peer_sites");
 660         for (auto& status : m_mirror_image_global_status.site_statuses) {
 661           m_formatter->open_object_section("peer_site");
 662
 663           auto name_it = m_peer_mirror_uuids_to_name.find(status.mirror_uuid);
 664           m_formatter->dump_string("site_name",
 665             (name_it != m_peer_mirror_uuids_to_name.end() ?
 666                name_it->second : ""));
 667           m_formatter->dump_string("mirror_uuids", status.mirror_uuid);
 668
 669           m_formatter->dump_string(
 670             "state", utils::mirror_image_site_status_state(status));
 671           m_formatter->dump_string("description", status.description);
 672           m_formatter->dump_string("last_update", utils::timestr(
 673             status.last_update));
 674           m_formatter->close_section(); // peer_site
 675         }
 676         m_formatter->close_section(); // peer_sites
 677       }
 678       m_formatter->close_section(); // image
 679     } else {
 680       std::cout << std::endl
 681                 << m_mirror_image_global_status.name << ":" << std::endl
 682                 << "  global_id:   "
 683                 << m_mirror_image_global_status.info.global_id << std::endl;
 684       if (local_site_r >= 0) {
 685         std::cout << "  state:       " << utils::mirror_image_site_status_state(
 686                     local_status) << std::endl
 687                   << "  description: " << local_status.description << std::endl;
 688         if (mirror_service != nullptr) {
 689           std::cout << "  service:     " <<
 690             mirror_service->get_image_description() << std::endl;
 691         }
 692         std::cout << "  last_update: " << utils::timestr(
 693           local_status.last_update) << std::endl;
 694       }
 695       if (!m_mirror_image_global_status.site_statuses.empty()) {
 696         std::cout << "  peer_sites:" << std::endl;
 697         bool first_site = true;
 698         for (auto& site : m_mirror_image_global_status.site_statuses) {
 699           if (!first_site) {
 700             std::cout << std::endl;
 701           }
 702           first_site = false;
 703
 704           auto name_it = m_peer_mirror_uuids_to_name.find(site.mirror_uuid);
 705           std::cout << "    name: "
 706                     << (name_it != m_peer_mirror_uuids_to_name.end() ?
 707                           name_it->second : site.mirror_uuid)
 708                     << std::endl
 709                     << "    state: " << utils::mirror_image_site_status_state(
 710                       site) << std::endl
 711                     << "    description: " << site.description << std::endl
 712                     << "    last_update: " << utils::timestr(
 713                       site.last_update) << std::endl;
 714         }
 715       }
 716     }
 717   }
 718
 719   std::string get_action_type() const override {
 720     return "status";
 721   }
 722
 723 private:
 724   const std::map<std::string, std::string> &m_instance_ids;
 725   const std::vector<librbd::mirror_peer_site_t> &m_mirror_peers;
 726   const std::map<std::string, std::string> &m_peer_mirror_uuids_to_name;
 727   const MirrorDaemonServiceInfo &m_daemon_service_info;
 728   at::Format::Formatter m_formatter;
 729   std::string m_image_id;
 730   librbd::mirror_image_global_status_t m_mirror_image_global_status;
 731 };
 732
 733 template <typename RequestT>
 734 class ImageRequestAllocator {
 735 public:
 736   template <class... Args>
 737   RequestT *operator()(librados::IoCtx &io_ctx, OrderedThrottle &throttle,
 738                        const std::string &image_name, Args&&... args) {
 739     return new RequestT(io_ctx, throttle, image_name,
 740                         std::forward<Args>(args)...);
 741   }
 742 };
 743
 744 template <typename RequestT>
 745 class ImageRequestGenerator {
 746 public:
 747   template <class... Args>
 748   ImageRequestGenerator(librados::IoCtx &io_ctx, Args&&... args)
 749     : m_io_ctx(io_ctx),
 750       m_factory(std::bind(ImageRequestAllocator<RequestT>(),
 751                           std::ref(m_io_ctx), std::ref(m_throttle),
 752                           std::placeholders::_1, std::forward<Args>(args)...)),
 753       m_throttle(g_conf().get_val<uint64_t>("rbd_concurrent_management_ops"),
 754                  true) {
 755   }
 756
 757   int execute() {
 758     // use the alphabetical list of image names for pool-level
 759     // mirror image operations
 760     librbd::RBD rbd;
 761     int r = rbd.list2(m_io_ctx, &m_images);
 762     if (r < 0 && r != -ENOENT) {
 763       std::cerr << "rbd: failed to list images within pool" << std::endl;
 764       return r;
 765     }
 766
 767     for (auto &image : m_images) {
 768       auto request = m_factory(image.name);
 769       request->send();
 770     }
 771
 772     return m_throttle.wait_for_ret();
 773   }
 774 private:
 775   typedef std::function<RequestT*(const std::string&)>  Factory;
 776
 777   librados::IoCtx &m_io_ctx;
 778   Factory m_factory;
 779
 780   OrderedThrottle m_throttle;
 781
 782   std::vector<librbd::image_spec_t> m_images;
 783
 784 };
 785
 786 int get_mirror_image_status(
 787     librados::IoCtx& io_ctx, uint32_t* total_images,
 788     std::map<librbd::mirror_image_status_state_t, int>* mirror_image_states,
 789     MirrorHealth* mirror_image_health) {
 790   librbd::RBD rbd;
 791   int r = rbd.mirror_image_status_summary(io_ctx, mirror_image_states);
 792   if (r < 0) {
 793     std::cerr << "rbd: failed to get status summary for mirrored images: "
 794               << cpp_strerror(r) << std::endl;
 795     return r;
 796   }
 797
 798   *mirror_image_health = MIRROR_HEALTH_OK;
 799   for (auto &it : *mirror_image_states) {
 800     auto &state = it.first;
 801     if (*mirror_image_health < MIRROR_HEALTH_WARNING &&
 802         (state != MIRROR_IMAGE_STATUS_STATE_REPLAYING &&
 803          state != MIRROR_IMAGE_STATUS_STATE_STOPPED)) {
 804       *mirror_image_health = MIRROR_HEALTH_WARNING;
 805     }
 806     if (*mirror_image_health < MIRROR_HEALTH_ERROR &&
 807         state == MIRROR_IMAGE_STATUS_STATE_ERROR) {
 808       *mirror_image_health = MIRROR_HEALTH_ERROR;
 809     }
 810     *total_images += it.second;
 811   }
 812
 813   return 0;
 814 }
 815
 816 } // anonymous namespace
 817
 818 void get_peer_bootstrap_create_arguments(po::options_description *positional,
 819                                          po::options_description *options) {
 820   at::add_pool_options(positional, options, false);
 821   options->add_options()
 822     (SITE_NAME.c_str(), po::value<std::string>(), "local site name");
 823 }
 824
 825 int execute_peer_bootstrap_create(
 826     const po::variables_map &vm,
 827     const std::vector<std::string> &ceph_global_init_args) {
 828   std::string pool_name;
 829   size_t arg_index = 0;
 830   int r = utils::get_pool_and_namespace_names(vm, true, true, &pool_name,
 831                                               nullptr, &arg_index);
 832   if (r < 0) {
 833     return r;
 834   }
 835
 836   librados::Rados rados;
 837   librados::IoCtx io_ctx;
 838   r = utils::init(pool_name, "", &rados, &io_ctx);
 839   if (r < 0) {
 840     return r;
 841   }
 842
 843   r = validate_mirroring_enabled(io_ctx);
 844   if (r < 0) {
 845     return r;
 846   }
 847
 848   if (vm.count(SITE_NAME)) {
 849     r = set_site_name(rados, vm[SITE_NAME].as<std::string>());
 850     if (r < 0) {
 851       return r;
 852     }
 853   }
 854
 855   librbd::RBD rbd;
 856   std::string token;
 857   r = rbd.mirror_peer_bootstrap_create(io_ctx, &token);
 858   if (r == -EEXIST) {
 859     std::cerr << "rbd: mismatch with pre-existing RBD mirroring peer user caps"
 860               << std::endl;
 861   } else if (r < 0) {
 862     std::cerr << "rbd: failed to create mirroring bootstrap token: "
 863               << cpp_strerror(r) << std::endl;
 864     return r;
 865   }
 866
 867   std::cout << token << std::endl;
 868   return 0;
 869 }
 870
 871 void get_peer_bootstrap_import_arguments(po::options_description *positional,
 872                                          po::options_description *options) {
 873   at::add_pool_options(positional, options, false);
 874   options->add_options()
 875     (SITE_NAME.c_str(), po::value<std::string>(), "local site name");
 876   positional->add_options()
 877     ("token-path", po::value<std::string>(),
 878      "bootstrap token file (or '-' for stdin)");
 879   options->add_options()
 880     ("token-path", po::value<std::string>(),
 881      "bootstrap token file (or '-' for stdin)");
 882   add_direction_optional(options);
 883 }
 884
 885 int execute_peer_bootstrap_import(
 886     const po::variables_map &vm,
 887     const std::vector<std::string> &ceph_global_init_args) {
 888   std::string pool_name;
 889   size_t arg_index = 0;
 890   int r = utils::get_pool_and_namespace_names(vm, true, true, &pool_name,
 891                                               nullptr, &arg_index);
 892   if (r < 0) {
 893     return r;
 894   }
 895
 896   std::string token_path;
 897   if (vm.count("token-path")) {
 898     token_path = vm["token-path"].as<std::string>();
 899   } else {
 900     token_path = utils::get_positional_argument(vm, arg_index++);
 901   }
 902
 903   if (token_path.empty()) {
 904     std::cerr << "rbd: token path was not specified" << std::endl;
 905     return -EINVAL;
 906   }
 907
 908   rbd_mirror_peer_direction_t mirror_peer_direction =
 909     RBD_MIRROR_PEER_DIRECTION_RX_TX;
 910   if (vm.count("direction")) {
 911     mirror_peer_direction = vm["direction"].as<rbd_mirror_peer_direction_t>();
 912   }
 913
 914   int fd = STDIN_FILENO;
 915   if (token_path != "-") {
 916     fd = open(token_path.c_str(), O_RDONLY);
 917     if (fd < 0) {
 918       r = -errno;
 919       std::cerr << "rbd: error opening " << token_path << std::endl;
 920       return r;
 921     }
 922   }
 923
 924   char token[1024];
 925   memset(token, 0, sizeof(token));
 926   r = safe_read(fd, token, sizeof(token) - 1);
 927   if (fd != STDIN_FILENO) {
 928     VOID_TEMP_FAILURE_RETRY(close(fd));
 929   }
 930
 931   if (r < 0) {
 932     std::cerr << "rbd: error reading token file: " << cpp_strerror(r)
 933               << std::endl;
 934     return r;
 935   }
 936
 937   librados::Rados rados;
 938   librados::IoCtx io_ctx;
 939   r = utils::init(pool_name, "", &rados, &io_ctx);
 940   if (r < 0) {
 941     return r;
 942   }
 943
 944   if (vm.count(SITE_NAME)) {
 945     r = set_site_name(rados, vm[SITE_NAME].as<std::string>());
 946     if (r < 0) {
 947       return r;
 948     }
 949   }
 950
 951   librbd::RBD rbd;
 952   r = rbd.mirror_peer_bootstrap_import(io_ctx, mirror_peer_direction, token);
 953   if (r == -ENOSYS) {
 954     std::cerr << "rbd: mirroring is not enabled on remote peer" << std::endl;
 955     return r;
 956   } else if (r < 0) {
 957     std::cerr << "rbd: failed to import peer bootstrap token" << std::endl;
 958     return r;
 959   }
 960
 961   return 0;
 962 }
 963
 964 void get_peer_add_arguments(po::options_description *positional,
 965                             po::options_description *options) {
 966   at::add_pool_options(positional, options, false);
 967   positional->add_options()
 968     ("remote-cluster-spec", "remote cluster spec\n"
 969      "(example: [<client name>@]<cluster name>)");
 970   options->add_options()
 971     ("remote-client-name", po::value<std::string>(), "remote client name")
 972     ("remote-cluster", po::value<std::string>(), "remote cluster name")
 973     ("remote-mon-host", po::value<std::string>(), "remote mon host(s)")
 974     ("remote-key-file", po::value<std::string>(),
 975      "path to file containing remote key");
 976   add_direction_optional(options);
 977 }
 978
 979 int execute_peer_add(const po::variables_map &vm,
 980                      const std::vector<std::string> &ceph_global_init_args) {
 981   std::string pool_name;
 982   size_t arg_index = 0;
 983   int r = utils::get_pool_and_namespace_names(vm, true, true, &pool_name,
 984                                               nullptr, &arg_index);
 985   if (r < 0) {
 986     return r;
 987   }
 988
 989   std::string remote_client_name = g_ceph_context->_conf->name.to_str();
 990   std::string remote_cluster;
 991   std::map<std::string, std::string> attributes;
 992   r = get_remote_cluster_spec(
 993     vm, utils::get_positional_argument(vm, arg_index),
 994     &remote_client_name, &remote_cluster, &attributes);
 995   if (r < 0) {
 996     return r;
 997   }
 998
 999   librados::Rados rados;
1000   librados::IoCtx io_ctx;
1001   r = utils::init(pool_name, "", &rados, &io_ctx);
1002   if (r < 0) {
1003     return r;
1004   }
1005
1006   r = validate_mirroring_enabled(io_ctx);
1007   if (r < 0) {
1008     return r;
1009   }
1010
1011   // TODO: temporary restriction to prevent adding multiple peers
1012   // until rbd-mirror daemon can properly handle the scenario
1013   librbd::RBD rbd;
1014   std::vector<librbd::mirror_peer_site_t> mirror_peers;
1015   r = rbd.mirror_peer_site_list(io_ctx, &mirror_peers);
1016   if (r < 0) {
1017     std::cerr << "rbd: failed to list mirror peers" << std::endl;
1018     return r;
1019   }
1020   if (!mirror_peers.empty()) {
1021     std::cerr << "rbd: multiple peers are not currently supported" << std::endl;
1022     return -EINVAL;
1023   }
1024
1025   rbd_mirror_peer_direction_t mirror_peer_direction =
1026     RBD_MIRROR_PEER_DIRECTION_RX_TX;
1027   if (vm.count("direction")) {
1028     mirror_peer_direction = vm["direction"].as<rbd_mirror_peer_direction_t>();
1029   }
1030
1031   std::string uuid;
1032   r = rbd.mirror_peer_site_add(
1033     io_ctx, &uuid, mirror_peer_direction, remote_cluster, remote_client_name);
1034   if (r < 0) {
1035     std::cerr << "rbd: error adding mirror peer" << std::endl;
1036     return r;
1037   }
1038
1039   if (!attributes.empty()) {
1040     r = set_peer_config_key(io_ctx, uuid, std::move(attributes));
1041     if (r < 0) {
1042       return r;
1043     }
1044   }
1045
1046   std::cout << uuid << std::endl;
1047   return 0;
1048 }
1049
1050 void get_peer_remove_arguments(po::options_description *positional,
1051                                po::options_description *options) {
1052   at::add_pool_options(positional, options, false);
1053   add_uuid_option(positional);
1054 }
1055
1056 int execute_peer_remove(const po::variables_map &vm,
1057                         const std::vector<std::string> &ceph_global_init_args) {
1058   std::string pool_name;
1059   size_t arg_index = 0;
1060   int r = utils::get_pool_and_namespace_names(vm, true, true, &pool_name,
1061                                               nullptr, &arg_index);
1062   if (r < 0) {
1063     return r;
1064   }
1065
1066   std::string uuid;
1067   r = get_uuid(vm, arg_index, &uuid);
1068   if (r < 0) {
1069     return r;
1070   }
1071
1072   librados::Rados rados;
1073   librados::IoCtx io_ctx;
1074   r = utils::init(pool_name, "", &rados, &io_ctx);
1075   if (r < 0) {
1076     return r;
1077   }
1078
1079   r = validate_mirroring_enabled(io_ctx);
1080   if (r < 0) {
1081     return r;
1082   }
1083
1084   librbd::RBD rbd;
1085   r = rbd.mirror_peer_site_remove(io_ctx, uuid);
1086   if (r < 0) {
1087     std::cerr << "rbd: error removing mirror peer" << std::endl;
1088     return r;
1089   }
1090   return 0;
1091 }
1092
1093 void get_peer_set_arguments(po::options_description *positional,
1094                             po::options_description *options) {
1095   at::add_pool_options(positional, options, false);
1096   add_uuid_option(positional);
1097   positional->add_options()
1098     ("key", "peer parameter\n"
1099             "(direction, site-name, client, mon-host, key-file)")
1100     ("value", "new value for specified key\n"
1101               "(rx-only, tx-only, or rx-tx for direction)");
1102 }
1103
1104 int execute_peer_set(const po::variables_map &vm,
1105                      const std::vector<std::string> &ceph_global_init_args) {
1106   std::string pool_name;
1107   size_t arg_index = 0;
1108   int r = utils::get_pool_and_namespace_names(vm, true, true, &pool_name,
1109                                               nullptr, &arg_index);
1110   if (r < 0) {
1111     return r;
1112   }
1113
1114   std::string uuid;
1115   r = get_uuid(vm, arg_index++, &uuid);
1116   if (r < 0) {
1117     return r;
1118   }
1119
1120   std::set<std::string> valid_keys{{"direction", "site-name", "cluster",
1121                                     "client", "mon-host", "key-file"}};
1122   std::string key = utils::get_positional_argument(vm, arg_index++);
1123   if (valid_keys.find(key) == valid_keys.end()) {
1124     std::cerr << "rbd: must specify ";
1125     for (auto& valid_key : valid_keys) {
1126       std::cerr << "'" << valid_key << "'";
1127       if (&valid_key != &(*valid_keys.rbegin())) {
1128         std::cerr << ", ";
1129       }
1130     }
1131     std::cerr <<  " key." << std::endl;
1132     return -EINVAL;
1133   }
1134
1135   std::string value = utils::get_positional_argument(vm, arg_index++);
1136   if (value.empty() && (key == "client" || key == "cluster")) {
1137     std::cerr << "rbd: must specify new " << key << " value." << std::endl;
1138   } else if (key == "key-file") {
1139     key = "key";
1140     r = read_key_file(value, &value);
1141     if (r < 0) {
1142       return r;
1143     }
1144   } else if (key == "mon-host") {
1145     key = "mon_host";
1146   }
1147
1148   librados::Rados rados;
1149   librados::IoCtx io_ctx;
1150   r = utils::init(pool_name, "", &rados, &io_ctx);
1151   if (r < 0) {
1152     return r;
1153   }
1154
1155   r = validate_mirroring_enabled(io_ctx);
1156   if (r < 0) {
1157     return r;
1158   }
1159
1160   librbd::RBD rbd;
1161   if (key == "client") {
1162     r = rbd.mirror_peer_site_set_client_name(io_ctx, uuid.c_str(),
1163                                              value.c_str());
1164   } else if (key == "site-name" || key == "cluster") {
1165     r = rbd.mirror_peer_site_set_name(io_ctx, uuid.c_str(), value.c_str());
1166   } else if (key == "direction") {
1167     MirrorPeerDirection tag;
1168     boost::any direction;
1169     try {
1170       validate(direction, {value}, &tag, 1);
1171     } catch (...) {
1172       std::cerr << "rbd: invalid direction" << std::endl;
1173       return -EINVAL;
1174     }
1175
1176     r = rbd.mirror_peer_site_set_direction(
1177       io_ctx, uuid, boost::any_cast<rbd_mirror_peer_direction_t>(direction));
1178   } else {
1179     r = update_peer_config_key(io_ctx, uuid, key, value);
1180   }
1181
1182   if (r  == -ENOENT) {
1183     std::cerr << "rbd: mirror peer " << uuid << " does not exist"
1184               << std::endl;
1185   }
1186
1187   if (r < 0) {
1188     return r;
1189   }
1190   return 0;
1191 }
1192
1193 void get_disable_arguments(po::options_description *positional,
1194                            po::options_description *options) {
1195   at::add_pool_options(positional, options, true);
1196 }
1197
1198 void get_enable_arguments(po::options_description *positional,
1199                           po::options_description *options) {
1200   at::add_pool_options(positional, options, true);
1201   positional->add_options()
1202     ("mode", "mirror mode [image or pool]");
1203   add_site_name_optional(options);
1204 }
1205
1206 int execute_enable_disable(librados::IoCtx& io_ctx,
1207                            rbd_mirror_mode_t next_mirror_mode,
1208                            const std::string &mode, bool ignore_no_update) {
1209   librbd::RBD rbd;
1210   rbd_mirror_mode_t current_mirror_mode;
1211   int r = rbd.mirror_mode_get(io_ctx, &current_mirror_mode);
1212   if (r < 0) {
1213     std::cerr << "rbd: failed to retrieve mirror mode: "
1214               << cpp_strerror(r) << std::endl;
1215     return r;
1216   }
1217
1218   if (current_mirror_mode == next_mirror_mode) {
1219     if (!ignore_no_update) {
1220       if (mode == "disabled") {
1221         std::cout << "rbd: mirroring is already " << mode << std::endl;
1222       } else {
1223         std::cout << "rbd: mirroring is already configured for "
1224                   << mode << " mode" << std::endl;
1225       }
1226     }
1227     return 0;
1228   } else if (next_mirror_mode == RBD_MIRROR_MODE_IMAGE &&
1229              current_mirror_mode == RBD_MIRROR_MODE_POOL) {
1230     std::cout << "note: changing mirroring mode from pool to image"
1231               << std::endl;
1232   } else if (next_mirror_mode == RBD_MIRROR_MODE_POOL &&
1233              current_mirror_mode == RBD_MIRROR_MODE_IMAGE) {
1234     std::cout << "note: changing mirroring mode from image to pool"
1235               << std::endl;
1236   }
1237
1238   r = rbd.mirror_mode_set(io_ctx, next_mirror_mode);
1239   if (r < 0) {
1240     return r;
1241   }
1242   return 0;
1243 }
1244
1245 int execute_disable(const po::variables_map &vm,
1246                     const std::vector<std::string> &ceph_global_init_args) {
1247   std::string pool_name;
1248   std::string namespace_name;
1249   size_t arg_index = 0;
1250   int r = utils::get_pool_and_namespace_names(vm, true, true, &pool_name,
1251                                               &namespace_name, &arg_index);
1252   if (r < 0) {
1253     return r;
1254   }
1255
1256   librados::Rados rados;
1257   librados::IoCtx io_ctx;
1258
1259   r = utils::init(pool_name, namespace_name, &rados, &io_ctx);
1260   if (r < 0) {
1261     return r;
1262   }
1263
1264   return execute_enable_disable(io_ctx, RBD_MIRROR_MODE_DISABLED, "disabled",
1265                                 false);
1266 }
1267
1268 int execute_enable(const po::variables_map &vm,
1269                    const std::vector<std::string> &ceph_global_init_args) {
1270   std::string pool_name;
1271   std::string namespace_name;
1272   size_t arg_index = 0;
1273   int r = utils::get_pool_and_namespace_names(vm, true, true, &pool_name,
1274                                               &namespace_name, &arg_index);
1275   if (r < 0) {
1276     return r;
1277   }
1278
1279   rbd_mirror_mode_t mirror_mode;
1280   std::string mode = utils::get_positional_argument(vm, arg_index++);
1281   if (mode == "image") {
1282     mirror_mode = RBD_MIRROR_MODE_IMAGE;
1283   } else if (mode == "pool") {
1284     mirror_mode = RBD_MIRROR_MODE_POOL;
1285   } else {
1286     std::cerr << "rbd: must specify 'image' or 'pool' mode." << std::endl;
1287     return -EINVAL;
1288   }
1289
1290   librados::Rados rados;
1291   librados::IoCtx io_ctx;
1292
1293   r = utils::init(pool_name, namespace_name, &rados, &io_ctx);
1294   if (r < 0) {
1295     return r;
1296   }
1297
1298   bool updated = false;
1299   if (vm.count(SITE_NAME)) {
1300     librbd::RBD rbd;
1301
1302     auto site_name = vm[SITE_NAME].as<std::string>();
1303     std::string original_site_name;
1304     r = rbd.mirror_site_name_get(rados, &original_site_name);
1305     updated = (r >= 0 && site_name != original_site_name);
1306
1307     r = set_site_name(rados, site_name);
1308     if (r < 0) {
1309       return r;
1310     }
1311   }
1312
1313   return execute_enable_disable(io_ctx, mirror_mode, mode, updated);
1314 }
1315
1316 void get_info_arguments(po::options_description *positional,
1317                         po::options_description *options) {
1318   at::add_pool_options(positional, options, true);
1319   at::add_format_options(options);
1320   options->add_options()
1321     (ALL_NAME.c_str(), po::bool_switch(), "list all attributes");
1322 }
1323
1324 int execute_info(const po::variables_map &vm,
1325                  const std::vector<std::string> &ceph_global_init_args) {
1326   std::string pool_name;
1327   std::string namespace_name;
1328   size_t arg_index = 0;
1329   int r = utils::get_pool_and_namespace_names(vm, true, false, &pool_name,
1330                                               &namespace_name, &arg_index);
1331   if (r < 0) {
1332     return r;
1333   }
1334
1335   at::Format::Formatter formatter;
1336   r = utils::get_formatter(vm, &formatter);
1337   if (r < 0) {
1338     return r;
1339   }
1340
1341   librados::Rados rados;
1342   librados::IoCtx io_ctx;
1343   r = utils::init(pool_name, namespace_name, &rados, &io_ctx);
1344   if (r < 0) {
1345     return r;
1346   }
1347
1348   librbd::RBD rbd;
1349   rbd_mirror_mode_t mirror_mode;
1350   r = rbd.mirror_mode_get(io_ctx, &mirror_mode);
1351   if (r < 0) {
1352     return r;
1353   }
1354
1355   std::string site_name;
1356   r = rbd.mirror_site_name_get(rados, &site_name);
1357   if (r < 0 && r != -EOPNOTSUPP) {
1358     return r;
1359   }
1360
1361   std::vector<librbd::mirror_peer_site_t> mirror_peers;
1362   if (namespace_name.empty()) {
1363     r = rbd.mirror_peer_site_list(io_ctx, &mirror_peers);
1364     if (r < 0) {
1365       return r;
1366     }
1367   }
1368
1369   std::string mirror_mode_desc;
1370   switch (mirror_mode) {
1371   case RBD_MIRROR_MODE_DISABLED:
1372     mirror_mode_desc = "disabled";
1373     break;
1374   case RBD_MIRROR_MODE_IMAGE:
1375     mirror_mode_desc = "image";
1376     break;
1377   case RBD_MIRROR_MODE_POOL:
1378     mirror_mode_desc = "pool";
1379     break;
1380   default:
1381     mirror_mode_desc = "unknown";
1382     break;
1383   }
1384
1385   if (formatter != nullptr) {
1386     formatter->open_object_section("mirror");
1387     formatter->dump_string("mode", mirror_mode_desc);
1388   } else {
1389     std::cout << "Mode: " << mirror_mode_desc << std::endl;
1390   }
1391
1392   if (mirror_mode != RBD_MIRROR_MODE_DISABLED && namespace_name.empty()) {
1393     if (formatter != nullptr) {
1394       formatter->dump_string("site_name", site_name);
1395     } else {
1396       std::cout << "Site Name: " << site_name << std::endl
1397                 << std::endl;
1398     }
1399
1400     r = format_mirror_peers(io_ctx, formatter, mirror_peers,
1401                             vm[ALL_NAME].as<bool>());
1402     if (r < 0) {
1403       return r;
1404     }
1405   }
1406   if (formatter != nullptr) {
1407     formatter->close_section();
1408     formatter->flush(std::cout);
1409   }
1410   return 0;
1411 }
1412
1413 void get_status_arguments(po::options_description *positional,
1414                           po::options_description *options) {
1415   at::add_pool_options(positional, options, true);
1416   at::add_format_options(options);
1417   at::add_verbose_option(options);
1418 }
1419
1420 int execute_status(const po::variables_map &vm,
1421                    const std::vector<std::string> &ceph_global_init_args) {
1422   std::string pool_name;
1423   std::string namespace_name;
1424   size_t arg_index = 0;
1425   int r = utils::get_pool_and_namespace_names(vm, true, false, &pool_name,
1426                                               &namespace_name, &arg_index);
1427   if (r < 0) {
1428     return r;
1429   }
1430
1431   at::Format::Formatter formatter;
1432   r = utils::get_formatter(vm, &formatter);
1433   if (r < 0) {
1434     return r;
1435   }
1436
1437   bool verbose = vm[at::VERBOSE].as<bool>();
1438
1439   librados::Rados rados;
1440   librados::IoCtx io_ctx;
1441   r = utils::init(pool_name, namespace_name, &rados, &io_ctx);
1442   if (r < 0) {
1443     return r;
1444   }
1445
1446   r = validate_mirroring_enabled(io_ctx);
1447   if (r < 0) {
1448     return r;
1449   }
1450
1451   librbd::RBD rbd;
1452
1453   uint32_t total_images = 0;
1454   std::map<librbd::mirror_image_status_state_t, int> mirror_image_states;
1455   MirrorHealth mirror_image_health = MIRROR_HEALTH_UNKNOWN;
1456   r = get_mirror_image_status(io_ctx, &total_images, &mirror_image_states,
1457                               &mirror_image_health);
1458   if (r < 0) {
1459     return r;
1460   }
1461
1462   MirrorDaemonServiceInfo daemon_service_info(io_ctx);
1463   daemon_service_info.init();
1464
1465   MirrorHealth mirror_daemon_health = daemon_service_info.get_daemon_health();
1466   auto mirror_services = daemon_service_info.get_mirror_services();
1467
1468   auto mirror_health = std::max(mirror_image_health, mirror_daemon_health);
1469
1470   if (formatter != nullptr) {
1471     formatter->open_object_section("status");
1472     formatter->open_object_section("summary");
1473     formatter->dump_stream("health") << mirror_health;
1474     formatter->dump_stream("daemon_health") << mirror_daemon_health;
1475     formatter->dump_stream("image_health") << mirror_image_health;
1476     formatter->open_object_section("states");
1477     for (auto &it : mirror_image_states) {
1478       std::string state_name = utils::mirror_image_status_state(it.first);
1479       formatter->dump_int(state_name.c_str(), it.second);
1480     }
1481     formatter->close_section(); // states
1482     formatter->close_section(); // summary
1483   } else {
1484     std::cout << "health: " << mirror_health << std::endl;
1485     std::cout << "daemon health: " << mirror_daemon_health << std::endl;
1486     std::cout << "image health: " << mirror_image_health << std::endl;
1487     std::cout << "images: " << total_images << " total" << std::endl;
1488     for (auto &it : mirror_image_states) {
1489       std::cout << "    " << it.second << " "
1490                 << utils::mirror_image_status_state(it.first) << std::endl;
1491     }
1492   }
1493
1494   int ret = 0;
1495
1496   if (verbose) {
1497     // dump per-daemon status
1498     if (formatter != nullptr) {
1499       formatter->open_array_section("daemons");
1500       for (auto& mirror_service : mirror_services) {
1501         formatter->open_object_section("daemon");
1502         formatter->dump_string("service_id", mirror_service.service_id);
1503         formatter->dump_string("instance_id", mirror_service.instance_id);
1504         formatter->dump_string("client_id", mirror_service.client_id);
1505         formatter->dump_string("hostname", mirror_service.hostname);
1506         formatter->dump_string("ceph_version", mirror_service.ceph_version);
1507         formatter->dump_bool("leader", mirror_service.leader);
1508         formatter->dump_stream("health") << mirror_service.health;
1509         if (!mirror_service.callouts.empty()) {
1510           formatter->open_array_section("callouts");
1511           for (auto& callout : mirror_service.callouts) {
1512             formatter->dump_string("callout", callout);
1513           }
1514           formatter->close_section(); // callouts
1515         }
1516         formatter->close_section(); // daemon
1517       }
1518       formatter->close_section(); // daemons
1519     } else {
1520       std::cout << std::endl << "DAEMONS" << std::endl;
1521       if (mirror_services.empty()) {
1522         std::cout << "  none" << std::endl;
1523       }
1524       for (auto& mirror_service : mirror_services) {
1525         std::cout << "service " << mirror_service.service_id << ":"
1526                   << std::endl
1527                   << "  instance_id: " << mirror_service.instance_id
1528                   << std::endl
1529                   << "  client_id: " << mirror_service.client_id << std::endl
1530                   << "  hostname: " << mirror_service.hostname << std::endl
1531                   << "  version: " << mirror_service.ceph_version << std::endl
1532                   << "  leader: " << (mirror_service.leader ? "true" : "false")
1533                   << std::endl
1534                   << "  health: " << mirror_service.health << std::endl;
1535         if (!mirror_service.callouts.empty()) {
1536           std::cout << "  callouts: " << mirror_service.callouts << std::endl;
1537         }
1538         std::cout << std::endl;
1539       }
1540       std::cout << std::endl;
1541     }
1542
1543     // dump per-image status
1544     librados::IoCtx default_ns_io_ctx;
1545     default_ns_io_ctx.dup(io_ctx);
1546     default_ns_io_ctx.set_namespace("");
1547     std::vector<librbd::mirror_peer_site_t> mirror_peers;
1548     utils::get_mirror_peer_sites(default_ns_io_ctx, &mirror_peers);
1549
1550     std::map<std::string, std::string> peer_mirror_uuids_to_name;
1551     utils::get_mirror_peer_mirror_uuids_to_names(mirror_peers,
1552                                                  &peer_mirror_uuids_to_name);
1553
1554     if (formatter != nullptr) {
1555       formatter->open_array_section("images");
1556     } else {
1557       std::cout << "IMAGES";
1558     }
1559
1560     std::map<std::string, std::string> instance_ids;
1561
1562     std::string start_image_id;
1563     while (true) {
1564       std::map<std::string, std::string> ids;
1565       r = rbd.mirror_image_instance_id_list(io_ctx, start_image_id, 1024, &ids);
1566       if (r < 0) {
1567         if (r == -EOPNOTSUPP) {
1568           std::cerr << "rbd: newer release of Ceph OSDs required to map image "
1569                     << "to rbd-mirror daemon instance" << std::endl;
1570         } else {
1571           std::cerr << "rbd: failed to get instance id list: "
1572                     << cpp_strerror(r) << std::endl;
1573         }
1574         // not fatal
1575         break;
1576       }
1577       if (ids.empty()) {
1578         break;
1579       }
1580       instance_ids.insert(ids.begin(), ids.end());
1581       start_image_id = ids.rbegin()->first;
1582     }
1583
1584     ImageRequestGenerator<StatusImageRequest> generator(
1585       io_ctx, instance_ids, mirror_peers, peer_mirror_uuids_to_name,
1586       daemon_service_info, formatter);
1587     ret = generator.execute();
1588
1589     if (formatter != nullptr) {
1590       formatter->close_section(); // images
1591     }
1592   }
1593
1594   if (formatter != nullptr) {
1595     formatter->close_section(); // status
1596     formatter->flush(std::cout);
1597   }
1598
1599   return ret;
1600 }
1601
1602 void get_promote_arguments(po::options_description *positional,
1603                            po::options_description *options) {
1604   options->add_options()
1605     ("force", po::bool_switch(),
1606      "promote even if not cleanly demoted by remote cluster");
1607   at::add_pool_options(positional, options, true);
1608 }
1609
1610 int execute_promote(const po::variables_map &vm,
1611                     const std::vector<std::string> &ceph_global_init_args) {
1612   std::string pool_name;
1613   std::string namespace_name;
1614   size_t arg_index = 0;
1615   int r = utils::get_pool_and_namespace_names(vm, true, true, &pool_name,
1616                                               &namespace_name, &arg_index);
1617   if (r < 0) {
1618     return r;
1619   }
1620
1621   librados::Rados rados;
1622   librados::IoCtx io_ctx;
1623   r = utils::init(pool_name, namespace_name, &rados, &io_ctx);
1624   if (r < 0) {
1625     return r;
1626   }
1627
1628   r = validate_mirroring_enabled(io_ctx);
1629   if (r < 0) {
1630     return r;
1631   }
1632
1633   utils::disable_cache();
1634
1635   std::atomic<unsigned> counter = { 0 };
1636   ImageRequestGenerator<PromoteImageRequest> generator(io_ctx, &counter,
1637                                                        vm["force"].as<bool>());
1638   r = generator.execute();
1639
1640   std::cout << "Promoted " << counter.load() << " mirrored images" << std::endl;
1641   return r;
1642 }
1643
1644 void get_demote_arguments(po::options_description *positional,
1645                            po::options_description *options) {
1646   at::add_pool_options(positional, options, true);
1647 }
1648
1649 int execute_demote(const po::variables_map &vm,
1650                    const std::vector<std::string> &ceph_global_init_args) {
1651   std::string pool_name;
1652   std::string namespace_name;
1653   size_t arg_index = 0;
1654   int r = utils::get_pool_and_namespace_names(vm, true, true, &pool_name,
1655                                               &namespace_name, &arg_index);
1656   if (r < 0) {
1657     return r;
1658   }
1659
1660   librados::Rados rados;
1661   librados::IoCtx io_ctx;
1662   r = utils::init(pool_name, namespace_name, &rados, &io_ctx);
1663   if (r < 0) {
1664     return r;
1665   }
1666
1667   r = validate_mirroring_enabled(io_ctx);
1668   if (r < 0) {
1669     return r;
1670   }
1671
1672   utils::disable_cache();
1673
1674   std::atomic<unsigned> counter { 0 };
1675   ImageRequestGenerator<DemoteImageRequest> generator(io_ctx, &counter);
1676   r = generator.execute();
1677
1678   std::cout << "Demoted " << counter.load() << " mirrored images" << std::endl;
1679   return r;
1680 }
1681
1682 Shell::Action action_bootstrap_create(
1683   {"mirror", "pool", "peer", "bootstrap", "create"}, {},
1684   "Create a peer bootstrap token to import in a remote cluster", "",
1685   &get_peer_bootstrap_create_arguments, &execute_peer_bootstrap_create);
1686 Shell::Action action_bootstreap_import(
1687   {"mirror", "pool", "peer", "bootstrap", "import"}, {},
1688   "Import a peer bootstrap token created from a remote cluster", "",
1689   &get_peer_bootstrap_import_arguments, &execute_peer_bootstrap_import);
1690
1691 Shell::Action action_add(
1692   {"mirror", "pool", "peer", "add"}, {},
1693   "Add a mirroring peer to a pool.", "",
1694   &get_peer_add_arguments, &execute_peer_add);
1695 Shell::Action action_remove(
1696   {"mirror", "pool", "peer", "remove"}, {},
1697   "Remove a mirroring peer from a pool.", "",
1698   &get_peer_remove_arguments, &execute_peer_remove);
1699 Shell::Action action_set(
1700   {"mirror", "pool", "peer", "set"}, {},
1701   "Update mirroring peer settings.", "",
1702   &get_peer_set_arguments, &execute_peer_set);
1703
1704 Shell::Action action_disable(
1705   {"mirror", "pool", "disable"}, {},
1706   "Disable RBD mirroring by default within a pool.", "",
1707   &get_disable_arguments, &execute_disable);
1708 Shell::Action action_enable(
1709   {"mirror", "pool", "enable"}, {},
1710   "Enable RBD mirroring by default within a pool.", "",
1711   &get_enable_arguments, &execute_enable);
1712 Shell::Action action_info(
1713   {"mirror", "pool", "info"}, {},
1714   "Show information about the pool mirroring configuration.", {},
1715   &get_info_arguments, &execute_info);
1716 Shell::Action action_status(
1717   {"mirror", "pool", "status"}, {},
1718   "Show status for all mirrored images in the pool.", {},
1719   &get_status_arguments, &execute_status);
1720 Shell::Action action_promote(
1721   {"mirror", "pool", "promote"}, {},
1722   "Promote all non-primary images in the pool.", {},
1723   &get_promote_arguments, &execute_promote);
1724 Shell::Action action_demote(
1725   {"mirror", "pool", "demote"}, {},
1726   "Demote all primary images in the pool.", {},
1727   &get_demote_arguments, &execute_demote);
1728
1729 } // namespace mirror_pool
1730 } // namespace action
1731 } // namespace rbd