ceph/src/mon/FSCommands.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3 /*
   4  * Ceph - scalable distributed file system
   5  *
   6  * Copyright (C) 2017 Red Hat Ltd
   7  *
   8  * This is free software; you can redistribute it and/or
   9  * modify it under the terms of the GNU Lesser General Public
  10  * License version 2.1, as published by the Free Software
  11  * Foundation.  See file COPYING.
  12  *
  13  */
  14
  15
  16 #include "OSDMonitor.h"
  17
  18 #include "FSCommands.h"
  19 #include "MDSMonitor.h"
  20 #include "MgrStatMonitor.h"
  21 #include "mds/cephfs_features.h"
  22
  23 using TOPNSPC::common::cmd_getval;
  24
  25 using std::dec;
  26 using std::hex;
  27 using std::list;
  28 using std::map;
  29 using std::make_pair;
  30 using std::pair;
  31 using std::set;
  32 using std::string;
  33 using std::to_string;
  34 using std::vector;
  35
  36 using ceph::bufferlist;
  37 using ceph::decode;
  38 using ceph::encode;
  39 using ceph::ErasureCodeInterfaceRef;
  40 using ceph::ErasureCodeProfile;
  41 using ceph::Formatter;
  42 using ceph::JSONFormatter;
  43 using ceph::make_message;
  44 using ceph::mono_clock;
  45 using ceph::mono_time;
  46
  47 class FlagSetHandler : public FileSystemCommandHandler
  48 {
  49   public:
  50   FlagSetHandler()
  51     : FileSystemCommandHandler("fs flag set")
  52   {
  53   }
  54
  55   int handle(
  56       Monitor *mon,
  57       FSMap& fsmap,
  58       MonOpRequestRef op,
  59       const cmdmap_t& cmdmap,
  60       std::ostream &ss) override
  61   {
  62     string flag_name;
  63     cmd_getval(cmdmap, "flag_name", flag_name);
  64
  65     string flag_val;
  66     cmd_getval(cmdmap, "val", flag_val);
  67
  68     bool sure = false;
  69     cmd_getval(cmdmap, "yes_i_really_mean_it", sure);
  70
  71     if (flag_name == "enable_multiple") {
  72       bool flag_bool = false;
  73       int r = parse_bool(flag_val, &flag_bool, ss);
  74       if (r != 0) {
  75         ss << "Invalid boolean value '" << flag_val << "'";
  76         return r;
  77       }
  78
  79       fsmap.set_enable_multiple(flag_bool);
  80       return 0;
  81     } else {
  82       ss << "Unknown flag '" << flag_name << "'";
  83       return -EINVAL;
  84     }
  85   }
  86 };
  87
  88 class FailHandler : public FileSystemCommandHandler
  89 {
  90   public:
  91   FailHandler()
  92     : FileSystemCommandHandler("fs fail")
  93   {
  94   }
  95
  96   int handle(
  97       Monitor* mon,
  98       FSMap& fsmap,
  99       MonOpRequestRef op,
 100       const cmdmap_t& cmdmap,
 101       std::ostream& ss) override
 102   {
 103     if (!mon->osdmon()->is_writeable()) {
 104       // not allowed to write yet, so retry when we can
 105       mon->osdmon()->wait_for_writeable(op, new PaxosService::C_RetryMessage(mon->mdsmon(), op));
 106       return -EAGAIN;
 107     }
 108
 109     std::string fs_name;
 110     if (!cmd_getval(cmdmap, "fs_name", fs_name) || fs_name.empty()) {
 111       ss << "Missing filesystem name";
 112       return -EINVAL;
 113     }
 114
 115     auto fs = fsmap.get_filesystem(fs_name);
 116
 117     auto f = [](auto fs) {
 118       fs->mds_map.set_flag(CEPH_MDSMAP_NOT_JOINABLE);
 119     };
 120     fsmap.modify_filesystem(fs->fscid, std::move(f));
 121
 122     std::vector<mds_gid_t> to_fail;
 123     for (const auto& p : fs->mds_map.get_mds_info()) {
 124       to_fail.push_back(p.first);
 125     }
 126
 127     for (const auto& gid : to_fail) {
 128       mon->mdsmon()->fail_mds_gid(fsmap, gid);
 129     }
 130     if (!to_fail.empty()) {
 131       mon->osdmon()->propose_pending();
 132     }
 133
 134     ss << fs_name;
 135     ss << " marked not joinable; MDS cannot join the cluster. All MDS ranks marked failed.";
 136
 137     return 0;
 138   }
 139 };
 140
 141 class FsNewHandler : public FileSystemCommandHandler
 142 {
 143   public:
 144   explicit FsNewHandler(Paxos *paxos)
 145     : FileSystemCommandHandler("fs new"), m_paxos(paxos)
 146   {
 147   }
 148
 149   bool batched_propose() override {
 150     return true;
 151   }
 152
 153   int handle(
 154       Monitor *mon,
 155       FSMap& fsmap,
 156       MonOpRequestRef op,
 157       const cmdmap_t& cmdmap,
 158       std::ostream &ss) override
 159   {
 160     ceph_assert(m_paxos->is_plugged());
 161
 162     string metadata_name;
 163     cmd_getval(cmdmap, "metadata", metadata_name);
 164     int64_t metadata = mon->osdmon()->osdmap.lookup_pg_pool_name(metadata_name);
 165     if (metadata < 0) {
 166       ss << "pool '" << metadata_name << "' does not exist";
 167       return -ENOENT;
 168     }
 169
 170     string data_name;
 171     cmd_getval(cmdmap, "data", data_name);
 172     int64_t data = mon->osdmon()->osdmap.lookup_pg_pool_name(data_name);
 173     if (data < 0) {
 174       ss << "pool '" << data_name << "' does not exist";
 175       return -ENOENT;
 176     }
 177     if (data == 0) {
 178       ss << "pool '" << data_name << "' has id 0, which CephFS does not allow. Use another pool or recreate it to get a non-zero pool id.";
 179       return -EINVAL;
 180     }
 181
 182     string fs_name;
 183     cmd_getval(cmdmap, "fs_name", fs_name);
 184     if (fs_name.empty()) {
 185         // Ensure fs name is not empty so that we can implement
 186         // commmands that refer to FS by name in future.
 187         ss << "Filesystem name may not be empty";
 188         return -EINVAL;
 189     }
 190
 191     if (fsmap.get_filesystem(fs_name)) {
 192       auto fs = fsmap.get_filesystem(fs_name);
 193       if (*(fs->mds_map.get_data_pools().begin()) == data
 194           && fs->mds_map.get_metadata_pool() == metadata) {
 195         // Identical FS created already, this is a no-op
 196         ss << "filesystem '" << fs_name << "' already exists";
 197         return 0;
 198       } else {
 199         ss << "filesystem already exists with name '" << fs_name << "'";
 200         return -EINVAL;
 201       }
 202     }
 203
 204     bool force = false;
 205     cmd_getval(cmdmap, "force", force);
 206
 207     const pool_stat_t *stat = mon->mgrstatmon()->get_pool_stat(metadata);
 208     if (stat) {
 209       int64_t metadata_num_objects = stat->stats.sum.num_objects;
 210       if (!force && metadata_num_objects > 0) {
 211         ss << "pool '" << metadata_name
 212            << "' already contains some objects. Use an empty pool instead.";
 213         return -EINVAL;
 214       }
 215     }
 216
 217     if (fsmap.filesystem_count() > 0
 218         && !fsmap.get_enable_multiple()) {
 219       ss << "Creation of multiple filesystems is disabled.  To enable "
 220             "this experimental feature, use 'ceph fs flag set enable_multiple "
 221             "true'";
 222       return -EINVAL;
 223     }
 224
 225     for (auto& fs : fsmap.get_filesystems()) {
 226       const std::vector<int64_t> &data_pools = fs->mds_map.get_data_pools();
 227
 228       bool sure = false;
 229       cmd_getval(cmdmap,
 230                  "allow_dangerous_metadata_overlay", sure);
 231
 232       if ((std::find(data_pools.begin(), data_pools.end(), data) != data_pools.end()
 233            || fs->mds_map.get_metadata_pool() == metadata)
 234           && !sure) {
 235         ss << "Filesystem '" << fs_name
 236            << "' is already using one of the specified RADOS pools. This should ONLY be done in emergencies and after careful reading of the documentation. Pass --allow-dangerous-metadata-overlay to permit this.";
 237         return -EEXIST;
 238       }
 239     }
 240
 241     int64_t fscid = FS_CLUSTER_ID_NONE;
 242     if (cmd_getval(cmdmap, "fscid", fscid)) {
 243       if (!force) {
 244         ss << "Pass --force to create a file system with a specific ID";
 245         return -EINVAL;
 246       }
 247       if (fsmap.filesystem_exists(fscid)) {
 248         ss << "filesystem already exists with id '" << fscid << "'";
 249         return -EINVAL;
 250       }
 251     }
 252
 253     pg_pool_t const *data_pool = mon->osdmon()->osdmap.get_pg_pool(data);
 254     ceph_assert(data_pool != NULL);  // Checked it existed above
 255     pg_pool_t const *metadata_pool = mon->osdmon()->osdmap.get_pg_pool(metadata);
 256     ceph_assert(metadata_pool != NULL);  // Checked it existed above
 257
 258     int r = _check_pool(mon->osdmon()->osdmap, data, POOL_DATA_DEFAULT, force, &ss);
 259     if (r < 0) {
 260       return r;
 261     }
 262
 263     r = _check_pool(mon->osdmon()->osdmap, metadata, POOL_METADATA, force, &ss);
 264     if (r < 0) {
 265       return r;
 266     }
 267
 268     if (!mon->osdmon()->is_writeable()) {
 269       // not allowed to write yet, so retry when we can
 270       mon->osdmon()->wait_for_writeable(op, new PaxosService::C_RetryMessage(mon->mdsmon(), op));
 271       return -EAGAIN;
 272     }
 273     mon->osdmon()->do_application_enable(data,
 274                                          pg_pool_t::APPLICATION_NAME_CEPHFS,
 275                                          "data", fs_name, true);
 276     mon->osdmon()->do_application_enable(metadata,
 277                                          pg_pool_t::APPLICATION_NAME_CEPHFS,
 278                                          "metadata", fs_name, true);
 279     mon->osdmon()->do_set_pool_opt(metadata,
 280                                    pool_opts_t::RECOVERY_PRIORITY,
 281                                    static_cast<int64_t>(5));
 282     mon->osdmon()->do_set_pool_opt(metadata,
 283                                    pool_opts_t::PG_NUM_MIN,
 284                                    static_cast<int64_t>(16));
 285     mon->osdmon()->do_set_pool_opt(metadata,
 286                                    pool_opts_t::PG_AUTOSCALE_BIAS,
 287                                    static_cast<double>(4.0));
 288     mon->osdmon()->propose_pending();
 289
 290     bool recover = false;
 291     cmd_getval(cmdmap, "recover", recover);
 292
 293     // All checks passed, go ahead and create.
 294     auto&& fs = fsmap.create_filesystem(fs_name, metadata, data,
 295         mon->get_quorum_con_features(), fscid, recover);
 296
 297     ss << "new fs with metadata pool " << metadata << " and data pool " << data;
 298
 299     if (recover) {
 300       return 0;
 301     }
 302
 303     // assign a standby to rank 0 to avoid health warnings
 304     auto info = fsmap.find_replacement_for({fs->fscid, 0});
 305
 306     if (info) {
 307       mon->clog->info() << info->human_name() << " assigned to filesystem "
 308           << fs_name << " as rank 0";
 309       fsmap.promote(info->global_id, *fs, 0);
 310     }
 311
 312     return 0;
 313   }
 314
 315 private:
 316   Paxos *m_paxos;
 317 };
 318
 319 class SetHandler : public FileSystemCommandHandler
 320 {
 321 public:
 322   SetHandler()
 323     : FileSystemCommandHandler("fs set")
 324   {}
 325
 326   int handle(
 327       Monitor *mon,
 328       FSMap& fsmap,
 329       MonOpRequestRef op,
 330       const cmdmap_t& cmdmap,
 331       std::ostream &ss) override
 332   {
 333     std::string fs_name;
 334     if (!cmd_getval(cmdmap, "fs_name", fs_name) || fs_name.empty()) {
 335       ss << "Missing filesystem name";
 336       return -EINVAL;
 337     }
 338
 339     auto fs = fsmap.get_filesystem(fs_name);
 340     string var;
 341     if (!cmd_getval(cmdmap, "var", var) || var.empty()) {
 342       ss << "Invalid variable";
 343       return -EINVAL;
 344     }
 345     string val;
 346     string interr;
 347     int64_t n = 0;
 348     if (!cmd_getval(cmdmap, "val", val)) {
 349       return -EINVAL;
 350     }
 351     // we got a string.  see if it contains an int.
 352     n = strict_strtoll(val.c_str(), 10, &interr);
 353     if (var == "max_mds") {
 354       // NOTE: see also "mds set_max_mds", which can modify the same field.
 355       if (interr.length()) {
 356         ss << interr;
 357         return -EINVAL;
 358       }
 359
 360       if (n <= 0) {
 361         ss << "You must specify at least one MDS";
 362         return -EINVAL;
 363       }
 364
 365       if (n > 1 && n > fs->mds_map.get_max_mds()) {
 366         if (fs->mds_map.was_snaps_ever_allowed() &&
 367             !fs->mds_map.allows_multimds_snaps()) {
 368           ss << "multi-active MDS is not allowed while there are snapshots possibly created by pre-mimic MDS";
 369           return -EINVAL;
 370         }
 371       }
 372       if (n > MAX_MDS) {
 373         ss << "may not have more than " << MAX_MDS << " MDS ranks";
 374         return -EINVAL;
 375       }
 376
 377       fsmap.modify_filesystem(
 378           fs->fscid,
 379           [n](std::shared_ptr<Filesystem> fs)
 380       {
 381         fs->mds_map.clear_flag(CEPH_MDSMAP_NOT_JOINABLE);
 382         fs->mds_map.set_max_mds(n);
 383       });
 384     } else if (var == "inline_data") {
 385       bool enable_inline = false;
 386       int r = parse_bool(val, &enable_inline, ss);
 387       if (r != 0) {
 388         return r;
 389       }
 390
 391       if (enable_inline) {
 392         bool confirm = false;
 393         cmd_getval(cmdmap, "yes_i_really_really_mean_it", confirm);
 394         if (!confirm) {
 395           ss << "Inline data support is deprecated and will be removed in a future release. "
 396              << "Add --yes-i-really-really-mean-it if you are certain you want this enabled.";
 397           return -EPERM;
 398         }
 399         ss << "inline data enabled";
 400
 401         fsmap.modify_filesystem(
 402             fs->fscid,
 403             [](std::shared_ptr<Filesystem> fs)
 404         {
 405           fs->mds_map.set_inline_data_enabled(true);
 406         });
 407       } else {
 408         ss << "inline data disabled";
 409         fsmap.modify_filesystem(
 410             fs->fscid,
 411             [](std::shared_ptr<Filesystem> fs)
 412         {
 413           fs->mds_map.set_inline_data_enabled(false);
 414         });
 415       }
 416     } else if (var == "balancer") {
 417       if (val.empty()) {
 418         ss << "unsetting the metadata load balancer";
 419       } else {
 420         ss << "setting the metadata load balancer to " << val;
 421       }
 422       fsmap.modify_filesystem(
 423         fs->fscid,
 424         [val](std::shared_ptr<Filesystem> fs)
 425         {
 426           fs->mds_map.set_balancer(val);
 427         });
 428       return true;
 429     } else if (var == "max_file_size") {
 430       if (interr.length()) {
 431         ss << var << " requires an integer value";
 432         return -EINVAL;
 433       }
 434       if (n < CEPH_MIN_STRIPE_UNIT) {
 435         ss << var << " must at least " << CEPH_MIN_STRIPE_UNIT;
 436         return -ERANGE;
 437       }
 438       fsmap.modify_filesystem(
 439           fs->fscid,
 440           [n](std::shared_ptr<Filesystem> fs)
 441       {
 442         fs->mds_map.set_max_filesize(n);
 443       });
 444     } else if (var == "allow_new_snaps") {
 445       bool enable_snaps = false;
 446       int r = parse_bool(val, &enable_snaps, ss);
 447       if (r != 0) {
 448         return r;
 449       }
 450
 451       if (!enable_snaps) {
 452         fsmap.modify_filesystem(
 453             fs->fscid,
 454             [](std::shared_ptr<Filesystem> fs)
 455         {
 456           fs->mds_map.clear_snaps_allowed();
 457         });
 458         ss << "disabled new snapshots";
 459       } else {
 460         fsmap.modify_filesystem(
 461             fs->fscid,
 462             [](std::shared_ptr<Filesystem> fs)
 463         {
 464           fs->mds_map.set_snaps_allowed();
 465         });
 466         ss << "enabled new snapshots";
 467       }
 468     } else if (var == "allow_multimds") {
 469         ss << "Multiple MDS is always enabled. Use the max_mds"
 470            << " parameter to control the number of active MDSs"
 471            << " allowed. This command is DEPRECATED and will be"
 472            << " REMOVED from future releases.";
 473     } else if (var == "allow_multimds_snaps") {
 474       bool enable = false;
 475       int r = parse_bool(val, &enable, ss);
 476       if (r != 0) {
 477         return r;
 478       }
 479
 480       string confirm;
 481       if (!cmd_getval(cmdmap, "confirm", confirm) ||
 482           confirm != "--yes-i-am-really-a-mds") {
 483         ss << "Warning! This command is for MDS only. Do not run it manually";
 484         return -EPERM;
 485       }
 486
 487       if (enable) {
 488         ss << "enabled multimds with snapshot";
 489         fsmap.modify_filesystem(
 490             fs->fscid,
 491             [](std::shared_ptr<Filesystem> fs)
 492         {
 493           fs->mds_map.set_multimds_snaps_allowed();
 494         });
 495       } else {
 496         ss << "disabled multimds with snapshot";
 497         fsmap.modify_filesystem(
 498             fs->fscid,
 499             [](std::shared_ptr<Filesystem> fs)
 500         {
 501           fs->mds_map.clear_multimds_snaps_allowed();
 502         });
 503       }
 504     } else if (var == "allow_dirfrags") {
 505         ss << "Directory fragmentation is now permanently enabled."
 506            << " This command is DEPRECATED and will be REMOVED from future releases.";
 507     } else if (var == "down") {
 508       bool is_down = false;
 509       int r = parse_bool(val, &is_down, ss);
 510       if (r != 0) {
 511         return r;
 512       }
 513
 514       ss << fs->mds_map.get_fs_name();
 515
 516       fsmap.modify_filesystem(
 517           fs->fscid,
 518           [is_down](std::shared_ptr<Filesystem> fs)
 519       {
 520         if (is_down) {
 521           if (fs->mds_map.get_max_mds() > 0) {
 522             fs->mds_map.set_old_max_mds();
 523             fs->mds_map.set_max_mds(0);
 524           } /* else already down! */
 525         } else {
 526           mds_rank_t oldmax = fs->mds_map.get_old_max_mds();
 527           fs->mds_map.set_max_mds(oldmax ? oldmax : 1);
 528         }
 529       });
 530
 531       if (is_down) {
 532         ss << " marked down. ";
 533       } else {
 534         ss << " marked up, max_mds = " << fs->mds_map.get_max_mds();
 535       }
 536     } else if (var == "cluster_down" || var == "joinable") {
 537       bool joinable = true;
 538       int r = parse_bool(val, &joinable, ss);
 539       if (r != 0) {
 540         return r;
 541       }
 542       if (var == "cluster_down") {
 543         joinable = !joinable;
 544       }
 545
 546       ss << fs->mds_map.get_fs_name();
 547
 548       fsmap.modify_filesystem(
 549           fs->fscid,
 550           [joinable](std::shared_ptr<Filesystem> fs)
 551       {
 552         if (joinable) {
 553           fs->mds_map.clear_flag(CEPH_MDSMAP_NOT_JOINABLE);
 554         } else {
 555           fs->mds_map.set_flag(CEPH_MDSMAP_NOT_JOINABLE);
 556         }
 557       });
 558
 559       if (joinable) {
 560         ss << " marked joinable; MDS may join as newly active.";
 561       } else {
 562         ss << " marked not joinable; MDS cannot join as newly active.";
 563       }
 564
 565       if (var == "cluster_down") {
 566         ss << " WARNING: cluster_down flag is deprecated and will be"
 567            << " removed in a future version. Please use \"joinable\".";
 568       }
 569     } else if (var == "standby_count_wanted") {
 570       if (interr.length()) {
 571        ss << var << " requires an integer value";
 572        return -EINVAL;
 573       }
 574       if (n < 0) {
 575        ss << var << " must be non-negative";
 576        return -ERANGE;
 577       }
 578       fsmap.modify_filesystem(
 579           fs->fscid,
 580           [n](std::shared_ptr<Filesystem> fs)
 581       {
 582         fs->mds_map.set_standby_count_wanted(n);
 583       });
 584     } else if (var == "session_timeout") {
 585       if (interr.length()) {
 586        ss << var << " requires an integer value";
 587        return -EINVAL;
 588       }
 589       if (n < 30) {
 590        ss << var << " must be at least 30s";
 591        return -ERANGE;
 592       }
 593       fsmap.modify_filesystem(
 594           fs->fscid,
 595           [n](std::shared_ptr<Filesystem> fs)
 596       {
 597         fs->mds_map.set_session_timeout((uint32_t)n);
 598       });
 599     } else if (var == "session_autoclose") {
 600       if (interr.length()) {
 601        ss << var << " requires an integer value";
 602        return -EINVAL;
 603       }
 604       if (n < 30) {
 605        ss << var << " must be at least 30s";
 606        return -ERANGE;
 607       }
 608       fsmap.modify_filesystem(
 609           fs->fscid,
 610           [n](std::shared_ptr<Filesystem> fs)
 611       {
 612         fs->mds_map.set_session_autoclose((uint32_t)n);
 613       });
 614     } else if (var == "allow_standby_replay") {
 615       bool allow = false;
 616       int r = parse_bool(val, &allow, ss);
 617       if (r != 0) {
 618         return r;
 619       }
 620
 621       if (!allow) {
 622         if (!mon->osdmon()->is_writeable()) {
 623           // not allowed to write yet, so retry when we can
 624           mon->osdmon()->wait_for_writeable(op, new PaxosService::C_RetryMessage(mon->mdsmon(), op));
 625           return -EAGAIN;
 626         }
 627         std::vector<mds_gid_t> to_fail;
 628         for (const auto& [gid, info]: fs->mds_map.get_mds_info()) {
 629           if (info.state == MDSMap::STATE_STANDBY_REPLAY) {
 630             to_fail.push_back(gid);
 631           }
 632         }
 633
 634         for (const auto& gid : to_fail) {
 635           mon->mdsmon()->fail_mds_gid(fsmap, gid);
 636         }
 637         if (!to_fail.empty()) {
 638           mon->osdmon()->propose_pending();
 639         }
 640       }
 641
 642       auto f = [allow](auto& fs) {
 643         if (allow) {
 644           fs->mds_map.set_standby_replay_allowed();
 645         } else {
 646           fs->mds_map.clear_standby_replay_allowed();
 647         }
 648       };
 649       fsmap.modify_filesystem(fs->fscid, std::move(f));
 650     } else if (var == "min_compat_client") {
 651       auto vno = ceph_release_from_name(val.c_str());
 652       if (!vno) {
 653         ss << "version " << val << " is not recognized";
 654         return -EINVAL;
 655       }
 656       ss << "WARNING: setting min_compat_client is deprecated"
 657             " and may not do what you want.\n"
 658             "The oldest release to set is octopus.\n"
 659             "Please migrate to `ceph fs required_client_features ...`.";
 660       auto f = [vno](auto&& fs) {
 661         fs->mds_map.set_min_compat_client(vno);
 662       };
 663       fsmap.modify_filesystem(fs->fscid, std::move(f));
 664     } else {
 665       ss << "unknown variable " << var;
 666       return -EINVAL;
 667     }
 668
 669     return 0;
 670   }
 671 };
 672
 673 class CompatSetHandler : public FileSystemCommandHandler
 674 {
 675   public:
 676     CompatSetHandler()
 677       : FileSystemCommandHandler("fs compat")
 678     {
 679     }
 680
 681     int handle(
 682         Monitor *mon,
 683         FSMap &fsmap,
 684         MonOpRequestRef op,
 685         const cmdmap_t& cmdmap,
 686         std::ostream &ss) override
 687     {
 688       static const std::set<std::string> subops = {"rm_incompat", "rm_compat", "add_incompat", "add_compat"};
 689
 690       std::string fs_name;
 691       if (!cmd_getval(cmdmap, "fs_name", fs_name) || fs_name.empty()) {
 692         ss << "Missing filesystem name";
 693         return -EINVAL;
 694       }
 695       auto fs = fsmap.get_filesystem(fs_name);
 696       if (fs == nullptr) {
 697         ss << "Not found: '" << fs_name << "'";
 698         return -ENOENT;
 699       }
 700
 701       string subop;
 702       if (!cmd_getval(cmdmap, "subop", subop) || subops.count(subop) == 0) {
 703         ss << "subop `" << subop << "' not recognized. Must be one of: " << subops;
 704         return -EINVAL;
 705       }
 706
 707       int64_t feature;
 708       if (!cmd_getval(cmdmap, "feature", feature) || feature <= 0) {
 709         ss << "Invalid feature";
 710         return -EINVAL;
 711       }
 712
 713       if (fs->mds_map.get_num_up_mds() > 0) {
 714         ss << "file system must be failed or down; use `ceph fs fail` to bring down";
 715         return -EBUSY;
 716       }
 717
 718       CompatSet cs = fs->mds_map.compat;
 719       if (subop == "rm_compat") {
 720         if (cs.compat.contains(feature)) {
 721           ss << "removed compat feature " << feature;
 722           cs.compat.remove(feature);
 723         } else {
 724           ss << "already removed compat feature " << feature;
 725         }
 726       } else if (subop == "rm_incompat") {
 727         if (cs.incompat.contains(feature)) {
 728           ss << "removed incompat feature " << feature;
 729           cs.incompat.remove(feature);
 730         } else {
 731           ss << "already removed incompat feature " << feature;
 732         }
 733       } else if (subop == "add_compat" || subop == "add_incompat") {
 734         string feature_str;
 735         if (!cmd_getval(cmdmap, "feature_str", feature_str) || feature_str.empty()) {
 736           ss << "adding a feature requires a feature string";
 737           return -EINVAL;
 738         }
 739         auto f = CompatSet::Feature(feature, feature_str);
 740         if (subop == "add_compat") {
 741           if (cs.compat.contains(feature)) {
 742             auto name = cs.compat.get_name(feature);
 743             if (name == feature_str) {
 744               ss << "feature already exists";
 745             } else {
 746               ss << "feature with differing name `" << name << "' exists";
 747               return -EEXIST;
 748             }
 749           } else {
 750             cs.compat.insert(f);
 751             ss << "added compat feature " << f;
 752           }
 753         } else if (subop == "add_incompat") {
 754           if (cs.incompat.contains(feature)) {
 755             auto name = cs.incompat.get_name(feature);
 756             if (name == feature_str) {
 757               ss << "feature already exists";
 758             } else {
 759               ss << "feature with differing name `" << name << "' exists";
 760               return -EEXIST;
 761             }
 762           } else {
 763             cs.incompat.insert(f);
 764             ss << "added incompat feature " << f;
 765           }
 766         } else ceph_assert(0);
 767       } else ceph_assert(0);
 768
 769       auto modifyf = [cs = std::move(cs)](auto&& fs) {
 770         fs->mds_map.compat = cs;
 771       };
 772
 773       fsmap.modify_filesystem(fs->fscid, std::move(modifyf));
 774       return 0;
 775     }
 776 };
 777
 778 class RequiredClientFeaturesHandler : public FileSystemCommandHandler
 779 {
 780   public:
 781     RequiredClientFeaturesHandler()
 782       : FileSystemCommandHandler("fs required_client_features")
 783     {
 784     }
 785
 786     int handle(
 787         Monitor *mon,
 788         FSMap &fsmap,
 789         MonOpRequestRef op,
 790         const cmdmap_t& cmdmap,
 791         std::ostream &ss) override
 792     {
 793       std::string fs_name;
 794       if (!cmd_getval(cmdmap, "fs_name", fs_name) || fs_name.empty()) {
 795         ss << "Missing filesystem name";
 796         return -EINVAL;
 797       }
 798       auto fs = fsmap.get_filesystem(fs_name);
 799       if (fs == nullptr) {
 800         ss << "Not found: '" << fs_name << "'";
 801         return -ENOENT;
 802       }
 803       string subop;
 804       if (!cmd_getval(cmdmap, "subop", subop) ||
 805           (subop != "add" && subop != "rm")) {
 806         ss << "Must either add or rm a feature; " << subop << " is not recognized";
 807         return -EINVAL;
 808       }
 809       string val;
 810       if (!cmd_getval(cmdmap, "val", val) || val.empty()) {
 811         ss << "Missing feature id/name";
 812         return -EINVAL;
 813       }
 814
 815       int feature = cephfs_feature_from_name(val);
 816       if (feature < 0) {
 817         string err;
 818         feature = strict_strtol(val.c_str(), 10, &err);
 819         if (err.length()) {
 820           ss << "Invalid feature name: " << val;
 821           return -EINVAL;
 822         }
 823         if (feature < 0 || feature > CEPHFS_FEATURE_MAX) {
 824           ss << "Invalid feature id: " << feature;
 825           return -EINVAL;
 826         }
 827       }
 828
 829       if (subop == "add") {
 830         bool ret = false;
 831         fsmap.modify_filesystem(
 832             fs->fscid,
 833             [feature, &ret](auto&& fs)
 834         {
 835           if (fs->mds_map.get_required_client_features().test(feature))
 836             return;
 837           fs->mds_map.add_required_client_feature(feature);
 838           ret = true;
 839         });
 840         if (ret) {
 841           ss << "added feature '" << cephfs_feature_name(feature) << "' to required_client_features";
 842         } else {
 843           ss << "feature '" << cephfs_feature_name(feature) << "' is already set";
 844         }
 845       } else {
 846         bool ret = false;
 847         fsmap.modify_filesystem(
 848             fs->fscid,
 849             [feature, &ret](auto&& fs)
 850         {
 851           if (!fs->mds_map.get_required_client_features().test(feature))
 852             return;
 853           fs->mds_map.remove_required_client_feature(feature);
 854           ret = true;
 855         });
 856         if (ret) {
 857           ss << "removed feature '" << cephfs_feature_name(feature) << "' from required_client_features";
 858         } else {
 859           ss << "feature '" << cephfs_feature_name(feature) << "' is already unset";
 860         }
 861       }
 862       return 0;
 863    }
 864 };
 865
 866
 867 class AddDataPoolHandler : public FileSystemCommandHandler
 868 {
 869   public:
 870   explicit AddDataPoolHandler(Paxos *paxos)
 871     : FileSystemCommandHandler("fs add_data_pool"), m_paxos(paxos)
 872   {}
 873
 874   bool batched_propose() override {
 875     return true;
 876   }
 877
 878   int handle(
 879       Monitor *mon,
 880       FSMap& fsmap,
 881       MonOpRequestRef op,
 882       const cmdmap_t& cmdmap,
 883       std::ostream &ss) override
 884   {
 885     ceph_assert(m_paxos->is_plugged());
 886
 887     string poolname;
 888     cmd_getval(cmdmap, "pool", poolname);
 889
 890     std::string fs_name;
 891     if (!cmd_getval(cmdmap, "fs_name", fs_name)
 892         || fs_name.empty()) {
 893       ss << "Missing filesystem name";
 894       return -EINVAL;
 895     }
 896
 897     int64_t poolid = mon->osdmon()->osdmap.lookup_pg_pool_name(poolname);
 898     if (poolid < 0) {
 899       string err;
 900       poolid = strict_strtol(poolname.c_str(), 10, &err);
 901       if (err.length()) {
 902         ss << "pool '" << poolname << "' does not exist";
 903         return -ENOENT;
 904       }
 905     }
 906
 907     int r = _check_pool(mon->osdmon()->osdmap, poolid, POOL_DATA_EXTRA, false, &ss);
 908     if (r != 0) {
 909       return r;
 910     }
 911
 912     auto fs = fsmap.get_filesystem(fs_name);
 913     // no-op when the data_pool already on fs
 914     if (fs->mds_map.is_data_pool(poolid)) {
 915       ss << "data pool " << poolid << " is already on fs " << fs_name;
 916       return 0;
 917     }
 918
 919     if (!mon->osdmon()->is_writeable()) {
 920       // not allowed to write yet, so retry when we can
 921       mon->osdmon()->wait_for_writeable(op, new PaxosService::C_RetryMessage(mon->mdsmon(), op));
 922       return -EAGAIN;
 923     }
 924     mon->osdmon()->do_application_enable(poolid,
 925                                          pg_pool_t::APPLICATION_NAME_CEPHFS,
 926                                          "data", fs_name, true);
 927     mon->osdmon()->propose_pending();
 928
 929     fsmap.modify_filesystem(
 930         fs->fscid,
 931         [poolid](std::shared_ptr<Filesystem> fs)
 932     {
 933       fs->mds_map.add_data_pool(poolid);
 934     });
 935
 936     ss << "added data pool " << poolid << " to fsmap";
 937
 938     return 0;
 939   }
 940
 941 private:
 942   Paxos *m_paxos;
 943 };
 944
 945 class SetDefaultHandler : public FileSystemCommandHandler
 946 {
 947   public:
 948   SetDefaultHandler()
 949     : FileSystemCommandHandler("fs set-default")
 950   {}
 951
 952   int handle(
 953       Monitor *mon,
 954       FSMap& fsmap,
 955       MonOpRequestRef op,
 956       const cmdmap_t& cmdmap,
 957       std::ostream &ss) override
 958   {
 959     std::string fs_name;
 960     cmd_getval(cmdmap, "fs_name", fs_name);
 961     auto fs = fsmap.get_filesystem(fs_name);
 962     if (fs == nullptr) {
 963         ss << "filesystem '" << fs_name << "' does not exist";
 964         return -ENOENT;
 965     }
 966
 967     fsmap.set_legacy_client_fscid(fs->fscid);
 968     return 0;
 969   }
 970 };
 971
 972 class RemoveFilesystemHandler : public FileSystemCommandHandler
 973 {
 974   public:
 975   RemoveFilesystemHandler()
 976     : FileSystemCommandHandler("fs rm")
 977   {}
 978
 979   int handle(
 980       Monitor *mon,
 981       FSMap& fsmap,
 982       MonOpRequestRef op,
 983       const cmdmap_t& cmdmap,
 984       std::ostream &ss) override
 985   {
 986     /* We may need to blocklist ranks. */
 987     if (!mon->osdmon()->is_writeable()) {
 988       // not allowed to write yet, so retry when we can
 989       mon->osdmon()->wait_for_writeable(op, new PaxosService::C_RetryMessage(mon->mdsmon(), op));
 990       return -EAGAIN;
 991     }
 992
 993     // Check caller has correctly named the FS to delete
 994     // (redundant while there is only one FS, but command
 995     //  syntax should apply to multi-FS future)
 996     string fs_name;
 997     cmd_getval(cmdmap, "fs_name", fs_name);
 998     auto fs = fsmap.get_filesystem(fs_name);
 999     if (fs == nullptr) {
1000         // Consider absence success to make deletes idempotent
1001         ss << "filesystem '" << fs_name << "' does not exist";
1002         return 0;
1003     }
1004
1005     // Check that no MDS daemons are active
1006     if (fs->mds_map.get_num_up_mds() > 0) {
1007       ss << "all MDS daemons must be inactive/failed before removing filesystem. See `ceph fs fail`.";
1008       return -EINVAL;
1009     }
1010
1011     // Check for confirmation flag
1012     bool sure = false;
1013     cmd_getval(cmdmap, "yes_i_really_mean_it", sure);
1014     if (!sure) {
1015       ss << "this is a DESTRUCTIVE operation and will make data in your filesystem permanently" \
1016             " inaccessible.  Add --yes-i-really-mean-it if you are sure you wish to continue.";
1017       return -EPERM;
1018     }
1019
1020     if (fsmap.get_legacy_client_fscid() == fs->fscid) {
1021       fsmap.set_legacy_client_fscid(FS_CLUSTER_ID_NONE);
1022     }
1023
1024     std::vector<mds_gid_t> to_fail;
1025     // There may be standby_replay daemons left here
1026     for (const auto &i : fs->mds_map.get_mds_info()) {
1027       ceph_assert(i.second.state == MDSMap::STATE_STANDBY_REPLAY);
1028       to_fail.push_back(i.first);
1029     }
1030
1031     for (const auto &gid : to_fail) {
1032       // Standby replays don't write, so it isn't important to
1033       // wait for an osdmap propose here: ignore return value.
1034       mon->mdsmon()->fail_mds_gid(fsmap, gid);
1035     }
1036     if (!to_fail.empty()) {
1037       mon->osdmon()->propose_pending(); /* maybe new blocklists */
1038     }
1039
1040     fsmap.erase_filesystem(fs->fscid);
1041
1042     return 0;
1043   }
1044 };
1045
1046 class ResetFilesystemHandler : public FileSystemCommandHandler
1047 {
1048   public:
1049   ResetFilesystemHandler()
1050     : FileSystemCommandHandler("fs reset")
1051   {}
1052
1053   int handle(
1054       Monitor *mon,
1055       FSMap& fsmap,
1056       MonOpRequestRef op,
1057       const cmdmap_t& cmdmap,
1058       std::ostream &ss) override
1059   {
1060     string fs_name;
1061     cmd_getval(cmdmap, "fs_name", fs_name);
1062     auto fs = fsmap.get_filesystem(fs_name);
1063     if (fs == nullptr) {
1064         ss << "filesystem '" << fs_name << "' does not exist";
1065         // Unlike fs rm, we consider this case an error
1066         return -ENOENT;
1067     }
1068
1069     // Check that no MDS daemons are active
1070     if (fs->mds_map.get_num_up_mds() > 0) {
1071       ss << "all MDS daemons must be inactive before resetting filesystem: set the cluster_down flag"
1072             " and use `ceph mds fail` to make this so";
1073       return -EINVAL;
1074     }
1075
1076     // Check for confirmation flag
1077     bool sure = false;
1078     cmd_getval(cmdmap, "yes_i_really_mean_it", sure);
1079     if (!sure) {
1080       ss << "this is a potentially destructive operation, only for use by experts in disaster recovery.  "
1081         "Add --yes-i-really-mean-it if you are sure you wish to continue.";
1082       return -EPERM;
1083     }
1084
1085     fsmap.reset_filesystem(fs->fscid);
1086
1087     return 0;
1088   }
1089 };
1090
1091 class RenameFilesystemHandler : public FileSystemCommandHandler
1092 {
1093   public:
1094   explicit RenameFilesystemHandler(Paxos *paxos)
1095     : FileSystemCommandHandler("fs rename"), m_paxos(paxos)
1096   {
1097   }
1098
1099   bool batched_propose() override {
1100     return true;
1101   }
1102
1103   int handle(
1104       Monitor *mon,
1105       FSMap& fsmap,
1106       MonOpRequestRef op,
1107       const cmdmap_t& cmdmap,
1108       std::ostream &ss) override
1109   {
1110     ceph_assert(m_paxos->is_plugged());
1111
1112     string fs_name;
1113     cmd_getval(cmdmap, "fs_name", fs_name);
1114     auto fs = fsmap.get_filesystem(fs_name);
1115
1116     string new_fs_name;
1117     cmd_getval(cmdmap, "new_fs_name", new_fs_name);
1118     auto new_fs = fsmap.get_filesystem(new_fs_name);
1119
1120     if (fs == nullptr) {
1121         if (new_fs) {
1122           // make 'fs rename' idempotent
1123           ss << "File system may already have been renamed. Desired file system '"
1124              << new_fs_name << "' exists.";
1125           return 0;
1126         } else {
1127           ss << "File system '" << fs_name << "' does not exist";
1128           return -ENOENT;
1129         }
1130     }
1131
1132     if (new_fs) {
1133       ss << "Desired file system name '" << new_fs_name << "' already in use";
1134       return -EINVAL;
1135     }
1136
1137     if (fs->mirror_info.mirrored) {
1138       ss << "Mirroring is enabled on file system '"<< fs_name << "'. Disable mirroring on the "
1139         "file system after ensuring it's OK to do so, and then retry to rename.";
1140       return -EPERM;
1141     }
1142
1143     // Check for confirmation flag
1144     bool sure = false;
1145     cmd_getval(cmdmap, "yes_i_really_mean_it", sure);
1146     if (!sure) {
1147       ss << "this is a potentially disruptive operation, clients' cephx credentials need reauthorized "
1148         "to access the file system and its pools with the new name. "
1149         "Add --yes-i-really-mean-it if you are sure you wish to continue.";
1150       return -EPERM;
1151     }
1152
1153     if (!mon->osdmon()->is_writeable()) {
1154       // not allowed to write yet, so retry when we can
1155       mon->osdmon()->wait_for_writeable(op, new PaxosService::C_RetryMessage(mon->mdsmon(), op));
1156       return -EAGAIN;
1157     }
1158     for (const auto p : fs->mds_map.get_data_pools()) {
1159       mon->osdmon()->do_application_enable(p,
1160                                            pg_pool_t::APPLICATION_NAME_CEPHFS,
1161                                            "data", new_fs_name, true);
1162     }
1163
1164     mon->osdmon()->do_application_enable(fs->mds_map.get_metadata_pool(),
1165                                          pg_pool_t::APPLICATION_NAME_CEPHFS,
1166                                          "metadata", new_fs_name, true);
1167     mon->osdmon()->propose_pending();
1168
1169     auto f = [new_fs_name](auto fs) {
1170                     fs->mds_map.set_fs_name(new_fs_name);
1171              };
1172     fsmap.modify_filesystem(fs->fscid, std::move(f));
1173
1174     ss << "File system is renamed. cephx credentials authorized to "
1175           "old file system name need to be reauthorized to new file "
1176           "system name.";
1177
1178     return 0;
1179   }
1180
1181 private:
1182   Paxos *m_paxos;
1183 };
1184
1185 class RemoveDataPoolHandler : public FileSystemCommandHandler
1186 {
1187   public:
1188   RemoveDataPoolHandler()
1189     : FileSystemCommandHandler("fs rm_data_pool")
1190   {}
1191
1192   int handle(
1193       Monitor *mon,
1194       FSMap& fsmap,
1195       MonOpRequestRef op,
1196       const cmdmap_t& cmdmap,
1197       std::ostream &ss) override
1198   {
1199     string poolname;
1200     cmd_getval(cmdmap, "pool", poolname);
1201
1202     std::string fs_name;
1203     if (!cmd_getval(cmdmap, "fs_name", fs_name)
1204         || fs_name.empty()) {
1205       ss << "Missing filesystem name";
1206       return -EINVAL;
1207     }
1208
1209     int64_t poolid = mon->osdmon()->osdmap.lookup_pg_pool_name(poolname);
1210     if (poolid < 0) {
1211       string err;
1212       poolid = strict_strtol(poolname.c_str(), 10, &err);
1213       if (err.length()) {
1214         ss << "pool '" << poolname << "' does not exist";
1215         return -ENOENT;
1216       } else if (poolid < 0) {
1217         ss << "invalid pool id '" << poolid << "'";
1218         return -EINVAL;
1219       }
1220     }
1221
1222     ceph_assert(poolid >= 0);  // Checked by parsing code above
1223
1224     auto fs = fsmap.get_filesystem(fs_name);
1225     if (fs->mds_map.get_first_data_pool() == poolid) {
1226       ss << "cannot remove default data pool";
1227       return -EINVAL;
1228     }
1229
1230     int r = 0;
1231     fsmap.modify_filesystem(fs->fscid,
1232         [&r, poolid](std::shared_ptr<Filesystem> fs)
1233     {
1234       r = fs->mds_map.remove_data_pool(poolid);
1235     });
1236     if (r == -ENOENT) {
1237       // It was already removed, succeed in silence
1238       return 0;
1239     } else if (r == 0) {
1240       // We removed it, succeed
1241       ss << "removed data pool " << poolid << " from fsmap";
1242       return 0;
1243     } else {
1244       // Unexpected error, bubble up
1245       return r;
1246     }
1247   }
1248 };
1249
1250 /**
1251  * For commands with an alternative prefix
1252  */
1253 template<typename T>
1254 class AliasHandler : public T
1255 {
1256   std::string alias_prefix;
1257
1258   public:
1259   explicit AliasHandler(const std::string &new_prefix)
1260     : T()
1261   {
1262     alias_prefix = new_prefix;
1263   }
1264
1265   std::string const &get_prefix() const override {return alias_prefix;}
1266
1267   int handle(
1268       Monitor *mon,
1269       FSMap& fsmap,
1270       MonOpRequestRef op,
1271       const cmdmap_t& cmdmap,
1272       std::ostream &ss) override
1273   {
1274     return T::handle(mon, fsmap, op, cmdmap, ss);
1275   }
1276 };
1277
1278 class MirrorHandlerEnable : public FileSystemCommandHandler
1279 {
1280 public:
1281   MirrorHandlerEnable()
1282     : FileSystemCommandHandler("fs mirror enable")
1283   {}
1284
1285   int handle(Monitor *mon,
1286              FSMap &fsmap, MonOpRequestRef op,
1287              const cmdmap_t& cmdmap, std::ostream &ss) override {
1288     std::string fs_name;
1289     if (!cmd_getval(cmdmap, "fs_name", fs_name) || fs_name.empty()) {
1290       ss << "Missing filesystem name";
1291       return -EINVAL;
1292     }
1293
1294     auto fs = fsmap.get_filesystem(fs_name);
1295     if (fs == nullptr) {
1296       ss << "Filesystem '" << fs_name << "' not found";
1297       return -ENOENT;
1298     }
1299
1300     if (fs->mirror_info.is_mirrored()) {
1301       return 0;
1302     }
1303
1304     auto f = [](auto &&fs) {
1305                fs->mirror_info.enable_mirroring();
1306     };
1307     fsmap.modify_filesystem(fs->fscid, std::move(f));
1308
1309     return 0;
1310   }
1311 };
1312
1313 class MirrorHandlerDisable : public FileSystemCommandHandler
1314 {
1315 public:
1316   MirrorHandlerDisable()
1317     : FileSystemCommandHandler("fs mirror disable")
1318   {}
1319
1320   int handle(Monitor *mon,
1321              FSMap &fsmap, MonOpRequestRef op,
1322              const cmdmap_t& cmdmap, std::ostream &ss) override {
1323     std::string fs_name;
1324     if (!cmd_getval(cmdmap, "fs_name", fs_name) || fs_name.empty()) {
1325       ss << "Missing filesystem name";
1326       return -EINVAL;
1327     }
1328
1329     auto fs = fsmap.get_filesystem(fs_name);
1330     if (fs == nullptr) {
1331       ss << "Filesystem '" << fs_name << "' not found";
1332       return -ENOENT;
1333     }
1334
1335     if (!fs->mirror_info.is_mirrored()) {
1336       return 0;
1337     }
1338
1339     auto f = [](auto &&fs) {
1340       fs->mirror_info.disable_mirroring();
1341     };
1342     fsmap.modify_filesystem(fs->fscid, std::move(f));
1343
1344     return 0;
1345   }
1346 };
1347
1348 class MirrorHandlerAddPeer : public FileSystemCommandHandler
1349 {
1350 public:
1351   MirrorHandlerAddPeer()
1352     : FileSystemCommandHandler("fs mirror peer_add")
1353   {}
1354
1355   boost::optional<std::pair<string, string>>
1356   extract_remote_cluster_conf(const std::string &spec) {
1357     auto pos = spec.find("@");
1358     if (pos == std::string_view::npos) {
1359       return boost::optional<std::pair<string, string>>();
1360     }
1361
1362     auto client = spec.substr(0, pos);
1363     auto cluster = spec.substr(pos+1);
1364
1365     return std::make_pair(client, cluster);
1366   }
1367
1368   bool peer_add(FSMap &fsmap, Filesystem::const_ref &&fs,
1369                 const cmdmap_t &cmdmap, std::ostream &ss) {
1370     string peer_uuid;
1371     string remote_spec;
1372     string remote_fs_name;
1373     cmd_getval(cmdmap, "uuid", peer_uuid);
1374     cmd_getval(cmdmap, "remote_cluster_spec", remote_spec);
1375     cmd_getval(cmdmap, "remote_fs_name", remote_fs_name);
1376
1377     // verify (and extract) remote cluster specification
1378     auto remote_conf = extract_remote_cluster_conf(remote_spec);
1379     if (!remote_conf) {
1380       ss << "invalid remote cluster spec -- should be <client>@<cluster>";
1381       return false;
1382     }
1383
1384     if (fs->mirror_info.has_peer(peer_uuid)) {
1385       ss << "peer already exists";
1386       return true;
1387     }
1388     if (fs->mirror_info.has_peer((*remote_conf).first, (*remote_conf).second,
1389                                  remote_fs_name)) {
1390       ss << "peer already exists";
1391       return true;
1392     }
1393
1394     auto f = [peer_uuid, remote_conf, remote_fs_name](auto &&fs) {
1395                fs->mirror_info.peer_add(peer_uuid, (*remote_conf).first,
1396                                         (*remote_conf).second, remote_fs_name);
1397              };
1398     fsmap.modify_filesystem(fs->fscid, std::move(f));
1399     return true;
1400   }
1401
1402   int handle(Monitor *mon,
1403              FSMap &fsmap, MonOpRequestRef op,
1404              const cmdmap_t& cmdmap, std::ostream &ss) override {
1405     std::string fs_name;
1406     if (!cmd_getval(cmdmap, "fs_name", fs_name) || fs_name.empty()) {
1407       ss << "Missing filesystem name";
1408       return -EINVAL;
1409     }
1410
1411     auto fs = fsmap.get_filesystem(fs_name);
1412     if (fs == nullptr) {
1413       ss << "Filesystem '" << fs_name << "' not found";
1414       return -ENOENT;
1415     }
1416
1417     if (!fs->mirror_info.is_mirrored()) {
1418       ss << "Mirroring not enabled for filesystem '" << fs_name << "'";
1419       return -EINVAL;
1420     }
1421
1422     auto res = peer_add(fsmap, std::move(fs), cmdmap, ss);
1423     if (!res) {
1424       return -EINVAL;
1425     }
1426
1427     return 0;
1428   }
1429 };
1430
1431 class MirrorHandlerRemovePeer : public FileSystemCommandHandler
1432 {
1433 public:
1434   MirrorHandlerRemovePeer()
1435     : FileSystemCommandHandler("fs mirror peer_remove")
1436   {}
1437
1438   bool peer_remove(FSMap &fsmap, Filesystem::const_ref &&fs,
1439                    const cmdmap_t &cmdmap, std::ostream &ss) {
1440     string peer_uuid;
1441     cmd_getval(cmdmap, "uuid", peer_uuid);
1442
1443     if (!fs->mirror_info.has_peer(peer_uuid)) {
1444       ss << "cannot find peer with uuid: " << peer_uuid;
1445       return true;
1446     }
1447
1448     auto f = [peer_uuid](auto &&fs) {
1449                fs->mirror_info.peer_remove(peer_uuid);
1450              };
1451     fsmap.modify_filesystem(fs->fscid, std::move(f));
1452     return true;
1453   }
1454
1455   int handle(Monitor *mon,
1456              FSMap &fsmap, MonOpRequestRef op,
1457              const cmdmap_t& cmdmap, std::ostream &ss) override {
1458     std::string fs_name;
1459     if (!cmd_getval(cmdmap, "fs_name", fs_name) || fs_name.empty()) {
1460       ss << "Missing filesystem name";
1461       return -EINVAL;
1462     }
1463
1464     auto fs = fsmap.get_filesystem(fs_name);
1465     if (fs == nullptr) {
1466       ss << "Filesystem '" << fs_name << "' not found";
1467       return -ENOENT;
1468     }
1469
1470     if (!fs->mirror_info.is_mirrored()) {
1471       ss << "Mirroring not enabled for filesystem '" << fs_name << "'";
1472       return -EINVAL;
1473     }
1474
1475     auto res = peer_remove(fsmap, std::move(fs), cmdmap, ss);
1476     if (!res) {
1477       return -EINVAL;
1478     }
1479
1480     return 0;
1481   }
1482 };
1483
1484 std::list<std::shared_ptr<FileSystemCommandHandler> >
1485 FileSystemCommandHandler::load(Paxos *paxos)
1486 {
1487   std::list<std::shared_ptr<FileSystemCommandHandler> > handlers;
1488
1489   handlers.push_back(std::make_shared<SetHandler>());
1490   handlers.push_back(std::make_shared<FailHandler>());
1491   handlers.push_back(std::make_shared<FlagSetHandler>());
1492   handlers.push_back(std::make_shared<CompatSetHandler>());
1493   handlers.push_back(std::make_shared<RequiredClientFeaturesHandler>());
1494   handlers.push_back(std::make_shared<AddDataPoolHandler>(paxos));
1495   handlers.push_back(std::make_shared<RemoveDataPoolHandler>());
1496   handlers.push_back(std::make_shared<FsNewHandler>(paxos));
1497   handlers.push_back(std::make_shared<RemoveFilesystemHandler>());
1498   handlers.push_back(std::make_shared<ResetFilesystemHandler>());
1499   handlers.push_back(std::make_shared<RenameFilesystemHandler>(paxos));
1500
1501   handlers.push_back(std::make_shared<SetDefaultHandler>());
1502   handlers.push_back(std::make_shared<AliasHandler<SetDefaultHandler> >(
1503         "fs set_default"));
1504   handlers.push_back(std::make_shared<MirrorHandlerEnable>());
1505   handlers.push_back(std::make_shared<MirrorHandlerDisable>());
1506   handlers.push_back(std::make_shared<MirrorHandlerAddPeer>());
1507   handlers.push_back(std::make_shared<MirrorHandlerRemovePeer>());
1508
1509   return handlers;
1510 }
1511
1512 int FileSystemCommandHandler::_check_pool(
1513     OSDMap &osd_map,
1514     const int64_t pool_id,
1515     int type,
1516     bool force,
1517     std::ostream *ss) const
1518 {
1519   ceph_assert(ss != NULL);
1520
1521   const pg_pool_t *pool = osd_map.get_pg_pool(pool_id);
1522   if (!pool) {
1523     *ss << "pool id '" << pool_id << "' does not exist";
1524     return -ENOENT;
1525   }
1526
1527   const string& pool_name = osd_map.get_pool_name(pool_id);
1528
1529   if (pool->is_erasure()) {
1530     if (type == POOL_METADATA) {
1531       *ss << "pool '" << pool_name << "' (id '" << pool_id << "')"
1532          << " is an erasure-coded pool.  Use of erasure-coded pools"
1533          << " for CephFS metadata is not permitted";
1534       return -EINVAL;
1535     } else if (type == POOL_DATA_DEFAULT && !force) {
1536       *ss << "pool '" << pool_name << "' (id '" << pool_id << "')"
1537              " is an erasure-coded pool."
1538              " Use of an EC pool for the default data pool is discouraged;"
1539              " see the online CephFS documentation for more information."
1540              " Use --force to override.";
1541       return -EINVAL;
1542     } else if (!pool->allows_ecoverwrites()) {
1543       // non-overwriteable EC pools are only acceptable with a cache tier overlay
1544       if (!pool->has_tiers() || !pool->has_read_tier() || !pool->has_write_tier()) {
1545         *ss << "pool '" << pool_name << "' (id '" << pool_id << "')"
1546             << " is an erasure-coded pool, with no overwrite support";
1547         return -EINVAL;
1548       }
1549
1550       // That cache tier overlay must be writeback, not readonly (it's the
1551       // write operations like modify+truncate we care about support for)
1552       const pg_pool_t *write_tier = osd_map.get_pg_pool(
1553           pool->write_tier);
1554       ceph_assert(write_tier != NULL);  // OSDMonitor shouldn't allow DNE tier
1555       if (write_tier->cache_mode == pg_pool_t::CACHEMODE_FORWARD
1556           || write_tier->cache_mode == pg_pool_t::CACHEMODE_READONLY) {
1557         *ss << "EC pool '" << pool_name << "' has a write tier ("
1558             << osd_map.get_pool_name(pool->write_tier)
1559             << ") that is configured "
1560                "to forward writes.  Use a cache mode such as 'writeback' for "
1561                "CephFS";
1562         return -EINVAL;
1563       }
1564     }
1565   }
1566
1567   if (pool->is_tier()) {
1568     *ss << " pool '" << pool_name << "' (id '" << pool_id
1569       << "') is already in use as a cache tier.";
1570     return -EINVAL;
1571   }
1572
1573   if (!force && !pool->application_metadata.empty() &&
1574       pool->application_metadata.count(
1575         pg_pool_t::APPLICATION_NAME_CEPHFS) == 0) {
1576     *ss << " pool '" << pool_name << "' (id '" << pool_id
1577         << "') has a non-CephFS application enabled.";
1578     return -EINVAL;
1579   }
1580
1581   // Nothing special about this pool, so it is permissible
1582   return 0;
1583 }
1584
1585 int FileSystemCommandHandler::is_op_allowed(
1586     const MonOpRequestRef& op, const FSMap& fsmap, const cmdmap_t& cmdmap,
1587     std::ostream &ss) const
1588 {
1589     string fs_name;
1590     cmd_getval(cmdmap, "fs_name", fs_name);
1591
1592     // so that fsmap can filtered and the original copy is untouched.
1593     FSMap fsmap_copy = fsmap;
1594     fsmap_copy.filter(op->get_session()->get_allowed_fs_names());
1595
1596     auto fs = fsmap_copy.get_filesystem(fs_name);
1597     if (fs == nullptr) {
1598       auto prefix = get_prefix();
1599       /* let "fs rm" and "fs rename" handle idempotent cases where file systems do not exist */
1600       if (!(prefix == "fs rm" || prefix == "fs rename") && fsmap.get_filesystem(fs_name) == nullptr) {
1601         ss << "Filesystem not found: '" << fs_name << "'";
1602         return -ENOENT;
1603       }
1604     }
1605
1606     if (!op->get_session()->fs_name_capable(fs_name, MON_CAP_W)) {
1607       ss << "Permission denied: '" << fs_name << "'";
1608       return -EPERM;
1609     }
1610
1611   return 1;
1612 }