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