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/Shell.h"
   6 #include "tools/rbd/Utils.h"
   7 #include "include/Context.h"
   8 #include "include/stringify.h"
   9 #include "include/rbd/librbd.hpp"
  10 #include "common/config.h"
  11 #include "common/debug.h"
  12 #include "common/errno.h"
  13 #include "common/Formatter.h"
  14 #include "common/TextTable.h"
  15 #include "common/Throttle.h"
  16 #include "global/global_context.h"
  17 #include <functional>
  18 #include <iostream>
  19 #include <boost/program_options.hpp>
  20 #include <boost/regex.hpp>
  21 #include "include/assert.h"
  22
  23 #include <atomic>
  24
  25 #define dout_context g_ceph_context
  26 #define dout_subsys ceph_subsys_rbd
  27 #undef dout_prefix
  28 #define dout_prefix *_dout << "rbd::action::MirrorPool: "
  29
  30 namespace rbd {
  31 namespace action {
  32 namespace mirror_pool {
  33
  34 namespace at = argument_types;
  35 namespace po = boost::program_options;
  36
  37 namespace {
  38
  39 int validate_uuid(const std::string &uuid) {
  40   boost::regex pattern("^[A-F0-9]{8}-[A-F0-9]{4}-[A-F0-9]{4}-[A-F0-9]{4}-[A-F0-9]{12}$",
  41                        boost::regex::icase);
  42   boost::smatch match;
  43   if (!boost::regex_match(uuid, match, pattern)) {
  44     std::cerr << "rbd: invalid uuid '" << uuid << "'" << std::endl;
  45     return -EINVAL;
  46   }
  47   return 0;
  48 }
  49
  50 void add_uuid_option(po::options_description *positional) {
  51   positional->add_options()
  52     ("uuid", po::value<std::string>(), "peer uuid");
  53 }
  54
  55 int get_uuid(const po::variables_map &vm, size_t arg_index,
  56              std::string *uuid) {
  57   *uuid = utils::get_positional_argument(vm, arg_index);
  58   if (uuid->empty()) {
  59     std::cerr << "rbd: must specify peer uuid" << std::endl;
  60     return -EINVAL;
  61   }
  62   return validate_uuid(*uuid);
  63 }
  64
  65 int get_remote_cluster_spec(const po::variables_map &vm,
  66                             const std::string &spec,
  67                             std::string *remote_client_name,
  68                             std::string *remote_cluster) {
  69   if (vm.count("remote-client-name")) {
  70     *remote_client_name = vm["remote-client-name"].as<std::string>();
  71   }
  72   if (vm.count("remote-cluster")) {
  73     *remote_cluster = vm["remote-cluster"].as<std::string>();
  74   }
  75
  76   if (!spec.empty()) {
  77     boost::regex pattern("^(?:(client\\.[^@]+)@)?([^/@]+)$");
  78     boost::smatch match;
  79     if (!boost::regex_match(spec, match, pattern)) {
  80       std::cerr << "rbd: invalid spec '" << spec << "'" << std::endl;
  81       return -EINVAL;
  82     }
  83     if (match[1].matched) {
  84       *remote_client_name = match[1];
  85     }
  86     *remote_cluster = match[2];
  87   }
  88
  89   if (remote_cluster->empty()) {
  90     std::cerr << "rbd: remote cluster was not specified" << std::endl;
  91     return -EINVAL;
  92   }
  93   return 0;
  94 }
  95
  96 void format_mirror_peers(const std::string &config_path,
  97                          at::Format::Formatter formatter,
  98                          const std::vector<librbd::mirror_peer_t> &peers) {
  99   if (formatter != nullptr) {
 100     formatter->open_array_section("peers");
 101     for (auto &peer : peers) {
 102       formatter->open_object_section("peer");
 103       formatter->dump_string("uuid", peer.uuid);
 104       formatter->dump_string("cluster_name", peer.cluster_name);
 105       formatter->dump_string("client_name", peer.client_name);
 106       formatter->close_section();
 107     }
 108     formatter->close_section();
 109   } else {
 110     std::cout << "Peers: ";
 111     if (peers.empty()) {
 112       std::cout << "none" << std::endl;
 113     } else {
 114       TextTable tbl;
 115       tbl.define_column("", TextTable::LEFT, TextTable::LEFT);
 116       tbl.define_column("UUID", TextTable::LEFT, TextTable::LEFT);
 117       tbl.define_column("NAME", TextTable::LEFT, TextTable::LEFT);
 118       tbl.define_column("CLIENT", TextTable::LEFT, TextTable::LEFT);
 119       for (auto &peer : peers) {
 120         tbl << " "
 121             << peer.uuid
 122             << peer.cluster_name
 123             << peer.client_name
 124             << TextTable::endrow;
 125       }
 126       std::cout << std::endl << tbl;
 127     }
 128   }
 129 }
 130
 131 class ImageRequestBase {
 132 public:
 133   void send() {
 134     dout(20) << this << " " << __func__ << ": image_name=" << m_image_name
 135              << dendl;
 136
 137     auto ctx = new FunctionContext([this](int r) {
 138         handle_finalize(r);
 139       });
 140
 141     // will pause here until slots are available
 142     m_finalize_ctx = m_throttle.start_op(ctx);
 143
 144     open_image();
 145   }
 146
 147 protected:
 148   ImageRequestBase(librados::IoCtx &io_ctx, OrderedThrottle &throttle,
 149                    const std::string &image_name)
 150     : m_io_ctx(io_ctx), m_throttle(throttle), m_image_name(image_name) {
 151   }
 152   virtual ~ImageRequestBase() {
 153   }
 154
 155   virtual bool skip_get_info() const {
 156     return false;
 157   }
 158   virtual void get_info(librbd::Image &image, librbd::mirror_image_info_t *info,
 159                         librbd::RBD::AioCompletion *aio_comp) {
 160     image.aio_mirror_image_get_info(info, sizeof(librbd::mirror_image_info_t),
 161                                     aio_comp);
 162   }
 163
 164   virtual bool skip_action(const librbd::mirror_image_info_t &info) const {
 165     return false;
 166   }
 167   virtual void execute_action(librbd::Image &image,
 168                               librbd::RBD::AioCompletion *aio_comp) = 0;
 169   virtual void handle_execute_action(int r) {
 170     dout(20) << this << " " << __func__ << ": r=" << r << dendl;
 171
 172     if (r < 0 && r != -ENOENT) {
 173       std::cerr << "rbd: failed to " << get_action_type() << " image "
 174                 << m_image_name << ": " << cpp_strerror(r) << std::endl;
 175       m_ret_val = r;
 176     }
 177
 178     close_image();
 179   }
 180
 181   virtual void finalize_action() {
 182   }
 183   virtual std::string get_action_type() const = 0;
 184
 185 private:
 186   /**
 187    * @verbatim
 188    *
 189    * <start>
 190    *    |
 191    *    v
 192    * OPEN_IMAGE
 193    *    |
 194    *    v
 195    * GET_INFO
 196    *    |
 197    *    v
 198    * EXECUTE_ACTION
 199    *    |
 200    *    v
 201    * CLOSE_IMAGE
 202    *    |
 203    *    v
 204    * FINALIZE_ACTION
 205    *    |
 206    *    v
 207    * <finish>
 208    *
 209    * @endverbatim
 210    */
 211
 212   librados::IoCtx &m_io_ctx;
 213   OrderedThrottle &m_throttle;
 214   const std::string m_image_name;
 215
 216   librbd::Image m_image;
 217   Context *m_finalize_ctx;
 218
 219   librbd::mirror_image_info_t m_mirror_image_info;
 220
 221   int m_ret_val = 0;
 222
 223   void open_image() {
 224     dout(20) << this << " " << __func__ << dendl;
 225
 226     librbd::RBD rbd;
 227     auto aio_completion = utils::create_aio_completion<
 228       ImageRequestBase, &ImageRequestBase::handle_open_image>(this);
 229     rbd.aio_open(m_io_ctx, m_image, m_image_name.c_str(), nullptr,
 230                  aio_completion);
 231   }
 232
 233   void handle_open_image(int r) {
 234     dout(20) << this << " " << __func__ << ": r=" << r << dendl;
 235
 236     if (r < 0) {
 237       std::cerr << "rbd: failed to open image "
 238                 << m_image_name << ": " << cpp_strerror(r) << std::endl;
 239       m_finalize_ctx->complete(r);
 240       return;
 241     }
 242
 243     get_info();
 244   }
 245
 246   void get_info() {
 247     if (skip_get_info()) {
 248       execute_action();
 249       return;
 250     }
 251     dout(20) << this << " " << __func__ << dendl;
 252
 253     auto aio_completion = utils::create_aio_completion<
 254       ImageRequestBase, &ImageRequestBase::handle_get_info>(this);
 255     get_info(m_image, &m_mirror_image_info, aio_completion);
 256   }
 257
 258   void handle_get_info(int r) {
 259     dout(20) << this << " " << __func__ << ": r=" << r << dendl;
 260
 261     if (r == -ENOENT) {
 262       close_image();
 263       return;
 264     } else if (r < 0) {
 265       std::cerr << "rbd: failed to retrieve mirror image info for "
 266                 << m_image_name << ": " << cpp_strerror(r) << std::endl;
 267       m_ret_val = r;
 268       close_image();
 269       return;
 270     }
 271
 272     execute_action();
 273   }
 274
 275   void execute_action() {
 276     if (skip_action(m_mirror_image_info)) {
 277       close_image();
 278       return;
 279     }
 280     dout(20) << this << " " << __func__ << dendl;
 281
 282     auto aio_completion = utils::create_aio_completion<
 283       ImageRequestBase, &ImageRequestBase::handle_execute_action>(this);
 284     execute_action(m_image, aio_completion);
 285   }
 286
 287   void close_image() {
 288     dout(20) << this << " " << __func__ << dendl;
 289
 290     auto aio_completion = utils::create_aio_completion<
 291       ImageRequestBase, &ImageRequestBase::handle_close_image>(this);
 292     m_image.aio_close(aio_completion);
 293   }
 294
 295   void handle_close_image(int r) {
 296     dout(20) << this << " " << __func__ << ": r=" << r << dendl;
 297
 298     if (r < 0) {
 299       std::cerr << "rbd: failed to close image "
 300                 << m_image_name << ": " << cpp_strerror(r) << std::endl;
 301     }
 302
 303     m_finalize_ctx->complete(r);
 304   }
 305
 306   void handle_finalize(int r) {
 307     dout(20) << this << " " << __func__ << ": r=" << r << dendl;
 308
 309     if (r == 0 && m_ret_val < 0) {
 310       r = m_ret_val;
 311     }
 312     if (r >= 0) {
 313       finalize_action();
 314     }
 315     m_throttle.end_op(r);
 316   }
 317
 318 };
 319
 320 class PromoteImageRequest : public ImageRequestBase {
 321 public:
 322   PromoteImageRequest(librados::IoCtx &io_ctx, OrderedThrottle &throttle,
 323                       const std::string &image_name, std::atomic<unsigned> *counter,
 324                       bool force)
 325     : ImageRequestBase(io_ctx, throttle, image_name), m_counter(counter),
 326       m_force(force) {
 327   }
 328
 329 protected:
 330   bool skip_action(const librbd::mirror_image_info_t &info) const override {
 331     return info.primary;
 332   }
 333
 334   void execute_action(librbd::Image &image,
 335                       librbd::RBD::AioCompletion *aio_comp) override {
 336     image.aio_mirror_image_promote(m_force, aio_comp);
 337   }
 338
 339   void handle_execute_action(int r) override {
 340     if (r >= 0) {
 341       (*m_counter)++;
 342     }
 343   }
 344
 345   std::string get_action_type() const override {
 346     return "promote";
 347   }
 348
 349 private:
 350   std::atomic<unsigned> *m_counter = nullptr;
 351   bool m_force;
 352 };
 353
 354 class DemoteImageRequest : public ImageRequestBase {
 355 public:
 356   DemoteImageRequest(librados::IoCtx &io_ctx, OrderedThrottle &throttle,
 357                      const std::string &image_name, std::atomic<unsigned> *counter)
 358     : ImageRequestBase(io_ctx, throttle, image_name), m_counter(counter) {
 359   }
 360
 361 protected:
 362   bool skip_action(const librbd::mirror_image_info_t &info) const override {
 363     return !info.primary;
 364   }
 365
 366   void execute_action(librbd::Image &image,
 367                       librbd::RBD::AioCompletion *aio_comp) override {
 368     image.aio_mirror_image_demote(aio_comp);
 369   }
 370   void handle_execute_action(int r) override {
 371     if (r >= 0) {
 372       (*m_counter)++;
 373     }
 374     ImageRequestBase::handle_execute_action(r);
 375   }
 376
 377   std::string get_action_type() const override {
 378     return "demote";
 379   }
 380
 381 private:
 382   std::atomic<unsigned> *m_counter = nullptr;
 383 };
 384
 385 class StatusImageRequest : public ImageRequestBase {
 386 public:
 387   StatusImageRequest(librados::IoCtx &io_ctx, OrderedThrottle &throttle,
 388                      const std::string &image_name,
 389                      at::Format::Formatter formatter)
 390     : ImageRequestBase(io_ctx, throttle, image_name),
 391       m_formatter(formatter) {
 392   }
 393
 394 protected:
 395   bool skip_get_info() const override {
 396     return true;
 397   }
 398
 399   void execute_action(librbd::Image &image,
 400                       librbd::RBD::AioCompletion *aio_comp) override {
 401     image.aio_mirror_image_get_status(&m_mirror_image_status,
 402                                       sizeof(m_mirror_image_status), aio_comp);
 403   }
 404
 405   void finalize_action() override {
 406     std::string state = utils::mirror_image_status_state(m_mirror_image_status);
 407     std::string last_update = (
 408       m_mirror_image_status.last_update == 0 ?
 409         "" : utils::timestr(m_mirror_image_status.last_update));
 410
 411     if (m_formatter != nullptr) {
 412       m_formatter->open_object_section("image");
 413       m_formatter->dump_string("name", m_mirror_image_status.name);
 414       m_formatter->dump_string("global_id",
 415                                m_mirror_image_status.info.global_id);
 416       m_formatter->dump_string("state", state);
 417       m_formatter->dump_string("description",
 418                                m_mirror_image_status.description);
 419       m_formatter->dump_string("last_update", last_update);
 420       m_formatter->close_section(); // image
 421     } else {
 422       std::cout << "\n" << m_mirror_image_status.name << ":\n"
 423                 << "  global_id:   "
 424                 << m_mirror_image_status.info.global_id << "\n"
 425                 << "  state:       " << state << "\n"
 426                 << "  description: "
 427                 << m_mirror_image_status.description << "\n"
 428                 << "  last_update: " << last_update << std::endl;
 429     }
 430   }
 431
 432   std::string get_action_type() const override {
 433     return "status";
 434   }
 435
 436 private:
 437   at::Format::Formatter m_formatter;
 438   librbd::mirror_image_status_t m_mirror_image_status;
 439
 440 };
 441
 442 template <typename RequestT>
 443 class ImageRequestAllocator {
 444 public:
 445   template <class... Args>
 446   RequestT *operator()(librados::IoCtx &io_ctx, OrderedThrottle &throttle,
 447                        const std::string &image_name, Args&&... args) {
 448     return new RequestT(io_ctx, throttle, image_name,
 449                         std::forward<Args>(args)...);
 450   }
 451 };
 452
 453 template <typename RequestT>
 454 class ImageRequestGenerator {
 455 public:
 456   template <class... Args>
 457   ImageRequestGenerator(librados::IoCtx &io_ctx, Args&&... args)
 458     : m_io_ctx(io_ctx),
 459       m_factory(std::bind(ImageRequestAllocator<RequestT>(),
 460                           std::ref(m_io_ctx), std::ref(m_throttle),
 461                           std::placeholders::_1, std::forward<Args>(args)...)),
 462       m_throttle(g_conf->get_val<int64_t>("rbd_concurrent_management_ops"),
 463                  true) {
 464   }
 465
 466   int execute() {
 467     // use the alphabetical list of image names for pool-level
 468     // mirror image operations
 469     librbd::RBD rbd;
 470     int r = rbd.list(m_io_ctx, m_image_names);
 471     if (r < 0 && r != -ENOENT) {
 472       std::cerr << "rbd: failed to list images within pool" << std::endl;
 473       return r;
 474     }
 475
 476     for (auto &image_name : m_image_names) {
 477       auto request = m_factory(image_name);
 478       request->send();
 479     }
 480
 481     return m_throttle.wait_for_ret();
 482   }
 483 private:
 484   typedef std::function<RequestT*(const std::string&)>  Factory;
 485
 486   librados::IoCtx &m_io_ctx;
 487   Factory m_factory;
 488
 489   OrderedThrottle m_throttle;
 490
 491   std::vector<std::string> m_image_names;
 492
 493 };
 494
 495 } // anonymous namespace
 496
 497 void get_peer_add_arguments(po::options_description *positional,
 498                             po::options_description *options) {
 499   at::add_pool_options(positional, options);
 500   positional->add_options()
 501     ("remote-cluster-spec", "remote cluster spec\n"
 502      "(example: [<client name>@]<cluster name>");
 503   options->add_options()
 504     ("remote-client-name", po::value<std::string>(), "remote client name")
 505     ("remote-cluster", po::value<std::string>(), "remote cluster name");
 506 }
 507
 508 int execute_peer_add(const po::variables_map &vm) {
 509   size_t arg_index = 0;
 510   std::string pool_name = utils::get_pool_name(vm, &arg_index);
 511
 512   std::string remote_client_name = g_ceph_context->_conf->name.to_str();
 513   std::string remote_cluster;
 514   int r = get_remote_cluster_spec(
 515     vm, utils::get_positional_argument(vm, arg_index),
 516     &remote_client_name, &remote_cluster);
 517   if (r < 0) {
 518     return r;
 519   }
 520
 521   std::string config_path;
 522   if (vm.count(at::CONFIG_PATH)) {
 523     config_path = vm[at::CONFIG_PATH].as<std::string>();
 524   }
 525
 526   librados::Rados rados;
 527   librados::IoCtx io_ctx;
 528   r = utils::init(pool_name, &rados, &io_ctx);
 529   if (r < 0) {
 530     return r;
 531   }
 532
 533   librbd::RBD rbd;
 534   rbd_mirror_mode_t mirror_mode;
 535   r = rbd.mirror_mode_get(io_ctx, &mirror_mode);
 536   if (r < 0) {
 537     std::cerr << "rbd: failed to retrieve mirror mode: "
 538               << cpp_strerror(r) << std::endl;
 539     return r;
 540   }
 541
 542   if (mirror_mode == RBD_MIRROR_MODE_DISABLED) {
 543     std::cerr << "rbd: failed to add mirror peer: "
 544               << "mirroring must be enabled on the pool "
 545               << pool_name << std::endl;
 546     return -EINVAL;
 547   }
 548
 549   // TODO: temporary restriction to prevent adding multiple peers
 550   // until rbd-mirror daemon can properly handle the scenario
 551   std::vector<librbd::mirror_peer_t> mirror_peers;
 552   r = rbd.mirror_peer_list(io_ctx, &mirror_peers);
 553   if (r < 0) {
 554     std::cerr << "rbd: failed to list mirror peers" << std::endl;
 555     return r;
 556   }
 557   if (!mirror_peers.empty()) {
 558     std::cerr << "rbd: multiple peers are not currently supported" << std::endl;
 559     return -EINVAL;
 560   }
 561
 562   std::string uuid;
 563   r = rbd.mirror_peer_add(io_ctx, &uuid, remote_cluster, remote_client_name);
 564   if (r < 0) {
 565     std::cerr << "rbd: error adding mirror peer" << std::endl;
 566     return r;
 567   }
 568
 569   std::cout << uuid << std::endl;
 570   return 0;
 571 }
 572
 573 void get_peer_remove_arguments(po::options_description *positional,
 574                                po::options_description *options) {
 575   at::add_pool_options(positional, options);
 576   add_uuid_option(positional);
 577 }
 578
 579 int execute_peer_remove(const po::variables_map &vm) {
 580   size_t arg_index = 0;
 581   std::string pool_name = utils::get_pool_name(vm, &arg_index);
 582
 583   std::string uuid;
 584   int r = get_uuid(vm, arg_index, &uuid);
 585   if (r < 0) {
 586     return r;
 587   }
 588
 589   librados::Rados rados;
 590   librados::IoCtx io_ctx;
 591   r = utils::init(pool_name, &rados, &io_ctx);
 592   if (r < 0) {
 593     return r;
 594   }
 595
 596   librbd::RBD rbd;
 597   r = rbd.mirror_peer_remove(io_ctx, uuid);
 598   if (r < 0) {
 599     std::cerr << "rbd: error removing mirror peer" << std::endl;
 600     return r;
 601   }
 602   return 0;
 603 }
 604
 605 void get_peer_set_arguments(po::options_description *positional,
 606                             po::options_description *options) {
 607   at::add_pool_options(positional, options);
 608   add_uuid_option(positional);
 609   positional->add_options()
 610     ("key", "peer parameter [client or cluster]")
 611     ("value", "new client or cluster name");
 612 }
 613
 614 int execute_peer_set(const po::variables_map &vm) {
 615   size_t arg_index = 0;
 616   std::string pool_name = utils::get_pool_name(vm, &arg_index);
 617
 618   std::string uuid;
 619   int r = get_uuid(vm, arg_index++, &uuid);
 620   if (r < 0) {
 621     return r;
 622   }
 623
 624   std::string key = utils::get_positional_argument(vm, arg_index++);
 625   if (key != "client" && key != "cluster") {
 626     std::cerr << "rbd: must specify 'client' or 'cluster' key." << std::endl;
 627     return -EINVAL;
 628   }
 629
 630   std::string value = utils::get_positional_argument(vm, arg_index++);
 631   if (value.empty()) {
 632     std::cerr << "rbd: must specify new " << key << " value." << std::endl;
 633   }
 634
 635   librados::Rados rados;
 636   librados::IoCtx io_ctx;
 637   r = utils::init(pool_name, &rados, &io_ctx);
 638   if (r < 0) {
 639     return r;
 640   }
 641
 642   librbd::RBD rbd;
 643   if (key == "client") {
 644     r = rbd.mirror_peer_set_client(io_ctx, uuid.c_str(), value.c_str());
 645   } else {
 646     r = rbd.mirror_peer_set_cluster(io_ctx, uuid.c_str(), value.c_str());
 647   }
 648   if (r < 0) {
 649     return r;
 650   }
 651   return 0;
 652 }
 653
 654 void get_disable_arguments(po::options_description *positional,
 655                            po::options_description *options) {
 656   at::add_pool_options(positional, options);
 657 }
 658
 659 void get_enable_arguments(po::options_description *positional,
 660                           po::options_description *options) {
 661   at::add_pool_options(positional, options);
 662   positional->add_options()
 663     ("mode", "mirror mode [image or pool]");
 664 }
 665
 666 int execute_enable_disable(const std::string &pool_name,
 667                            rbd_mirror_mode_t next_mirror_mode,
 668                            const std::string &mode) {
 669   librados::Rados rados;
 670   librados::IoCtx io_ctx;
 671   rbd_mirror_mode_t current_mirror_mode;
 672
 673   int r = utils::init(pool_name, &rados, &io_ctx);
 674   if (r < 0) {
 675     return r;
 676   }
 677
 678   librbd::RBD rbd;
 679   r = rbd.mirror_mode_get(io_ctx, &current_mirror_mode);
 680   if (r < 0) {
 681     std::cerr << "rbd: failed to retrieve mirror mode: "
 682               << cpp_strerror(r) << std::endl;
 683     return r;
 684   }
 685
 686   if (current_mirror_mode == next_mirror_mode) {
 687     if (mode == "disabled") {
 688       std::cout << "mirroring is already " << mode << std::endl;
 689     } else {
 690       std::cout << "mirroring is already configured for "
 691                 << mode << " mode" << std::endl;
 692     }
 693     return 0;
 694   } else if (next_mirror_mode == RBD_MIRROR_MODE_IMAGE &&
 695              current_mirror_mode == RBD_MIRROR_MODE_POOL) {
 696     std::cout << "note: changing mirroring mode from pool to image"
 697               << std::endl;
 698   } else if (next_mirror_mode == RBD_MIRROR_MODE_POOL &&
 699              current_mirror_mode == RBD_MIRROR_MODE_IMAGE) {
 700     std::cout << "note: changing mirroring mode from image to pool"
 701               << std::endl;
 702   }
 703
 704   r = rbd.mirror_mode_set(io_ctx, next_mirror_mode);
 705   if (r < 0) {
 706     return r;
 707   }
 708   return 0;
 709 }
 710
 711 int execute_disable(const po::variables_map &vm) {
 712   size_t arg_index = 0;
 713   std::string pool_name = utils::get_pool_name(vm, &arg_index);
 714
 715   return execute_enable_disable(pool_name, RBD_MIRROR_MODE_DISABLED,
 716                                 "disabled");
 717 }
 718
 719 int execute_enable(const po::variables_map &vm) {
 720   size_t arg_index = 0;
 721   std::string pool_name = utils::get_pool_name(vm, &arg_index);
 722
 723   rbd_mirror_mode_t mirror_mode;
 724   std::string mode = utils::get_positional_argument(vm, arg_index++);
 725   if (mode == "image") {
 726     mirror_mode = RBD_MIRROR_MODE_IMAGE;
 727   } else if (mode == "pool") {
 728     mirror_mode = RBD_MIRROR_MODE_POOL;
 729   } else {
 730     std::cerr << "rbd: must specify 'image' or 'pool' mode." << std::endl;
 731     return -EINVAL;
 732   }
 733
 734   return execute_enable_disable(pool_name, mirror_mode, mode);
 735 }
 736
 737 void get_info_arguments(po::options_description *positional,
 738                         po::options_description *options) {
 739   at::add_pool_options(positional, options);
 740   at::add_format_options(options);
 741 }
 742
 743 int execute_info(const po::variables_map &vm) {
 744   size_t arg_index = 0;
 745   std::string pool_name = utils::get_pool_name(vm, &arg_index);
 746
 747   at::Format::Formatter formatter;
 748   int r = utils::get_formatter(vm, &formatter);
 749   if (r < 0) {
 750     return r;
 751   }
 752
 753   std::string config_path;
 754   if (vm.count(at::CONFIG_PATH)) {
 755     config_path = vm[at::CONFIG_PATH].as<std::string>();
 756   }
 757
 758   librados::Rados rados;
 759   librados::IoCtx io_ctx;
 760   r = utils::init(pool_name, &rados, &io_ctx);
 761   if (r < 0) {
 762     return r;
 763   }
 764
 765   librbd::RBD rbd;
 766   rbd_mirror_mode_t mirror_mode;
 767   r = rbd.mirror_mode_get(io_ctx, &mirror_mode);
 768   if (r < 0) {
 769     return r;
 770   }
 771
 772   std::vector<librbd::mirror_peer_t> mirror_peers;
 773   r = rbd.mirror_peer_list(io_ctx, &mirror_peers);
 774   if (r < 0) {
 775     return r;
 776   }
 777
 778   std::string mirror_mode_desc;
 779   switch (mirror_mode) {
 780   case RBD_MIRROR_MODE_DISABLED:
 781     mirror_mode_desc = "disabled";
 782     break;
 783   case RBD_MIRROR_MODE_IMAGE:
 784     mirror_mode_desc = "image";
 785     break;
 786   case RBD_MIRROR_MODE_POOL:
 787     mirror_mode_desc = "pool";
 788     break;
 789   default:
 790     mirror_mode_desc = "unknown";
 791     break;
 792   }
 793
 794   if (formatter != nullptr) {
 795     formatter->open_object_section("mirror");
 796     formatter->dump_string("mode", mirror_mode_desc);
 797   } else {
 798     std::cout << "Mode: " << mirror_mode_desc << std::endl;
 799   }
 800
 801   if (mirror_mode != RBD_MIRROR_MODE_DISABLED) {
 802     format_mirror_peers(config_path, formatter, mirror_peers);
 803   }
 804   if (formatter != nullptr) {
 805     formatter->close_section();
 806     formatter->flush(std::cout);
 807   }
 808   return 0;
 809 }
 810
 811 void get_status_arguments(po::options_description *positional,
 812                           po::options_description *options) {
 813   at::add_pool_options(positional, options);
 814   at::add_format_options(options);
 815   at::add_verbose_option(options);
 816 }
 817
 818 int execute_status(const po::variables_map &vm) {
 819   size_t arg_index = 0;
 820   std::string pool_name = utils::get_pool_name(vm, &arg_index);
 821
 822   at::Format::Formatter formatter;
 823   int r = utils::get_formatter(vm, &formatter);
 824   if (r < 0) {
 825     return r;
 826   }
 827
 828   bool verbose = vm[at::VERBOSE].as<bool>();
 829
 830   std::string config_path;
 831   if (vm.count(at::CONFIG_PATH)) {
 832     config_path = vm[at::CONFIG_PATH].as<std::string>();
 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   librbd::RBD rbd;
 843
 844   std::map<librbd::mirror_image_status_state_t, int> states;
 845   r = rbd.mirror_image_status_summary(io_ctx, &states);
 846   if (r < 0) {
 847     std::cerr << "rbd: failed to get status summary for mirrored images: "
 848               << cpp_strerror(r) << std::endl;
 849     return r;
 850   }
 851
 852   if (formatter != nullptr) {
 853     formatter->open_object_section("status");
 854   }
 855
 856   enum Health {Ok = 0, Warning = 1, Error = 2} health = Ok;
 857   const char *names[] = {"OK", "WARNING", "ERROR"};
 858   int total = 0;
 859
 860   for (auto &it : states) {
 861     auto &state = it.first;
 862     if (health < Warning &&
 863         (state != MIRROR_IMAGE_STATUS_STATE_REPLAYING &&
 864          state != MIRROR_IMAGE_STATUS_STATE_STOPPED)) {
 865       health = Warning;
 866     }
 867     if (health < Error &&
 868         state == MIRROR_IMAGE_STATUS_STATE_ERROR) {
 869       health = Error;
 870     }
 871     total += it.second;
 872   }
 873
 874   if (formatter != nullptr) {
 875     formatter->open_object_section("summary");
 876     formatter->dump_string("health", names[health]);
 877     formatter->open_object_section("states");
 878     for (auto &it : states) {
 879       std::string state_name = utils::mirror_image_status_state(it.first);
 880       formatter->dump_int(state_name.c_str(), it.second);
 881     }
 882     formatter->close_section(); // states
 883     formatter->close_section(); // summary
 884   } else {
 885     std::cout << "health: " << names[health] << std::endl;
 886     std::cout << "images: " << total << " total" << std::endl;
 887     for (auto &it : states) {
 888       std::cout << "    " << it.second << " "
 889                 << utils::mirror_image_status_state(it.first) << std::endl;
 890     }
 891   }
 892
 893   int ret = 0;
 894
 895   if (verbose) {
 896     if (formatter != nullptr) {
 897       formatter->open_array_section("images");
 898     }
 899
 900     ImageRequestGenerator<StatusImageRequest> generator(io_ctx, formatter);
 901     ret = generator.execute();
 902
 903     if (formatter != nullptr) {
 904       formatter->close_section(); // images
 905     }
 906   }
 907
 908   if (formatter != nullptr) {
 909     formatter->close_section(); // status
 910     formatter->flush(std::cout);
 911   }
 912
 913   return ret;
 914 }
 915
 916 void get_promote_arguments(po::options_description *positional,
 917                            po::options_description *options) {
 918   options->add_options()
 919     ("force", po::bool_switch(),
 920      "promote even if not cleanly demoted by remote cluster");
 921   at::add_pool_options(positional, options);
 922 }
 923
 924 int execute_promote(const po::variables_map &vm) {
 925   size_t arg_index = 0;
 926   std::string pool_name = utils::get_pool_name(vm, &arg_index);
 927
 928   librados::Rados rados;
 929   librados::IoCtx io_ctx;
 930   int r = utils::init(pool_name, &rados, &io_ctx);
 931   if (r < 0) {
 932     return r;
 933   }
 934
 935   std::atomic<unsigned> counter = { 0 };
 936   ImageRequestGenerator<PromoteImageRequest> generator(io_ctx, &counter,
 937                                                        vm["force"].as<bool>());
 938   r = generator.execute();
 939
 940   std::cout << "Promoted " << counter.load() << " mirrored images" << std::endl;
 941   return r;
 942 }
 943
 944 void get_demote_arguments(po::options_description *positional,
 945                            po::options_description *options) {
 946   at::add_pool_options(positional, options);
 947 }
 948
 949 int execute_demote(const po::variables_map &vm) {
 950   size_t arg_index = 0;
 951   std::string pool_name = utils::get_pool_name(vm, &arg_index);
 952
 953   librados::Rados rados;
 954   librados::IoCtx io_ctx;
 955   int r = utils::init(pool_name, &rados, &io_ctx);
 956   if (r < 0) {
 957     return r;
 958   }
 959
 960   std::atomic<unsigned> counter { 0 };
 961   ImageRequestGenerator<DemoteImageRequest> generator(io_ctx, &counter);
 962   r = generator.execute();
 963
 964   std::cout << "Demoted " << counter.load() << " mirrored images" << std::endl;
 965   return r;
 966 }
 967
 968 Shell::Action action_add(
 969   {"mirror", "pool", "peer", "add"}, {},
 970   "Add a mirroring peer to a pool.", "",
 971   &get_peer_add_arguments, &execute_peer_add);
 972 Shell::Action action_remove(
 973   {"mirror", "pool", "peer", "remove"}, {},
 974   "Remove a mirroring peer from a pool.", "",
 975   &get_peer_remove_arguments, &execute_peer_remove);
 976 Shell::Action action_set(
 977   {"mirror", "pool", "peer", "set"}, {},
 978   "Update mirroring peer settings.", "",
 979   &get_peer_set_arguments, &execute_peer_set);
 980
 981 Shell::Action action_disable(
 982   {"mirror", "pool", "disable"}, {},
 983   "Disable RBD mirroring by default within a pool.", "",
 984   &get_disable_arguments, &execute_disable);
 985 Shell::Action action_enable(
 986   {"mirror", "pool", "enable"}, {},
 987   "Enable RBD mirroring by default within a pool.", "",
 988   &get_enable_arguments, &execute_enable);
 989 Shell::Action action_info(
 990   {"mirror", "pool", "info"}, {},
 991   "Show information about the pool mirroring configuration.", {},
 992   &get_info_arguments, &execute_info);
 993 Shell::Action action_status(
 994   {"mirror", "pool", "status"}, {},
 995   "Show status for all mirrored images in the pool.", {},
 996   &get_status_arguments, &execute_status);
 997 Shell::Action action_promote(
 998   {"mirror", "pool", "promote"}, {},
 999   "Promote all non-primary images in the pool.", {},
1000   &get_promote_arguments, &execute_promote);
1001 Shell::Action action_demote(
1002   {"mirror", "pool", "demote"}, {},
1003   "Demote all primary images in the pool.", {},
1004   &get_demote_arguments, &execute_demote);
1005
1006 } // namespace mirror_pool
1007 } // namespace action
1008 } // namespace rbd