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     // All checks passed, go ahead and create.
 291     auto&& fs = fsmap.create_filesystem(fs_name, metadata, data,
 292         mon->get_quorum_con_features(), fscid);
 293
 294     ss << "new fs with metadata pool " << metadata << " and data pool " << data;
 295
 296     // assign a standby to rank 0 to avoid health warnings
 297     auto info = fsmap.find_replacement_for({fs->fscid, 0});
 298
 299     if (info) {
 300       mon->clog->info() << info->human_name() << " assigned to filesystem "
 301           << fs_name << " as rank 0";
 302       fsmap.promote(info->global_id, *fs, 0);
 303     }
 304
 305     return 0;
 306   }
 307
 308 private:
 309   Paxos *m_paxos;
 310 };
 311
 312 class SetHandler : public FileSystemCommandHandler
 313 {
 314 public:
 315   SetHandler()
 316     : FileSystemCommandHandler("fs set")
 317   {}
 318
 319   int handle(
 320       Monitor *mon,
 321       FSMap& fsmap,
 322       MonOpRequestRef op,
 323       const cmdmap_t& cmdmap,
 324       std::ostream &ss) override
 325   {
 326     std::string fs_name;
 327     if (!cmd_getval(cmdmap, "fs_name", fs_name) || fs_name.empty()) {
 328       ss << "Missing filesystem name";
 329       return -EINVAL;
 330     }
 331
 332     auto fs = fsmap.get_filesystem(fs_name);
 333     string var;
 334     if (!cmd_getval(cmdmap, "var", var) || var.empty()) {
 335       ss << "Invalid variable";
 336       return -EINVAL;
 337     }
 338     string val;
 339     string interr;
 340     int64_t n = 0;
 341     if (!cmd_getval(cmdmap, "val", val)) {
 342       return -EINVAL;
 343     }
 344     // we got a string.  see if it contains an int.
 345     n = strict_strtoll(val.c_str(), 10, &interr);
 346     if (var == "max_mds") {
 347       // NOTE: see also "mds set_max_mds", which can modify the same field.
 348       if (interr.length()) {
 349         ss << interr;
 350         return -EINVAL;
 351       }
 352
 353       if (n <= 0) {
 354         ss << "You must specify at least one MDS";
 355         return -EINVAL;
 356       }
 357
 358       if (n > 1 && n > fs->mds_map.get_max_mds()) {
 359         if (fs->mds_map.was_snaps_ever_allowed() &&
 360             !fs->mds_map.allows_multimds_snaps()) {
 361           ss << "multi-active MDS is not allowed while there are snapshots possibly created by pre-mimic MDS";
 362           return -EINVAL;
 363         }
 364       }
 365       if (n > MAX_MDS) {
 366         ss << "may not have more than " << MAX_MDS << " MDS ranks";
 367         return -EINVAL;
 368       }
 369
 370       fsmap.modify_filesystem(
 371           fs->fscid,
 372           [n](std::shared_ptr<Filesystem> fs)
 373       {
 374         fs->mds_map.clear_flag(CEPH_MDSMAP_NOT_JOINABLE);
 375         fs->mds_map.set_max_mds(n);
 376       });
 377     } else if (var == "inline_data") {
 378       bool enable_inline = false;
 379       int r = parse_bool(val, &enable_inline, ss);
 380       if (r != 0) {
 381         return r;
 382       }
 383
 384       if (enable_inline) {
 385         bool confirm = false;
 386         cmd_getval(cmdmap, "yes_i_really_really_mean_it", confirm);
 387         if (!confirm) {
 388           ss << "Inline data support is deprecated and will be removed in a future release. "
 389              << "Add --yes-i-really-really-mean-it if you are certain you want this enabled.";
 390           return -EPERM;
 391         }
 392         ss << "inline data enabled";
 393
 394         fsmap.modify_filesystem(
 395             fs->fscid,
 396             [](std::shared_ptr<Filesystem> fs)
 397         {
 398           fs->mds_map.set_inline_data_enabled(true);
 399         });
 400       } else {
 401         ss << "inline data disabled";
 402         fsmap.modify_filesystem(
 403             fs->fscid,
 404             [](std::shared_ptr<Filesystem> fs)
 405         {
 406           fs->mds_map.set_inline_data_enabled(false);
 407         });
 408       }
 409     } else if (var == "balancer") {
 410       if (val.empty()) {
 411         ss << "unsetting the metadata load balancer";
 412       } else {
 413         ss << "setting the metadata load balancer to " << val;
 414       }
 415       fsmap.modify_filesystem(
 416         fs->fscid,
 417         [val](std::shared_ptr<Filesystem> fs)
 418         {
 419           fs->mds_map.set_balancer(val);
 420         });
 421       return true;
 422     } else if (var == "max_file_size") {
 423       if (interr.length()) {
 424         ss << var << " requires an integer value";
 425         return -EINVAL;
 426       }
 427       if (n < CEPH_MIN_STRIPE_UNIT) {
 428         ss << var << " must at least " << CEPH_MIN_STRIPE_UNIT;
 429         return -ERANGE;
 430       }
 431       fsmap.modify_filesystem(
 432           fs->fscid,
 433           [n](std::shared_ptr<Filesystem> fs)
 434       {
 435         fs->mds_map.set_max_filesize(n);
 436       });
 437     } else if (var == "allow_new_snaps") {
 438       bool enable_snaps = false;
 439       int r = parse_bool(val, &enable_snaps, ss);
 440       if (r != 0) {
 441         return r;
 442       }
 443
 444       if (!enable_snaps) {
 445         fsmap.modify_filesystem(
 446             fs->fscid,
 447             [](std::shared_ptr<Filesystem> fs)
 448         {
 449           fs->mds_map.clear_snaps_allowed();
 450         });
 451         ss << "disabled new snapshots";
 452       } else {
 453         fsmap.modify_filesystem(
 454             fs->fscid,
 455             [](std::shared_ptr<Filesystem> fs)
 456         {
 457           fs->mds_map.set_snaps_allowed();
 458         });
 459         ss << "enabled new snapshots";
 460       }
 461     } else if (var == "allow_multimds") {
 462         ss << "Multiple MDS is always enabled. Use the max_mds"
 463            << " parameter to control the number of active MDSs"
 464            << " allowed. This command is DEPRECATED and will be"
 465            << " REMOVED from future releases.";
 466     } else if (var == "allow_multimds_snaps") {
 467       bool enable = false;
 468       int r = parse_bool(val, &enable, ss);
 469       if (r != 0) {
 470         return r;
 471       }
 472
 473       string confirm;
 474       if (!cmd_getval(cmdmap, "confirm", confirm) ||
 475           confirm != "--yes-i-am-really-a-mds") {
 476         ss << "Warning! This command is for MDS only. Do not run it manually";
 477         return -EPERM;
 478       }
 479
 480       if (enable) {
 481         ss << "enabled multimds with snapshot";
 482         fsmap.modify_filesystem(
 483             fs->fscid,
 484             [](std::shared_ptr<Filesystem> fs)
 485         {
 486           fs->mds_map.set_multimds_snaps_allowed();
 487         });
 488       } else {
 489         ss << "disabled multimds with snapshot";
 490         fsmap.modify_filesystem(
 491             fs->fscid,
 492             [](std::shared_ptr<Filesystem> fs)
 493         {
 494           fs->mds_map.clear_multimds_snaps_allowed();
 495         });
 496       }
 497     } else if (var == "allow_dirfrags") {
 498         ss << "Directory fragmentation is now permanently enabled."
 499            << " This command is DEPRECATED and will be REMOVED from future releases.";
 500     } else if (var == "down") {
 501       bool is_down = false;
 502       int r = parse_bool(val, &is_down, ss);
 503       if (r != 0) {
 504         return r;
 505       }
 506
 507       ss << fs->mds_map.get_fs_name();
 508
 509       fsmap.modify_filesystem(
 510           fs->fscid,
 511           [is_down](std::shared_ptr<Filesystem> fs)
 512       {
 513         if (is_down) {
 514           if (fs->mds_map.get_max_mds() > 0) {
 515             fs->mds_map.set_old_max_mds();
 516             fs->mds_map.set_max_mds(0);
 517           } /* else already down! */
 518         } else {
 519           mds_rank_t oldmax = fs->mds_map.get_old_max_mds();
 520           fs->mds_map.set_max_mds(oldmax ? oldmax : 1);
 521         }
 522       });
 523
 524       if (is_down) {
 525         ss << " marked down. ";
 526       } else {
 527         ss << " marked up, max_mds = " << fs->mds_map.get_max_mds();
 528       }
 529     } else if (var == "cluster_down" || var == "joinable") {
 530       bool joinable = true;
 531       int r = parse_bool(val, &joinable, ss);
 532       if (r != 0) {
 533         return r;
 534       }
 535       if (var == "cluster_down") {
 536         joinable = !joinable;
 537       }
 538
 539       ss << fs->mds_map.get_fs_name();
 540
 541       fsmap.modify_filesystem(
 542           fs->fscid,
 543           [joinable](std::shared_ptr<Filesystem> fs)
 544       {
 545         if (joinable) {
 546           fs->mds_map.clear_flag(CEPH_MDSMAP_NOT_JOINABLE);
 547         } else {
 548           fs->mds_map.set_flag(CEPH_MDSMAP_NOT_JOINABLE);
 549         }
 550       });
 551
 552       if (joinable) {
 553         ss << " marked joinable; MDS may join as newly active.";
 554       } else {
 555         ss << " marked not joinable; MDS cannot join as newly active.";
 556       }
 557
 558       if (var == "cluster_down") {
 559         ss << " WARNING: cluster_down flag is deprecated and will be"
 560            << " removed in a future version. Please use \"joinable\".";
 561       }
 562     } else if (var == "standby_count_wanted") {
 563       if (interr.length()) {
 564        ss << var << " requires an integer value";
 565        return -EINVAL;
 566       }
 567       if (n < 0) {
 568        ss << var << " must be non-negative";
 569        return -ERANGE;
 570       }
 571       fsmap.modify_filesystem(
 572           fs->fscid,
 573           [n](std::shared_ptr<Filesystem> fs)
 574       {
 575         fs->mds_map.set_standby_count_wanted(n);
 576       });
 577     } else if (var == "session_timeout") {
 578       if (interr.length()) {
 579        ss << var << " requires an integer value";
 580        return -EINVAL;
 581       }
 582       if (n < 30) {
 583        ss << var << " must be at least 30s";
 584        return -ERANGE;
 585       }
 586       fsmap.modify_filesystem(
 587           fs->fscid,
 588           [n](std::shared_ptr<Filesystem> fs)
 589       {
 590         fs->mds_map.set_session_timeout((uint32_t)n);
 591       });
 592     } else if (var == "session_autoclose") {
 593       if (interr.length()) {
 594        ss << var << " requires an integer value";
 595        return -EINVAL;
 596       }
 597       if (n < 30) {
 598        ss << var << " must be at least 30s";
 599        return -ERANGE;
 600       }
 601       fsmap.modify_filesystem(
 602           fs->fscid,
 603           [n](std::shared_ptr<Filesystem> fs)
 604       {
 605         fs->mds_map.set_session_autoclose((uint32_t)n);
 606       });
 607     } else if (var == "allow_standby_replay") {
 608       bool allow = false;
 609       int r = parse_bool(val, &allow, ss);
 610       if (r != 0) {
 611         return r;
 612       }
 613
 614       if (!allow) {
 615         if (!mon->osdmon()->is_writeable()) {
 616           // not allowed to write yet, so retry when we can
 617           mon->osdmon()->wait_for_writeable(op, new PaxosService::C_RetryMessage(mon->mdsmon(), op));
 618           return -EAGAIN;
 619         }
 620         std::vector<mds_gid_t> to_fail;
 621         for (const auto& [gid, info]: fs->mds_map.get_mds_info()) {
 622           if (info.state == MDSMap::STATE_STANDBY_REPLAY) {
 623             to_fail.push_back(gid);
 624           }
 625         }
 626
 627         for (const auto& gid : to_fail) {
 628           mon->mdsmon()->fail_mds_gid(fsmap, gid);
 629         }
 630         if (!to_fail.empty()) {
 631           mon->osdmon()->propose_pending();
 632         }
 633       }
 634
 635       auto f = [allow](auto& fs) {
 636         if (allow) {
 637           fs->mds_map.set_standby_replay_allowed();
 638         } else {
 639           fs->mds_map.clear_standby_replay_allowed();
 640         }
 641       };
 642       fsmap.modify_filesystem(fs->fscid, std::move(f));
 643     } else if (var == "min_compat_client") {
 644       auto vno = ceph_release_from_name(val.c_str());
 645       if (!vno) {
 646         ss << "version " << val << " is not recognized";
 647         return -EINVAL;
 648       }
 649       ss << "WARNING: setting min_compat_client is deprecated"
 650             " and may not do what you want.\n"
 651             "The oldest release to set is octopus.\n"
 652             "Please migrate to `ceph fs required_client_features ...`.";
 653       auto f = [vno](auto&& fs) {
 654         fs->mds_map.set_min_compat_client(vno);
 655       };
 656       fsmap.modify_filesystem(fs->fscid, std::move(f));
 657     } else {
 658       ss << "unknown variable " << var;
 659       return -EINVAL;
 660     }
 661
 662     return 0;
 663   }
 664 };
 665
 666 class CompatSetHandler : public FileSystemCommandHandler
 667 {
 668   public:
 669     CompatSetHandler()
 670       : FileSystemCommandHandler("fs compat")
 671     {
 672     }
 673
 674     int handle(
 675         Monitor *mon,
 676         FSMap &fsmap,
 677         MonOpRequestRef op,
 678         const cmdmap_t& cmdmap,
 679         std::ostream &ss) override
 680     {
 681       static const std::set<std::string> subops = {"rm_incompat", "rm_compat", "add_incompat", "add_compat"};
 682
 683       std::string fs_name;
 684       if (!cmd_getval(cmdmap, "fs_name", fs_name) || fs_name.empty()) {
 685         ss << "Missing filesystem name";
 686         return -EINVAL;
 687       }
 688       auto fs = fsmap.get_filesystem(fs_name);
 689       if (fs == nullptr) {
 690         ss << "Not found: '" << fs_name << "'";
 691         return -ENOENT;
 692       }
 693
 694       string subop;
 695       if (!cmd_getval(cmdmap, "subop", subop) || subops.count(subop) == 0) {
 696         ss << "subop `" << subop << "' not recognized. Must be one of: " << subops;
 697         return -EINVAL;
 698       }
 699
 700       int64_t feature;
 701       if (!cmd_getval(cmdmap, "feature", feature) || feature <= 0) {
 702         ss << "Invalid feature";
 703         return -EINVAL;
 704       }
 705
 706       if (fs->mds_map.get_num_up_mds() > 0) {
 707         ss << "file system must be failed or down; use `ceph fs fail` to bring down";
 708         return -EBUSY;
 709       }
 710
 711       CompatSet cs = fs->mds_map.compat;
 712       if (subop == "rm_compat") {
 713         if (cs.compat.contains(feature)) {
 714           ss << "removed compat feature " << feature;
 715           cs.compat.remove(feature);
 716         } else {
 717           ss << "already removed compat feature " << feature;
 718         }
 719       } else if (subop == "rm_incompat") {
 720         if (cs.incompat.contains(feature)) {
 721           ss << "removed incompat feature " << feature;
 722           cs.incompat.remove(feature);
 723         } else {
 724           ss << "already removed incompat feature " << feature;
 725         }
 726       } else if (subop == "add_compat" || subop == "add_incompat") {
 727         string feature_str;
 728         if (!cmd_getval(cmdmap, "feature_str", feature_str) || feature_str.empty()) {
 729           ss << "adding a feature requires a feature string";
 730           return -EINVAL;
 731         }
 732         auto f = CompatSet::Feature(feature, feature_str);
 733         if (subop == "add_compat") {
 734           if (cs.compat.contains(feature)) {
 735             auto name = cs.compat.get_name(feature);
 736             if (name == feature_str) {
 737               ss << "feature already exists";
 738             } else {
 739               ss << "feature with differing name `" << name << "' exists";
 740               return -EEXIST;
 741             }
 742           } else {
 743             cs.compat.insert(f);
 744             ss << "added compat feature " << f;
 745           }
 746         } else if (subop == "add_incompat") {
 747           if (cs.incompat.contains(feature)) {
 748             auto name = cs.incompat.get_name(feature);
 749             if (name == feature_str) {
 750               ss << "feature already exists";
 751             } else {
 752               ss << "feature with differing name `" << name << "' exists";
 753               return -EEXIST;
 754             }
 755           } else {
 756             cs.incompat.insert(f);
 757             ss << "added incompat feature " << f;
 758           }
 759         } else ceph_assert(0);
 760       } else ceph_assert(0);
 761
 762       auto modifyf = [cs = std::move(cs)](auto&& fs) {
 763         fs->mds_map.compat = cs;
 764       };
 765
 766       fsmap.modify_filesystem(fs->fscid, std::move(modifyf));
 767       return 0;
 768     }
 769 };
 770
 771 class RequiredClientFeaturesHandler : public FileSystemCommandHandler
 772 {
 773   public:
 774     RequiredClientFeaturesHandler()
 775       : FileSystemCommandHandler("fs required_client_features")
 776     {
 777     }
 778
 779     int handle(
 780         Monitor *mon,
 781         FSMap &fsmap,
 782         MonOpRequestRef op,
 783         const cmdmap_t& cmdmap,
 784         std::ostream &ss) override
 785     {
 786       std::string fs_name;
 787       if (!cmd_getval(cmdmap, "fs_name", fs_name) || fs_name.empty()) {
 788         ss << "Missing filesystem name";
 789         return -EINVAL;
 790       }
 791       auto fs = fsmap.get_filesystem(fs_name);
 792       if (fs == nullptr) {
 793         ss << "Not found: '" << fs_name << "'";
 794         return -ENOENT;
 795       }
 796       string subop;
 797       if (!cmd_getval(cmdmap, "subop", subop) ||
 798           (subop != "add" && subop != "rm")) {
 799         ss << "Must either add or rm a feature; " << subop << " is not recognized";
 800         return -EINVAL;
 801       }
 802       string val;
 803       if (!cmd_getval(cmdmap, "val", val) || val.empty()) {
 804         ss << "Missing feature id/name";
 805         return -EINVAL;
 806       }
 807
 808       int feature = cephfs_feature_from_name(val);
 809       if (feature < 0) {
 810         string err;
 811         feature = strict_strtol(val.c_str(), 10, &err);
 812         if (err.length()) {
 813           ss << "Invalid feature name: " << val;
 814           return -EINVAL;
 815         }
 816         if (feature < 0 || feature > CEPHFS_FEATURE_MAX) {
 817           ss << "Invalid feature id: " << feature;
 818           return -EINVAL;
 819         }
 820       }
 821
 822       if (subop == "add") {
 823         bool ret = false;
 824         fsmap.modify_filesystem(
 825             fs->fscid,
 826             [feature, &ret](auto&& fs)
 827         {
 828           if (fs->mds_map.get_required_client_features().test(feature))
 829             return;
 830           fs->mds_map.add_required_client_feature(feature);
 831           ret = true;
 832         });
 833         if (ret) {
 834           ss << "added feature '" << cephfs_feature_name(feature) << "' to required_client_features";
 835         } else {
 836           ss << "feature '" << cephfs_feature_name(feature) << "' is already set";
 837         }
 838       } else {
 839         bool ret = false;
 840         fsmap.modify_filesystem(
 841             fs->fscid,
 842             [feature, &ret](auto&& fs)
 843         {
 844           if (!fs->mds_map.get_required_client_features().test(feature))
 845             return;
 846           fs->mds_map.remove_required_client_feature(feature);
 847           ret = true;
 848         });
 849         if (ret) {
 850           ss << "removed feature '" << cephfs_feature_name(feature) << "' from required_client_features";
 851         } else {
 852           ss << "feature '" << cephfs_feature_name(feature) << "' is already unset";
 853         }
 854       }
 855       return 0;
 856    }
 857 };
 858
 859
 860 class AddDataPoolHandler : public FileSystemCommandHandler
 861 {
 862   public:
 863   explicit AddDataPoolHandler(Paxos *paxos)
 864     : FileSystemCommandHandler("fs add_data_pool"), m_paxos(paxos)
 865   {}
 866
 867   bool batched_propose() override {
 868     return true;
 869   }
 870
 871   int handle(
 872       Monitor *mon,
 873       FSMap& fsmap,
 874       MonOpRequestRef op,
 875       const cmdmap_t& cmdmap,
 876       std::ostream &ss) override
 877   {
 878     ceph_assert(m_paxos->is_plugged());
 879
 880     string poolname;
 881     cmd_getval(cmdmap, "pool", poolname);
 882
 883     std::string fs_name;
 884     if (!cmd_getval(cmdmap, "fs_name", fs_name)
 885         || fs_name.empty()) {
 886       ss << "Missing filesystem name";
 887       return -EINVAL;
 888     }
 889
 890     int64_t poolid = mon->osdmon()->osdmap.lookup_pg_pool_name(poolname);
 891     if (poolid < 0) {
 892       string err;
 893       poolid = strict_strtol(poolname.c_str(), 10, &err);
 894       if (err.length()) {
 895         ss << "pool '" << poolname << "' does not exist";
 896         return -ENOENT;
 897       }
 898     }
 899
 900     int r = _check_pool(mon->osdmon()->osdmap, poolid, POOL_DATA_EXTRA, false, &ss);
 901     if (r != 0) {
 902       return r;
 903     }
 904
 905     auto fs = fsmap.get_filesystem(fs_name);
 906     // no-op when the data_pool already on fs
 907     if (fs->mds_map.is_data_pool(poolid)) {
 908       ss << "data pool " << poolid << " is already on fs " << fs_name;
 909       return 0;
 910     }
 911
 912     if (!mon->osdmon()->is_writeable()) {
 913       // not allowed to write yet, so retry when we can
 914       mon->osdmon()->wait_for_writeable(op, new PaxosService::C_RetryMessage(mon->mdsmon(), op));
 915       return -EAGAIN;
 916     }
 917     mon->osdmon()->do_application_enable(poolid,
 918                                          pg_pool_t::APPLICATION_NAME_CEPHFS,
 919                                          "data", fs_name, true);
 920     mon->osdmon()->propose_pending();
 921
 922     fsmap.modify_filesystem(
 923         fs->fscid,
 924         [poolid](std::shared_ptr<Filesystem> fs)
 925     {
 926       fs->mds_map.add_data_pool(poolid);
 927     });
 928
 929     ss << "added data pool " << poolid << " to fsmap";
 930
 931     return 0;
 932   }
 933
 934 private:
 935   Paxos *m_paxos;
 936 };
 937
 938 class SetDefaultHandler : public FileSystemCommandHandler
 939 {
 940   public:
 941   SetDefaultHandler()
 942     : FileSystemCommandHandler("fs set-default")
 943   {}
 944
 945   int handle(
 946       Monitor *mon,
 947       FSMap& fsmap,
 948       MonOpRequestRef op,
 949       const cmdmap_t& cmdmap,
 950       std::ostream &ss) override
 951   {
 952     std::string fs_name;
 953     cmd_getval(cmdmap, "fs_name", fs_name);
 954     auto fs = fsmap.get_filesystem(fs_name);
 955     if (fs == nullptr) {
 956         ss << "filesystem '" << fs_name << "' does not exist";
 957         return -ENOENT;
 958     }
 959
 960     fsmap.set_legacy_client_fscid(fs->fscid);
 961     return 0;
 962   }
 963 };
 964
 965 class RemoveFilesystemHandler : public FileSystemCommandHandler
 966 {
 967   public:
 968   RemoveFilesystemHandler()
 969     : FileSystemCommandHandler("fs rm")
 970   {}
 971
 972   int handle(
 973       Monitor *mon,
 974       FSMap& fsmap,
 975       MonOpRequestRef op,
 976       const cmdmap_t& cmdmap,
 977       std::ostream &ss) override
 978   {
 979     /* We may need to blocklist ranks. */
 980     if (!mon->osdmon()->is_writeable()) {
 981       // not allowed to write yet, so retry when we can
 982       mon->osdmon()->wait_for_writeable(op, new PaxosService::C_RetryMessage(mon->mdsmon(), op));
 983       return -EAGAIN;
 984     }
 985
 986     // Check caller has correctly named the FS to delete
 987     // (redundant while there is only one FS, but command
 988     //  syntax should apply to multi-FS future)
 989     string fs_name;
 990     cmd_getval(cmdmap, "fs_name", fs_name);
 991     auto fs = fsmap.get_filesystem(fs_name);
 992     if (fs == nullptr) {
 993         // Consider absence success to make deletes idempotent
 994         ss << "filesystem '" << fs_name << "' does not exist";
 995         return 0;
 996     }
 997
 998     // Check that no MDS daemons are active
 999     if (fs->mds_map.get_num_up_mds() > 0) {
1000       ss << "all MDS daemons must be inactive/failed before removing filesystem. See `ceph fs fail`.";
1001       return -EINVAL;
1002     }
1003
1004     // Check for confirmation flag
1005     bool sure = false;
1006     cmd_getval(cmdmap, "yes_i_really_mean_it", sure);
1007     if (!sure) {
1008       ss << "this is a DESTRUCTIVE operation and will make data in your filesystem permanently" \
1009             " inaccessible.  Add --yes-i-really-mean-it if you are sure you wish to continue.";
1010       return -EPERM;
1011     }
1012
1013     if (fsmap.get_legacy_client_fscid() == fs->fscid) {
1014       fsmap.set_legacy_client_fscid(FS_CLUSTER_ID_NONE);
1015     }
1016
1017     std::vector<mds_gid_t> to_fail;
1018     // There may be standby_replay daemons left here
1019     for (const auto &i : fs->mds_map.get_mds_info()) {
1020       ceph_assert(i.second.state == MDSMap::STATE_STANDBY_REPLAY);
1021       to_fail.push_back(i.first);
1022     }
1023
1024     for (const auto &gid : to_fail) {
1025       // Standby replays don't write, so it isn't important to
1026       // wait for an osdmap propose here: ignore return value.
1027       mon->mdsmon()->fail_mds_gid(fsmap, gid);
1028     }
1029     if (!to_fail.empty()) {
1030       mon->osdmon()->propose_pending(); /* maybe new blocklists */
1031     }
1032
1033     fsmap.erase_filesystem(fs->fscid);
1034
1035     return 0;
1036   }
1037 };
1038
1039 class ResetFilesystemHandler : public FileSystemCommandHandler
1040 {
1041   public:
1042   ResetFilesystemHandler()
1043     : FileSystemCommandHandler("fs reset")
1044   {}
1045
1046   int handle(
1047       Monitor *mon,
1048       FSMap& fsmap,
1049       MonOpRequestRef op,
1050       const cmdmap_t& cmdmap,
1051       std::ostream &ss) override
1052   {
1053     string fs_name;
1054     cmd_getval(cmdmap, "fs_name", fs_name);
1055     auto fs = fsmap.get_filesystem(fs_name);
1056     if (fs == nullptr) {
1057         ss << "filesystem '" << fs_name << "' does not exist";
1058         // Unlike fs rm, we consider this case an error
1059         return -ENOENT;
1060     }
1061
1062     // Check that no MDS daemons are active
1063     if (fs->mds_map.get_num_up_mds() > 0) {
1064       ss << "all MDS daemons must be inactive before resetting filesystem: set the cluster_down flag"
1065             " and use `ceph mds fail` to make this so";
1066       return -EINVAL;
1067     }
1068
1069     // Check for confirmation flag
1070     bool sure = false;
1071     cmd_getval(cmdmap, "yes_i_really_mean_it", sure);
1072     if (!sure) {
1073       ss << "this is a potentially destructive operation, only for use by experts in disaster recovery.  "
1074         "Add --yes-i-really-mean-it if you are sure you wish to continue.";
1075       return -EPERM;
1076     }
1077
1078     fsmap.reset_filesystem(fs->fscid);
1079
1080     return 0;
1081   }
1082 };
1083
1084 class RemoveDataPoolHandler : public FileSystemCommandHandler
1085 {
1086   public:
1087   RemoveDataPoolHandler()
1088     : FileSystemCommandHandler("fs rm_data_pool")
1089   {}
1090
1091   int handle(
1092       Monitor *mon,
1093       FSMap& fsmap,
1094       MonOpRequestRef op,
1095       const cmdmap_t& cmdmap,
1096       std::ostream &ss) override
1097   {
1098     string poolname;
1099     cmd_getval(cmdmap, "pool", poolname);
1100
1101     std::string fs_name;
1102     if (!cmd_getval(cmdmap, "fs_name", fs_name)
1103         || fs_name.empty()) {
1104       ss << "Missing filesystem name";
1105       return -EINVAL;
1106     }
1107
1108     int64_t poolid = mon->osdmon()->osdmap.lookup_pg_pool_name(poolname);
1109     if (poolid < 0) {
1110       string err;
1111       poolid = strict_strtol(poolname.c_str(), 10, &err);
1112       if (err.length()) {
1113         ss << "pool '" << poolname << "' does not exist";
1114         return -ENOENT;
1115       } else if (poolid < 0) {
1116         ss << "invalid pool id '" << poolid << "'";
1117         return -EINVAL;
1118       }
1119     }
1120
1121     ceph_assert(poolid >= 0);  // Checked by parsing code above
1122
1123     auto fs = fsmap.get_filesystem(fs_name);
1124     if (fs->mds_map.get_first_data_pool() == poolid) {
1125       ss << "cannot remove default data pool";
1126       return -EINVAL;
1127     }
1128
1129     int r = 0;
1130     fsmap.modify_filesystem(fs->fscid,
1131         [&r, poolid](std::shared_ptr<Filesystem> fs)
1132     {
1133       r = fs->mds_map.remove_data_pool(poolid);
1134     });
1135     if (r == -ENOENT) {
1136       // It was already removed, succeed in silence
1137       return 0;
1138     } else if (r == 0) {
1139       // We removed it, succeed
1140       ss << "removed data pool " << poolid << " from fsmap";
1141       return 0;
1142     } else {
1143       // Unexpected error, bubble up
1144       return r;
1145     }
1146   }
1147 };
1148
1149 /**
1150  * For commands with an alternative prefix
1151  */
1152 template<typename T>
1153 class AliasHandler : public T
1154 {
1155   std::string alias_prefix;
1156
1157   public:
1158   explicit AliasHandler(const std::string &new_prefix)
1159     : T()
1160   {
1161     alias_prefix = new_prefix;
1162   }
1163
1164   std::string const &get_prefix() const override {return alias_prefix;}
1165
1166   int handle(
1167       Monitor *mon,
1168       FSMap& fsmap,
1169       MonOpRequestRef op,
1170       const cmdmap_t& cmdmap,
1171       std::ostream &ss) override
1172   {
1173     return T::handle(mon, fsmap, op, cmdmap, ss);
1174   }
1175 };
1176
1177 class MirrorHandlerEnable : public FileSystemCommandHandler
1178 {
1179 public:
1180   MirrorHandlerEnable()
1181     : FileSystemCommandHandler("fs mirror enable")
1182   {}
1183
1184   int handle(Monitor *mon,
1185              FSMap &fsmap, MonOpRequestRef op,
1186              const cmdmap_t& cmdmap, std::ostream &ss) override {
1187     std::string fs_name;
1188     if (!cmd_getval(cmdmap, "fs_name", fs_name) || fs_name.empty()) {
1189       ss << "Missing filesystem name";
1190       return -EINVAL;
1191     }
1192
1193     auto fs = fsmap.get_filesystem(fs_name);
1194     if (fs == nullptr) {
1195       ss << "Filesystem '" << fs_name << "' not found";
1196       return -ENOENT;
1197     }
1198
1199     if (fs->mirror_info.is_mirrored()) {
1200       return 0;
1201     }
1202
1203     auto f = [](auto &&fs) {
1204                fs->mirror_info.enable_mirroring();
1205     };
1206     fsmap.modify_filesystem(fs->fscid, std::move(f));
1207
1208     return 0;
1209   }
1210 };
1211
1212 class MirrorHandlerDisable : public FileSystemCommandHandler
1213 {
1214 public:
1215   MirrorHandlerDisable()
1216     : FileSystemCommandHandler("fs mirror disable")
1217   {}
1218
1219   int handle(Monitor *mon,
1220              FSMap &fsmap, MonOpRequestRef op,
1221              const cmdmap_t& cmdmap, std::ostream &ss) override {
1222     std::string fs_name;
1223     if (!cmd_getval(cmdmap, "fs_name", fs_name) || fs_name.empty()) {
1224       ss << "Missing filesystem name";
1225       return -EINVAL;
1226     }
1227
1228     auto fs = fsmap.get_filesystem(fs_name);
1229     if (fs == nullptr) {
1230       ss << "Filesystem '" << fs_name << "' not found";
1231       return -ENOENT;
1232     }
1233
1234     if (!fs->mirror_info.is_mirrored()) {
1235       return 0;
1236     }
1237
1238     auto f = [](auto &&fs) {
1239       fs->mirror_info.disable_mirroring();
1240     };
1241     fsmap.modify_filesystem(fs->fscid, std::move(f));
1242
1243     return 0;
1244   }
1245 };
1246
1247 class MirrorHandlerAddPeer : public FileSystemCommandHandler
1248 {
1249 public:
1250   MirrorHandlerAddPeer()
1251     : FileSystemCommandHandler("fs mirror peer_add")
1252   {}
1253
1254   boost::optional<std::pair<string, string>>
1255   extract_remote_cluster_conf(const std::string &spec) {
1256     auto pos = spec.find("@");
1257     if (pos == std::string_view::npos) {
1258       return boost::optional<std::pair<string, string>>();
1259     }
1260
1261     auto client = spec.substr(0, pos);
1262     auto cluster = spec.substr(pos+1);
1263
1264     return std::make_pair(client, cluster);
1265   }
1266
1267   bool peer_add(FSMap &fsmap, Filesystem::const_ref &&fs,
1268                 const cmdmap_t &cmdmap, std::ostream &ss) {
1269     string peer_uuid;
1270     string remote_spec;
1271     string remote_fs_name;
1272     cmd_getval(cmdmap, "uuid", peer_uuid);
1273     cmd_getval(cmdmap, "remote_cluster_spec", remote_spec);
1274     cmd_getval(cmdmap, "remote_fs_name", remote_fs_name);
1275
1276     // verify (and extract) remote cluster specification
1277     auto remote_conf = extract_remote_cluster_conf(remote_spec);
1278     if (!remote_conf) {
1279       ss << "invalid remote cluster spec -- should be <client>@<cluster>";
1280       return false;
1281     }
1282
1283     if (fs->mirror_info.has_peer(peer_uuid)) {
1284       ss << "peer already exists";
1285       return true;
1286     }
1287     if (fs->mirror_info.has_peer((*remote_conf).first, (*remote_conf).second,
1288                                  remote_fs_name)) {
1289       ss << "peer already exists";
1290       return true;
1291     }
1292
1293     auto f = [peer_uuid, remote_conf, remote_fs_name](auto &&fs) {
1294                fs->mirror_info.peer_add(peer_uuid, (*remote_conf).first,
1295                                         (*remote_conf).second, remote_fs_name);
1296              };
1297     fsmap.modify_filesystem(fs->fscid, std::move(f));
1298     return true;
1299   }
1300
1301   int handle(Monitor *mon,
1302              FSMap &fsmap, MonOpRequestRef op,
1303              const cmdmap_t& cmdmap, std::ostream &ss) override {
1304     std::string fs_name;
1305     if (!cmd_getval(cmdmap, "fs_name", fs_name) || fs_name.empty()) {
1306       ss << "Missing filesystem name";
1307       return -EINVAL;
1308     }
1309
1310     auto fs = fsmap.get_filesystem(fs_name);
1311     if (fs == nullptr) {
1312       ss << "Filesystem '" << fs_name << "' not found";
1313       return -ENOENT;
1314     }
1315
1316     if (!fs->mirror_info.is_mirrored()) {
1317       ss << "Mirroring not enabled for filesystem '" << fs_name << "'";
1318       return -EINVAL;
1319     }
1320
1321     auto res = peer_add(fsmap, std::move(fs), cmdmap, ss);
1322     if (!res) {
1323       return -EINVAL;
1324     }
1325
1326     return 0;
1327   }
1328 };
1329
1330 class MirrorHandlerRemovePeer : public FileSystemCommandHandler
1331 {
1332 public:
1333   MirrorHandlerRemovePeer()
1334     : FileSystemCommandHandler("fs mirror peer_remove")
1335   {}
1336
1337   bool peer_remove(FSMap &fsmap, Filesystem::const_ref &&fs,
1338                    const cmdmap_t &cmdmap, std::ostream &ss) {
1339     string peer_uuid;
1340     cmd_getval(cmdmap, "uuid", peer_uuid);
1341
1342     if (!fs->mirror_info.has_peer(peer_uuid)) {
1343       ss << "cannot find peer with uuid: " << peer_uuid;
1344       return true;
1345     }
1346
1347     auto f = [peer_uuid](auto &&fs) {
1348                fs->mirror_info.peer_remove(peer_uuid);
1349              };
1350     fsmap.modify_filesystem(fs->fscid, std::move(f));
1351     return true;
1352   }
1353
1354   int handle(Monitor *mon,
1355              FSMap &fsmap, MonOpRequestRef op,
1356              const cmdmap_t& cmdmap, std::ostream &ss) override {
1357     std::string fs_name;
1358     if (!cmd_getval(cmdmap, "fs_name", fs_name) || fs_name.empty()) {
1359       ss << "Missing filesystem name";
1360       return -EINVAL;
1361     }
1362
1363     auto fs = fsmap.get_filesystem(fs_name);
1364     if (fs == nullptr) {
1365       ss << "Filesystem '" << fs_name << "' not found";
1366       return -ENOENT;
1367     }
1368
1369     if (!fs->mirror_info.is_mirrored()) {
1370       ss << "Mirroring not enabled for filesystem '" << fs_name << "'";
1371       return -EINVAL;
1372     }
1373
1374     auto res = peer_remove(fsmap, std::move(fs), cmdmap, ss);
1375     if (!res) {
1376       return -EINVAL;
1377     }
1378
1379     return 0;
1380   }
1381 };
1382
1383 std::list<std::shared_ptr<FileSystemCommandHandler> >
1384 FileSystemCommandHandler::load(Paxos *paxos)
1385 {
1386   std::list<std::shared_ptr<FileSystemCommandHandler> > handlers;
1387
1388   handlers.push_back(std::make_shared<SetHandler>());
1389   handlers.push_back(std::make_shared<FailHandler>());
1390   handlers.push_back(std::make_shared<FlagSetHandler>());
1391   handlers.push_back(std::make_shared<CompatSetHandler>());
1392   handlers.push_back(std::make_shared<RequiredClientFeaturesHandler>());
1393   handlers.push_back(std::make_shared<AddDataPoolHandler>(paxos));
1394   handlers.push_back(std::make_shared<RemoveDataPoolHandler>());
1395   handlers.push_back(std::make_shared<FsNewHandler>(paxos));
1396   handlers.push_back(std::make_shared<RemoveFilesystemHandler>());
1397   handlers.push_back(std::make_shared<ResetFilesystemHandler>());
1398
1399   handlers.push_back(std::make_shared<SetDefaultHandler>());
1400   handlers.push_back(std::make_shared<AliasHandler<SetDefaultHandler> >(
1401         "fs set_default"));
1402   handlers.push_back(std::make_shared<MirrorHandlerEnable>());
1403   handlers.push_back(std::make_shared<MirrorHandlerDisable>());
1404   handlers.push_back(std::make_shared<MirrorHandlerAddPeer>());
1405   handlers.push_back(std::make_shared<MirrorHandlerRemovePeer>());
1406
1407   return handlers;
1408 }
1409
1410 int FileSystemCommandHandler::_check_pool(
1411     OSDMap &osd_map,
1412     const int64_t pool_id,
1413     int type,
1414     bool force,
1415     std::ostream *ss) const
1416 {
1417   ceph_assert(ss != NULL);
1418
1419   const pg_pool_t *pool = osd_map.get_pg_pool(pool_id);
1420   if (!pool) {
1421     *ss << "pool id '" << pool_id << "' does not exist";
1422     return -ENOENT;
1423   }
1424
1425   const string& pool_name = osd_map.get_pool_name(pool_id);
1426
1427   if (pool->is_erasure()) {
1428     if (type == POOL_METADATA) {
1429       *ss << "pool '" << pool_name << "' (id '" << pool_id << "')"
1430          << " is an erasure-coded pool.  Use of erasure-coded pools"
1431          << " for CephFS metadata is not permitted";
1432       return -EINVAL;
1433     } else if (type == POOL_DATA_DEFAULT && !force) {
1434       *ss << "pool '" << pool_name << "' (id '" << pool_id << "')"
1435              " is an erasure-coded pool."
1436              " Use of an EC pool for the default data pool is discouraged;"
1437              " see the online CephFS documentation for more information."
1438              " Use --force to override.";
1439       return -EINVAL;
1440     } else if (!pool->allows_ecoverwrites()) {
1441       // non-overwriteable EC pools are only acceptable with a cache tier overlay
1442       if (!pool->has_tiers() || !pool->has_read_tier() || !pool->has_write_tier()) {
1443         *ss << "pool '" << pool_name << "' (id '" << pool_id << "')"
1444             << " is an erasure-coded pool, with no overwrite support";
1445         return -EINVAL;
1446       }
1447
1448       // That cache tier overlay must be writeback, not readonly (it's the
1449       // write operations like modify+truncate we care about support for)
1450       const pg_pool_t *write_tier = osd_map.get_pg_pool(
1451           pool->write_tier);
1452       ceph_assert(write_tier != NULL);  // OSDMonitor shouldn't allow DNE tier
1453       if (write_tier->cache_mode == pg_pool_t::CACHEMODE_FORWARD
1454           || write_tier->cache_mode == pg_pool_t::CACHEMODE_READONLY) {
1455         *ss << "EC pool '" << pool_name << "' has a write tier ("
1456             << osd_map.get_pool_name(pool->write_tier)
1457             << ") that is configured "
1458                "to forward writes.  Use a cache mode such as 'writeback' for "
1459                "CephFS";
1460         return -EINVAL;
1461       }
1462     }
1463   }
1464
1465   if (pool->is_tier()) {
1466     *ss << " pool '" << pool_name << "' (id '" << pool_id
1467       << "') is already in use as a cache tier.";
1468     return -EINVAL;
1469   }
1470
1471   if (!force && !pool->application_metadata.empty() &&
1472       pool->application_metadata.count(
1473         pg_pool_t::APPLICATION_NAME_CEPHFS) == 0) {
1474     *ss << " pool '" << pool_name << "' (id '" << pool_id
1475         << "') has a non-CephFS application enabled.";
1476     return -EINVAL;
1477   }
1478
1479   // Nothing special about this pool, so it is permissible
1480   return 0;
1481 }
1482
1483 int FileSystemCommandHandler::is_op_allowed(
1484     const MonOpRequestRef& op, const FSMap& fsmap, const cmdmap_t& cmdmap,
1485     std::ostream &ss) const
1486 {
1487     string fs_name;
1488     cmd_getval(cmdmap, "fs_name", fs_name);
1489
1490     // so that fsmap can filtered and the original copy is untouched.
1491     FSMap fsmap_copy = fsmap;
1492     fsmap_copy.filter(op->get_session()->get_allowed_fs_names());
1493
1494     auto fs = fsmap_copy.get_filesystem(fs_name);
1495     if (fs == nullptr) {
1496       /* let "fs rm" handle idempotent case where file system does not exist */
1497       if (!(get_prefix() == "fs rm" && fsmap.get_filesystem(fs_name) == nullptr)) {
1498         ss << "Filesystem not found: '" << fs_name << "'";
1499         return -ENOENT;
1500       }
1501     }
1502
1503     if (!op->get_session()->fs_name_capable(fs_name, MON_CAP_W)) {
1504       ss << "Permission denied: '" << fs_name << "'";
1505       return -EPERM;
1506     }
1507
1508   return 1;
1509 }