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   add_site_name_optional(options);
 822 }
 823
 824 int execute_peer_bootstrap_create(
 825     const po::variables_map &vm,
 826     const std::vector<std::string> &ceph_global_init_args) {
 827   std::string pool_name;
 828   size_t arg_index = 0;
 829   int r = utils::get_pool_and_namespace_names(vm, true, true, &pool_name,
 830                                               nullptr, &arg_index);
 831   if (r < 0) {
 832     return r;
 833   }
 834
 835   librados::Rados rados;
 836   librados::IoCtx io_ctx;
 837   r = utils::init(pool_name, "", &rados, &io_ctx);
 838   if (r < 0) {
 839     return r;
 840   }
 841
 842   r = validate_mirroring_enabled(io_ctx);
 843   if (r < 0) {
 844     return r;
 845   }
 846
 847   if (vm.count(SITE_NAME)) {
 848     r = set_site_name(rados, vm[SITE_NAME].as<std::string>());
 849     if (r < 0) {
 850       return r;
 851     }
 852   }
 853
 854   librbd::RBD rbd;
 855   std::string token;
 856   r = rbd.mirror_peer_bootstrap_create(io_ctx, &token);
 857   if (r == -EEXIST) {
 858     std::cerr << "rbd: mismatch with pre-existing RBD mirroring peer user caps"
 859               << std::endl;
 860   } else if (r < 0) {
 861     std::cerr << "rbd: failed to create mirroring bootstrap token: "
 862               << cpp_strerror(r) << std::endl;
 863     return r;
 864   }
 865
 866   std::cout << token << std::endl;
 867   return 0;
 868 }
 869
 870 void get_peer_bootstrap_import_arguments(po::options_description *positional,
 871                                          po::options_description *options) {
 872   at::add_pool_options(positional, options, false);
 873   add_site_name_optional(options);
 874   positional->add_options()
 875     ("token-path", po::value<std::string>(),
 876      "bootstrap token file (or '-' for stdin)");
 877   options->add_options()
 878     ("token-path", po::value<std::string>(),
 879      "bootstrap token file (or '-' for stdin)");
 880   add_direction_optional(options);
 881 }
 882
 883 int execute_peer_bootstrap_import(
 884     const po::variables_map &vm,
 885     const std::vector<std::string> &ceph_global_init_args) {
 886   std::string pool_name;
 887   size_t arg_index = 0;
 888   int r = utils::get_pool_and_namespace_names(vm, true, true, &pool_name,
 889                                               nullptr, &arg_index);
 890   if (r < 0) {
 891     return r;
 892   }
 893
 894   std::string token_path;
 895   if (vm.count("token-path")) {
 896     token_path = vm["token-path"].as<std::string>();
 897   } else {
 898     token_path = utils::get_positional_argument(vm, arg_index++);
 899   }
 900
 901   if (token_path.empty()) {
 902     std::cerr << "rbd: token path was not specified" << std::endl;
 903     return -EINVAL;
 904   }
 905
 906   rbd_mirror_peer_direction_t mirror_peer_direction =
 907     RBD_MIRROR_PEER_DIRECTION_RX_TX;
 908   if (vm.count("direction")) {
 909     mirror_peer_direction = vm["direction"].as<rbd_mirror_peer_direction_t>();
 910   }
 911
 912   int fd = STDIN_FILENO;
 913   if (token_path != "-") {
 914     fd = open(token_path.c_str(), O_RDONLY|O_BINARY);
 915     if (fd < 0) {
 916       r = -errno;
 917       std::cerr << "rbd: error opening " << token_path << std::endl;
 918       return r;
 919     }
 920   }
 921
 922   char token[1024];
 923   memset(token, 0, sizeof(token));
 924   r = safe_read(fd, token, sizeof(token) - 1);
 925   if (fd != STDIN_FILENO) {
 926     VOID_TEMP_FAILURE_RETRY(close(fd));
 927   }
 928
 929   if (r < 0) {
 930     std::cerr << "rbd: error reading token file: " << cpp_strerror(r)
 931               << std::endl;
 932     return r;
 933   }
 934
 935   librados::Rados rados;
 936   librados::IoCtx io_ctx;
 937   r = utils::init(pool_name, "", &rados, &io_ctx);
 938   if (r < 0) {
 939     return r;
 940   }
 941
 942   if (vm.count(SITE_NAME)) {
 943     r = set_site_name(rados, vm[SITE_NAME].as<std::string>());
 944     if (r < 0) {
 945       return r;
 946     }
 947   }
 948
 949   librbd::RBD rbd;
 950   r = rbd.mirror_peer_bootstrap_import(io_ctx, mirror_peer_direction, token);
 951   if (r == -ENOSYS) {
 952     std::cerr << "rbd: mirroring is not enabled on remote peer" << std::endl;
 953     return r;
 954   } else if (r < 0) {
 955     std::cerr << "rbd: failed to import peer bootstrap token" << std::endl;
 956     return r;
 957   }
 958
 959   return 0;
 960 }
 961
 962 void get_peer_add_arguments(po::options_description *positional,
 963                             po::options_description *options) {
 964   at::add_pool_options(positional, options, false);
 965   positional->add_options()
 966     ("remote-cluster-spec", "remote cluster spec\n"
 967      "(example: [<client name>@]<cluster name>)");
 968   options->add_options()
 969     ("remote-client-name", po::value<std::string>(), "remote client name")
 970     ("remote-cluster", po::value<std::string>(), "remote cluster name")
 971     ("remote-mon-host", po::value<std::string>(), "remote mon host(s)")
 972     ("remote-key-file", po::value<std::string>(),
 973      "path to file containing remote key");
 974   add_direction_optional(options);
 975 }
 976
 977 int execute_peer_add(const po::variables_map &vm,
 978                      const std::vector<std::string> &ceph_global_init_args) {
 979   std::string pool_name;
 980   size_t arg_index = 0;
 981   int r = utils::get_pool_and_namespace_names(vm, true, true, &pool_name,
 982                                               nullptr, &arg_index);
 983   if (r < 0) {
 984     return r;
 985   }
 986
 987   std::string remote_client_name = g_ceph_context->_conf->name.to_str();
 988   std::string remote_cluster;
 989   std::map<std::string, std::string> attributes;
 990   r = get_remote_cluster_spec(
 991     vm, utils::get_positional_argument(vm, arg_index),
 992     &remote_client_name, &remote_cluster, &attributes);
 993   if (r < 0) {
 994     return r;
 995   }
 996
 997   librados::Rados rados;
 998   librados::IoCtx io_ctx;
 999   r = utils::init(pool_name, "", &rados, &io_ctx);
1000   if (r < 0) {
1001     return r;
1002   }
1003
1004   r = validate_mirroring_enabled(io_ctx);
1005   if (r < 0) {
1006     return r;
1007   }
1008
1009   // TODO: temporary restriction to prevent adding multiple peers
1010   // until rbd-mirror daemon can properly handle the scenario
1011   librbd::RBD rbd;
1012   std::vector<librbd::mirror_peer_site_t> mirror_peers;
1013   r = rbd.mirror_peer_site_list(io_ctx, &mirror_peers);
1014   if (r < 0) {
1015     std::cerr << "rbd: failed to list mirror peers" << std::endl;
1016     return r;
1017   }
1018
1019   // ignore tx-only peers since the restriction is for rx
1020   mirror_peers.erase(
1021     std::remove_if(
1022       mirror_peers.begin(), mirror_peers.end(),
1023       [](const librbd::mirror_peer_site_t& peer) {
1024         return (peer.direction == RBD_MIRROR_PEER_DIRECTION_TX);
1025       }),
1026     mirror_peers.end());
1027
1028   if (!mirror_peers.empty()) {
1029     std::cerr << "rbd: multiple RX peers are not currently supported"
1030               << std::endl;
1031     return -EINVAL;
1032   }
1033
1034   rbd_mirror_peer_direction_t mirror_peer_direction =
1035     RBD_MIRROR_PEER_DIRECTION_RX_TX;
1036   if (vm.count("direction")) {
1037     mirror_peer_direction = vm["direction"].as<rbd_mirror_peer_direction_t>();
1038   }
1039
1040   std::string uuid;
1041   r = rbd.mirror_peer_site_add(
1042     io_ctx, &uuid, mirror_peer_direction, remote_cluster, remote_client_name);
1043   if (r == -EEXIST) {
1044     std::cerr << "rbd: mirror peer already exists" << std::endl;
1045     return r;
1046   } else if (r < 0) {
1047     std::cerr << "rbd: error adding mirror peer" << std::endl;
1048     return r;
1049   }
1050
1051   if (!attributes.empty()) {
1052     r = set_peer_config_key(io_ctx, uuid, std::move(attributes));
1053     if (r < 0) {
1054       return r;
1055     }
1056   }
1057
1058   std::cout << uuid << std::endl;
1059   return 0;
1060 }
1061
1062 void get_peer_remove_arguments(po::options_description *positional,
1063                                po::options_description *options) {
1064   at::add_pool_options(positional, options, false);
1065   add_uuid_option(positional);
1066 }
1067
1068 int execute_peer_remove(const po::variables_map &vm,
1069                         const std::vector<std::string> &ceph_global_init_args) {
1070   std::string pool_name;
1071   size_t arg_index = 0;
1072   int r = utils::get_pool_and_namespace_names(vm, true, true, &pool_name,
1073                                               nullptr, &arg_index);
1074   if (r < 0) {
1075     return r;
1076   }
1077
1078   std::string uuid;
1079   r = get_uuid(vm, arg_index, &uuid);
1080   if (r < 0) {
1081     return r;
1082   }
1083
1084   librados::Rados rados;
1085   librados::IoCtx io_ctx;
1086   r = utils::init(pool_name, "", &rados, &io_ctx);
1087   if (r < 0) {
1088     return r;
1089   }
1090
1091   r = validate_mirroring_enabled(io_ctx);
1092   if (r < 0) {
1093     return r;
1094   }
1095
1096   librbd::RBD rbd;
1097   r = rbd.mirror_peer_site_remove(io_ctx, uuid);
1098   if (r < 0) {
1099     std::cerr << "rbd: error removing mirror peer" << std::endl;
1100     return r;
1101   }
1102   return 0;
1103 }
1104
1105 void get_peer_set_arguments(po::options_description *positional,
1106                             po::options_description *options) {
1107   at::add_pool_options(positional, options, false);
1108   add_uuid_option(positional);
1109   positional->add_options()
1110     ("key", "peer parameter\n"
1111             "(direction, site-name, client, mon-host, key-file)")
1112     ("value", "new value for specified key\n"
1113               "(rx-only, tx-only, or rx-tx for direction)");
1114 }
1115
1116 int execute_peer_set(const po::variables_map &vm,
1117                      const std::vector<std::string> &ceph_global_init_args) {
1118   std::string pool_name;
1119   size_t arg_index = 0;
1120   int r = utils::get_pool_and_namespace_names(vm, true, true, &pool_name,
1121                                               nullptr, &arg_index);
1122   if (r < 0) {
1123     return r;
1124   }
1125
1126   std::string uuid;
1127   r = get_uuid(vm, arg_index++, &uuid);
1128   if (r < 0) {
1129     return r;
1130   }
1131
1132   std::set<std::string> valid_keys{{"direction", "site-name", "cluster",
1133                                     "client", "mon-host", "key-file"}};
1134   std::string key = utils::get_positional_argument(vm, arg_index++);
1135   if (valid_keys.find(key) == valid_keys.end()) {
1136     std::cerr << "rbd: must specify ";
1137     for (auto& valid_key : valid_keys) {
1138       std::cerr << "'" << valid_key << "'";
1139       if (&valid_key != &(*valid_keys.rbegin())) {
1140         std::cerr << ", ";
1141       }
1142     }
1143     std::cerr <<  " key." << std::endl;
1144     return -EINVAL;
1145   }
1146
1147   std::string value = utils::get_positional_argument(vm, arg_index++);
1148   if (value.empty() && (key == "client" || key == "cluster")) {
1149     std::cerr << "rbd: must specify new " << key << " value." << std::endl;
1150   } else if (key == "key-file") {
1151     key = "key";
1152     r = read_key_file(value, &value);
1153     if (r < 0) {
1154       return r;
1155     }
1156   } else if (key == "mon-host") {
1157     key = "mon_host";
1158   }
1159
1160   librados::Rados rados;
1161   librados::IoCtx io_ctx;
1162   r = utils::init(pool_name, "", &rados, &io_ctx);
1163   if (r < 0) {
1164     return r;
1165   }
1166
1167   r = validate_mirroring_enabled(io_ctx);
1168   if (r < 0) {
1169     return r;
1170   }
1171
1172   librbd::RBD rbd;
1173   if (key == "client") {
1174     r = rbd.mirror_peer_site_set_client_name(io_ctx, uuid.c_str(),
1175                                              value.c_str());
1176   } else if (key == "site-name" || key == "cluster") {
1177     r = rbd.mirror_peer_site_set_name(io_ctx, uuid.c_str(), value.c_str());
1178   } else if (key == "direction") {
1179     MirrorPeerDirection tag;
1180     boost::any direction;
1181     try {
1182       validate(direction, {value}, &tag, 1);
1183     } catch (...) {
1184       std::cerr << "rbd: invalid direction" << std::endl;
1185       return -EINVAL;
1186     }
1187
1188     auto peer_direction = boost::any_cast<rbd_mirror_peer_direction_t>(
1189       direction);
1190     if (peer_direction != RBD_MIRROR_PEER_DIRECTION_TX) {
1191       // TODO: temporary restriction to prevent adding multiple peers
1192       // until rbd-mirror daemon can properly handle the scenario
1193       std::vector<librbd::mirror_peer_site_t> mirror_peers;
1194       r = rbd.mirror_peer_site_list(io_ctx, &mirror_peers);
1195       if (r < 0) {
1196         std::cerr << "rbd: failed to list mirror peers" << std::endl;
1197         return r;
1198       }
1199
1200       // ignore peer to be updated and tx-only peers since the restriction is
1201       // for rx
1202       mirror_peers.erase(
1203         std::remove_if(
1204           mirror_peers.begin(), mirror_peers.end(),
1205           [uuid](const librbd::mirror_peer_site_t& peer) {
1206             return (peer.uuid == uuid ||
1207                     peer.direction == RBD_MIRROR_PEER_DIRECTION_TX);
1208           }),
1209         mirror_peers.end());
1210
1211       if (!mirror_peers.empty()) {
1212         std::cerr << "rbd: multiple RX peers are not currently supported"
1213                   << std::endl;
1214         return -EINVAL;
1215       }
1216     }
1217
1218     r = rbd.mirror_peer_site_set_direction(io_ctx, uuid, peer_direction);
1219   } else {
1220     r = update_peer_config_key(io_ctx, uuid, key, value);
1221   }
1222
1223   if (r  == -ENOENT) {
1224     std::cerr << "rbd: mirror peer " << uuid << " does not exist"
1225               << std::endl;
1226   }
1227
1228   if (r < 0) {
1229     return r;
1230   }
1231   return 0;
1232 }
1233
1234 void get_disable_arguments(po::options_description *positional,
1235                            po::options_description *options) {
1236   at::add_pool_options(positional, options, true);
1237 }
1238
1239 void get_enable_arguments(po::options_description *positional,
1240                           po::options_description *options) {
1241   at::add_pool_options(positional, options, true);
1242   positional->add_options()
1243     ("mode", "mirror mode [image or pool]");
1244   add_site_name_optional(options);
1245 }
1246
1247 int execute_enable_disable(librados::IoCtx& io_ctx,
1248                            rbd_mirror_mode_t next_mirror_mode,
1249                            const std::string &mode, bool ignore_no_update) {
1250   librbd::RBD rbd;
1251   rbd_mirror_mode_t current_mirror_mode;
1252   int r = rbd.mirror_mode_get(io_ctx, &current_mirror_mode);
1253   if (r < 0) {
1254     std::cerr << "rbd: failed to retrieve mirror mode: "
1255               << cpp_strerror(r) << std::endl;
1256     return r;
1257   }
1258
1259   if (current_mirror_mode == next_mirror_mode) {
1260     if (!ignore_no_update) {
1261       if (mode == "disabled") {
1262         std::cout << "rbd: mirroring is already " << mode << std::endl;
1263       } else {
1264         std::cout << "rbd: mirroring is already configured for "
1265                   << mode << " mode" << std::endl;
1266       }
1267     }
1268     return 0;
1269   } else if (next_mirror_mode == RBD_MIRROR_MODE_IMAGE &&
1270              current_mirror_mode == RBD_MIRROR_MODE_POOL) {
1271     std::cout << "note: changing mirroring mode from pool to image"
1272               << std::endl;
1273   } else if (next_mirror_mode == RBD_MIRROR_MODE_POOL &&
1274              current_mirror_mode == RBD_MIRROR_MODE_IMAGE) {
1275     std::cout << "note: changing mirroring mode from image to pool"
1276               << std::endl;
1277   }
1278
1279   r = rbd.mirror_mode_set(io_ctx, next_mirror_mode);
1280   if (r < 0) {
1281     return r;
1282   }
1283   return 0;
1284 }
1285
1286 int execute_disable(const po::variables_map &vm,
1287                     const std::vector<std::string> &ceph_global_init_args) {
1288   std::string pool_name;
1289   std::string namespace_name;
1290   size_t arg_index = 0;
1291   int r = utils::get_pool_and_namespace_names(vm, true, true, &pool_name,
1292                                               &namespace_name, &arg_index);
1293   if (r < 0) {
1294     return r;
1295   }
1296
1297   librados::Rados rados;
1298   librados::IoCtx io_ctx;
1299
1300   r = utils::init(pool_name, namespace_name, &rados, &io_ctx);
1301   if (r < 0) {
1302     return r;
1303   }
1304
1305   return execute_enable_disable(io_ctx, RBD_MIRROR_MODE_DISABLED, "disabled",
1306                                 false);
1307 }
1308
1309 int execute_enable(const po::variables_map &vm,
1310                    const std::vector<std::string> &ceph_global_init_args) {
1311   std::string pool_name;
1312   std::string namespace_name;
1313   size_t arg_index = 0;
1314   int r = utils::get_pool_and_namespace_names(vm, true, true, &pool_name,
1315                                               &namespace_name, &arg_index);
1316   if (r < 0) {
1317     return r;
1318   }
1319
1320   rbd_mirror_mode_t mirror_mode;
1321   std::string mode = utils::get_positional_argument(vm, arg_index++);
1322   if (mode == "image") {
1323     mirror_mode = RBD_MIRROR_MODE_IMAGE;
1324   } else if (mode == "pool") {
1325     mirror_mode = RBD_MIRROR_MODE_POOL;
1326   } else {
1327     std::cerr << "rbd: must specify 'image' or 'pool' mode." << std::endl;
1328     return -EINVAL;
1329   }
1330
1331   librados::Rados rados;
1332   librados::IoCtx io_ctx;
1333
1334   r = utils::init(pool_name, namespace_name, &rados, &io_ctx);
1335   if (r < 0) {
1336     return r;
1337   }
1338
1339   bool updated = false;
1340   if (vm.count(SITE_NAME)) {
1341     librbd::RBD rbd;
1342
1343     auto site_name = vm[SITE_NAME].as<std::string>();
1344     std::string original_site_name;
1345     r = rbd.mirror_site_name_get(rados, &original_site_name);
1346     updated = (r >= 0 && site_name != original_site_name);
1347
1348     r = set_site_name(rados, site_name);
1349     if (r < 0) {
1350       return r;
1351     }
1352   }
1353
1354   return execute_enable_disable(io_ctx, mirror_mode, mode, updated);
1355 }
1356
1357 void get_info_arguments(po::options_description *positional,
1358                         po::options_description *options) {
1359   at::add_pool_options(positional, options, true);
1360   at::add_format_options(options);
1361   options->add_options()
1362     (ALL_NAME.c_str(), po::bool_switch(), "list all attributes");
1363 }
1364
1365 int execute_info(const po::variables_map &vm,
1366                  const std::vector<std::string> &ceph_global_init_args) {
1367   std::string pool_name;
1368   std::string namespace_name;
1369   size_t arg_index = 0;
1370   int r = utils::get_pool_and_namespace_names(vm, true, false, &pool_name,
1371                                               &namespace_name, &arg_index);
1372   if (r < 0) {
1373     return r;
1374   }
1375
1376   at::Format::Formatter formatter;
1377   r = utils::get_formatter(vm, &formatter);
1378   if (r < 0) {
1379     return r;
1380   }
1381
1382   librados::Rados rados;
1383   librados::IoCtx io_ctx;
1384   r = utils::init(pool_name, namespace_name, &rados, &io_ctx);
1385   if (r < 0) {
1386     return r;
1387   }
1388
1389   librbd::RBD rbd;
1390   rbd_mirror_mode_t mirror_mode;
1391   r = rbd.mirror_mode_get(io_ctx, &mirror_mode);
1392   if (r < 0) {
1393     return r;
1394   }
1395
1396   std::string site_name;
1397   r = rbd.mirror_site_name_get(rados, &site_name);
1398   if (r < 0 && r != -EOPNOTSUPP) {
1399     return r;
1400   }
1401
1402   std::vector<librbd::mirror_peer_site_t> mirror_peers;
1403   if (namespace_name.empty()) {
1404     r = rbd.mirror_peer_site_list(io_ctx, &mirror_peers);
1405     if (r < 0) {
1406       return r;
1407     }
1408   }
1409
1410   std::string mirror_mode_desc;
1411   switch (mirror_mode) {
1412   case RBD_MIRROR_MODE_DISABLED:
1413     mirror_mode_desc = "disabled";
1414     break;
1415   case RBD_MIRROR_MODE_IMAGE:
1416     mirror_mode_desc = "image";
1417     break;
1418   case RBD_MIRROR_MODE_POOL:
1419     mirror_mode_desc = "pool";
1420     break;
1421   default:
1422     mirror_mode_desc = "unknown";
1423     break;
1424   }
1425
1426   if (formatter != nullptr) {
1427     formatter->open_object_section("mirror");
1428     formatter->dump_string("mode", mirror_mode_desc);
1429   } else {
1430     std::cout << "Mode: " << mirror_mode_desc << std::endl;
1431   }
1432
1433   if (mirror_mode != RBD_MIRROR_MODE_DISABLED && namespace_name.empty()) {
1434     if (formatter != nullptr) {
1435       formatter->dump_string("site_name", site_name);
1436     } else {
1437       std::cout << "Site Name: " << site_name << std::endl
1438                 << std::endl;
1439     }
1440
1441     r = format_mirror_peers(io_ctx, formatter, mirror_peers,
1442                             vm[ALL_NAME].as<bool>());
1443     if (r < 0) {
1444       return r;
1445     }
1446   }
1447   if (formatter != nullptr) {
1448     formatter->close_section();
1449     formatter->flush(std::cout);
1450   }
1451   return 0;
1452 }
1453
1454 void get_status_arguments(po::options_description *positional,
1455                           po::options_description *options) {
1456   at::add_pool_options(positional, options, true);
1457   at::add_format_options(options);
1458   at::add_verbose_option(options);
1459 }
1460
1461 int execute_status(const po::variables_map &vm,
1462                    const std::vector<std::string> &ceph_global_init_args) {
1463   std::string pool_name;
1464   std::string namespace_name;
1465   size_t arg_index = 0;
1466   int r = utils::get_pool_and_namespace_names(vm, true, false, &pool_name,
1467                                               &namespace_name, &arg_index);
1468   if (r < 0) {
1469     return r;
1470   }
1471
1472   at::Format::Formatter formatter;
1473   r = utils::get_formatter(vm, &formatter);
1474   if (r < 0) {
1475     return r;
1476   }
1477
1478   bool verbose = vm[at::VERBOSE].as<bool>();
1479
1480   librados::Rados rados;
1481   librados::IoCtx io_ctx;
1482   r = utils::init(pool_name, namespace_name, &rados, &io_ctx);
1483   if (r < 0) {
1484     return r;
1485   }
1486
1487   r = validate_mirroring_enabled(io_ctx);
1488   if (r < 0) {
1489     return r;
1490   }
1491
1492   librbd::RBD rbd;
1493
1494   uint32_t total_images = 0;
1495   std::map<librbd::mirror_image_status_state_t, int> mirror_image_states;
1496   MirrorHealth mirror_image_health = MIRROR_HEALTH_UNKNOWN;
1497   r = get_mirror_image_status(io_ctx, &total_images, &mirror_image_states,
1498                               &mirror_image_health);
1499   if (r < 0) {
1500     return r;
1501   }
1502
1503   MirrorDaemonServiceInfo daemon_service_info(io_ctx);
1504   daemon_service_info.init();
1505
1506   MirrorHealth mirror_daemon_health = daemon_service_info.get_daemon_health();
1507   auto mirror_services = daemon_service_info.get_mirror_services();
1508
1509   auto mirror_health = std::max(mirror_image_health, mirror_daemon_health);
1510
1511   if (formatter != nullptr) {
1512     formatter->open_object_section("status");
1513     formatter->open_object_section("summary");
1514     formatter->dump_stream("health") << mirror_health;
1515     formatter->dump_stream("daemon_health") << mirror_daemon_health;
1516     formatter->dump_stream("image_health") << mirror_image_health;
1517     formatter->open_object_section("states");
1518     for (auto &it : mirror_image_states) {
1519       std::string state_name = utils::mirror_image_status_state(it.first);
1520       formatter->dump_int(state_name.c_str(), it.second);
1521     }
1522     formatter->close_section(); // states
1523     formatter->close_section(); // summary
1524   } else {
1525     std::cout << "health: " << mirror_health << std::endl;
1526     std::cout << "daemon health: " << mirror_daemon_health << std::endl;
1527     std::cout << "image health: " << mirror_image_health << std::endl;
1528     std::cout << "images: " << total_images << " total" << std::endl;
1529     for (auto &it : mirror_image_states) {
1530       std::cout << "    " << it.second << " "
1531                 << utils::mirror_image_status_state(it.first) << std::endl;
1532     }
1533   }
1534
1535   int ret = 0;
1536
1537   if (verbose) {
1538     // dump per-daemon status
1539     if (formatter != nullptr) {
1540       formatter->open_array_section("daemons");
1541       for (auto& mirror_service : mirror_services) {
1542         formatter->open_object_section("daemon");
1543         formatter->dump_string("service_id", mirror_service.service_id);
1544         formatter->dump_string("instance_id", mirror_service.instance_id);
1545         formatter->dump_string("client_id", mirror_service.client_id);
1546         formatter->dump_string("hostname", mirror_service.hostname);
1547         formatter->dump_string("ceph_version", mirror_service.ceph_version);
1548         formatter->dump_bool("leader", mirror_service.leader);
1549         formatter->dump_stream("health") << mirror_service.health;
1550         if (!mirror_service.callouts.empty()) {
1551           formatter->open_array_section("callouts");
1552           for (auto& callout : mirror_service.callouts) {
1553             formatter->dump_string("callout", callout);
1554           }
1555           formatter->close_section(); // callouts
1556         }
1557         formatter->close_section(); // daemon
1558       }
1559       formatter->close_section(); // daemons
1560     } else {
1561       std::cout << std::endl << "DAEMONS" << std::endl;
1562       if (mirror_services.empty()) {
1563         std::cout << "  none" << std::endl;
1564       }
1565       for (auto& mirror_service : mirror_services) {
1566         std::cout << "service " << mirror_service.service_id << ":"
1567                   << std::endl
1568                   << "  instance_id: " << mirror_service.instance_id
1569                   << std::endl
1570                   << "  client_id: " << mirror_service.client_id << std::endl
1571                   << "  hostname: " << mirror_service.hostname << std::endl
1572                   << "  version: " << mirror_service.ceph_version << std::endl
1573                   << "  leader: " << (mirror_service.leader ? "true" : "false")
1574                   << std::endl
1575                   << "  health: " << mirror_service.health << std::endl;
1576         if (!mirror_service.callouts.empty()) {
1577           std::cout << "  callouts: " << mirror_service.callouts << std::endl;
1578         }
1579         std::cout << std::endl;
1580       }
1581       std::cout << std::endl;
1582     }
1583
1584     // dump per-image status
1585     librados::IoCtx default_ns_io_ctx;
1586     default_ns_io_ctx.dup(io_ctx);
1587     default_ns_io_ctx.set_namespace("");
1588     std::vector<librbd::mirror_peer_site_t> mirror_peers;
1589     utils::get_mirror_peer_sites(default_ns_io_ctx, &mirror_peers);
1590
1591     std::map<std::string, std::string> peer_mirror_uuids_to_name;
1592     utils::get_mirror_peer_mirror_uuids_to_names(mirror_peers,
1593                                                  &peer_mirror_uuids_to_name);
1594
1595     if (formatter != nullptr) {
1596       formatter->open_array_section("images");
1597     } else {
1598       std::cout << "IMAGES";
1599     }
1600
1601     std::map<std::string, std::string> instance_ids;
1602
1603     std::string start_image_id;
1604     while (true) {
1605       std::map<std::string, std::string> ids;
1606       r = rbd.mirror_image_instance_id_list(io_ctx, start_image_id, 1024, &ids);
1607       if (r < 0) {
1608         if (r == -EOPNOTSUPP) {
1609           std::cerr << "rbd: newer release of Ceph OSDs required to map image "
1610                     << "to rbd-mirror daemon instance" << std::endl;
1611         } else {
1612           std::cerr << "rbd: failed to get instance id list: "
1613                     << cpp_strerror(r) << std::endl;
1614         }
1615         // not fatal
1616         break;
1617       }
1618       if (ids.empty()) {
1619         break;
1620       }
1621       instance_ids.insert(ids.begin(), ids.end());
1622       start_image_id = ids.rbegin()->first;
1623     }
1624
1625     ImageRequestGenerator<StatusImageRequest> generator(
1626       io_ctx, instance_ids, mirror_peers, peer_mirror_uuids_to_name,
1627       daemon_service_info, formatter);
1628     ret = generator.execute();
1629
1630     if (formatter != nullptr) {
1631       formatter->close_section(); // images
1632     }
1633   }
1634
1635   if (formatter != nullptr) {
1636     formatter->close_section(); // status
1637     formatter->flush(std::cout);
1638   }
1639
1640   return ret;
1641 }
1642
1643 void get_promote_arguments(po::options_description *positional,
1644                            po::options_description *options) {
1645   options->add_options()
1646     ("force", po::bool_switch(),
1647      "promote even if not cleanly demoted by remote cluster");
1648   at::add_pool_options(positional, options, true);
1649 }
1650
1651 int execute_promote(const po::variables_map &vm,
1652                     const std::vector<std::string> &ceph_global_init_args) {
1653   std::string pool_name;
1654   std::string namespace_name;
1655   size_t arg_index = 0;
1656   int r = utils::get_pool_and_namespace_names(vm, true, true, &pool_name,
1657                                               &namespace_name, &arg_index);
1658   if (r < 0) {
1659     return r;
1660   }
1661
1662   librados::Rados rados;
1663   librados::IoCtx io_ctx;
1664   r = utils::init(pool_name, namespace_name, &rados, &io_ctx);
1665   if (r < 0) {
1666     return r;
1667   }
1668
1669   r = validate_mirroring_enabled(io_ctx);
1670   if (r < 0) {
1671     return r;
1672   }
1673
1674   utils::disable_cache();
1675
1676   std::atomic<unsigned> counter = { 0 };
1677   ImageRequestGenerator<PromoteImageRequest> generator(io_ctx, &counter,
1678                                                        vm["force"].as<bool>());
1679   r = generator.execute();
1680
1681   std::cout << "Promoted " << counter.load() << " mirrored images" << std::endl;
1682   return r;
1683 }
1684
1685 void get_demote_arguments(po::options_description *positional,
1686                            po::options_description *options) {
1687   at::add_pool_options(positional, options, true);
1688 }
1689
1690 int execute_demote(const po::variables_map &vm,
1691                    const std::vector<std::string> &ceph_global_init_args) {
1692   std::string pool_name;
1693   std::string namespace_name;
1694   size_t arg_index = 0;
1695   int r = utils::get_pool_and_namespace_names(vm, true, true, &pool_name,
1696                                               &namespace_name, &arg_index);
1697   if (r < 0) {
1698     return r;
1699   }
1700
1701   librados::Rados rados;
1702   librados::IoCtx io_ctx;
1703   r = utils::init(pool_name, namespace_name, &rados, &io_ctx);
1704   if (r < 0) {
1705     return r;
1706   }
1707
1708   r = validate_mirroring_enabled(io_ctx);
1709   if (r < 0) {
1710     return r;
1711   }
1712
1713   utils::disable_cache();
1714
1715   std::atomic<unsigned> counter { 0 };
1716   ImageRequestGenerator<DemoteImageRequest> generator(io_ctx, &counter);
1717   r = generator.execute();
1718
1719   std::cout << "Demoted " << counter.load() << " mirrored images" << std::endl;
1720   return r;
1721 }
1722
1723 Shell::Action action_bootstrap_create(
1724   {"mirror", "pool", "peer", "bootstrap", "create"}, {},
1725   "Create a peer bootstrap token to import in a remote cluster", "",
1726   &get_peer_bootstrap_create_arguments, &execute_peer_bootstrap_create);
1727 Shell::Action action_bootstreap_import(
1728   {"mirror", "pool", "peer", "bootstrap", "import"}, {},
1729   "Import a peer bootstrap token created from a remote cluster", "",
1730   &get_peer_bootstrap_import_arguments, &execute_peer_bootstrap_import);
1731
1732 Shell::Action action_add(
1733   {"mirror", "pool", "peer", "add"}, {},
1734   "Add a mirroring peer to a pool.", "",
1735   &get_peer_add_arguments, &execute_peer_add);
1736 Shell::Action action_remove(
1737   {"mirror", "pool", "peer", "remove"}, {},
1738   "Remove a mirroring peer from a pool.", "",
1739   &get_peer_remove_arguments, &execute_peer_remove);
1740 Shell::Action action_set(
1741   {"mirror", "pool", "peer", "set"}, {},
1742   "Update mirroring peer settings.", "",
1743   &get_peer_set_arguments, &execute_peer_set);
1744
1745 Shell::Action action_disable(
1746   {"mirror", "pool", "disable"}, {},
1747   "Disable RBD mirroring by default within a pool.", "",
1748   &get_disable_arguments, &execute_disable);
1749 Shell::Action action_enable(
1750   {"mirror", "pool", "enable"}, {},
1751   "Enable RBD mirroring by default within a pool.", "",
1752   &get_enable_arguments, &execute_enable);
1753 Shell::Action action_info(
1754   {"mirror", "pool", "info"}, {},
1755   "Show information about the pool mirroring configuration.", {},
1756   &get_info_arguments, &execute_info);
1757 Shell::Action action_status(
1758   {"mirror", "pool", "status"}, {},
1759   "Show status for all mirrored images in the pool.", {},
1760   &get_status_arguments, &execute_status);
1761 Shell::Action action_promote(
1762   {"mirror", "pool", "promote"}, {},
1763   "Promote all non-primary images in the pool.", {},
1764   &get_promote_arguments, &execute_promote);
1765 Shell::Action action_demote(
1766   {"mirror", "pool", "demote"}, {},
1767   "Demote all primary images in the pool.", {},
1768   &get_demote_arguments, &execute_demote);
1769
1770 } // namespace mirror_pool
1771 } // namespace action
1772 } // namespace rbd