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