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 #include "PGMonitor.h"
  18
  19 #include "FSCommands.h"
  20 #include "MDSMonitor.h"
  21
  22
  23
  24 static const string EXPERIMENTAL_WARNING("Warning! This feature is experimental."
  25 "It may cause problems up to and including data loss."
  26 "Consult the documentation at ceph.com, and if unsure, do not proceed."
  27 "Add --yes-i-really-mean-it if you are certain.");
  28
  29
  30
  31 class FlagSetHandler : public FileSystemCommandHandler
  32 {
  33   public:
  34   FlagSetHandler()
  35     : FileSystemCommandHandler("fs flag set")
  36   {
  37   }
  38
  39   int handle(
  40       Monitor *mon,
  41       FSMap &fsmap,
  42       MonOpRequestRef op,
  43       map<string, cmd_vartype> &cmdmap,
  44       std::stringstream &ss) override
  45   {
  46     string flag_name;
  47     cmd_getval(g_ceph_context, cmdmap, "flag_name", flag_name);
  48
  49     string flag_val;
  50     cmd_getval(g_ceph_context, cmdmap, "val", flag_val);
  51
  52     string confirm;
  53     cmd_getval(g_ceph_context, cmdmap, "confirm", confirm);
  54
  55     if (flag_name == "enable_multiple") {
  56       bool flag_bool = false;
  57       int r = parse_bool(flag_val, &flag_bool, ss);
  58       if (r != 0) {
  59         ss << "Invalid boolean value '" << flag_val << "'";
  60         return r;
  61       }
  62
  63       bool jewel = mon->get_quorum_con_features() & CEPH_FEATURE_SERVER_JEWEL;
  64       if (flag_bool && !jewel) {
  65         ss << "Multiple-filesystems are forbidden until all mons are updated";
  66         return -EINVAL;
  67       }
  68       if (confirm != "--yes-i-really-mean-it") {
  69         ss << EXPERIMENTAL_WARNING;
  70       }
  71       fsmap.set_enable_multiple(flag_bool);
  72       return 0;
  73     } else {
  74       ss << "Unknown flag '" << flag_name << "'";
  75       return -EINVAL;
  76     }
  77   }
  78 };
  79
  80 class FsNewHandler : public FileSystemCommandHandler
  81 {
  82   public:
  83   FsNewHandler(Paxos *paxos)
  84     : FileSystemCommandHandler("fs new"), m_paxos(paxos)
  85   {
  86   }
  87
  88   bool batched_propose() override {
  89     return true;
  90   }
  91
  92   int handle(
  93       Monitor *mon,
  94       FSMap &fsmap,
  95       MonOpRequestRef op,
  96       map<string, cmd_vartype> &cmdmap,
  97       std::stringstream &ss) override
  98   {
  99     assert(m_paxos->is_plugged());
 100
 101     string metadata_name;
 102     cmd_getval(g_ceph_context, cmdmap, "metadata", metadata_name);
 103     int64_t metadata = mon->osdmon()->osdmap.lookup_pg_pool_name(metadata_name);
 104     if (metadata < 0) {
 105       ss << "pool '" << metadata_name << "' does not exist";
 106       return -ENOENT;
 107     }
 108
 109     string force_str;
 110     cmd_getval(g_ceph_context,cmdmap, "force", force_str);
 111     bool force = (force_str == "--force");
 112     const pool_stat_t *stat = mon->pgservice->get_pool_stat(metadata);
 113     if (stat) {
 114       int64_t metadata_num_objects = stat->stats.sum.num_objects;
 115       if (!force && metadata_num_objects > 0) {
 116         ss << "pool '" << metadata_name
 117            << "' already contains some objects. Use an empty pool instead.";
 118         return -EINVAL;
 119       }
 120     }
 121
 122     string data_name;
 123     cmd_getval(g_ceph_context, cmdmap, "data", data_name);
 124     int64_t data = mon->osdmon()->osdmap.lookup_pg_pool_name(data_name);
 125     if (data < 0) {
 126       ss << "pool '" << data_name << "' does not exist";
 127       return -ENOENT;
 128     }
 129     if (data == 0) {
 130       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.";
 131       return -EINVAL;
 132     }
 133
 134     string fs_name;
 135     cmd_getval(g_ceph_context, cmdmap, "fs_name", fs_name);
 136     if (fs_name.empty()) {
 137         // Ensure fs name is not empty so that we can implement
 138         // commmands that refer to FS by name in future.
 139         ss << "Filesystem name may not be empty";
 140         return -EINVAL;
 141     }
 142
 143     if (fsmap.get_filesystem(fs_name)) {
 144       auto fs = fsmap.get_filesystem(fs_name);
 145       if (*(fs->mds_map.get_data_pools().begin()) == data
 146           && fs->mds_map.get_metadata_pool() == metadata) {
 147         // Identical FS created already, this is a no-op
 148         ss << "filesystem '" << fs_name << "' already exists";
 149         return 0;
 150       } else {
 151         ss << "filesystem already exists with name '" << fs_name << "'";
 152         return -EINVAL;
 153       }
 154     }
 155
 156     if (fsmap.filesystem_count() > 0
 157         && !fsmap.get_enable_multiple()) {
 158       ss << "Creation of multiple filesystems is disabled.  To enable "
 159             "this experimental feature, use 'ceph fs flag set enable_multiple "
 160             "true'";
 161       return -EINVAL;
 162     }
 163
 164     for (auto fs : fsmap.get_filesystems()) {
 165       const std::vector<int64_t> &data_pools = fs->mds_map.get_data_pools();
 166       string sure;
 167       if ((std::find(data_pools.begin(), data_pools.end(), data) != data_pools.end()
 168            || fs->mds_map.get_metadata_pool() == metadata)
 169           && ((!cmd_getval(g_ceph_context, cmdmap, "sure", sure)
 170                || sure != "--allow-dangerous-metadata-overlay"))) {
 171         ss << "Filesystem '" << fs_name
 172            << "' 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.";
 173         return -EEXIST;
 174       }
 175     }
 176
 177     pg_pool_t const *data_pool = mon->osdmon()->osdmap.get_pg_pool(data);
 178     assert(data_pool != NULL);  // Checked it existed above
 179     pg_pool_t const *metadata_pool = mon->osdmon()->osdmap.get_pg_pool(metadata);
 180     assert(metadata_pool != NULL);  // Checked it existed above
 181
 182     // we must make these checks before we even allow ourselves to *think*
 183     // about requesting a proposal to the osdmonitor and bail out now if
 184     // we believe we must.  bailing out *after* we request the proposal is
 185     // bad business as we could have changed the osdmon's state and ending up
 186     // returning an error to the user.
 187     int r = _check_pool(mon->osdmon()->osdmap, data, false, force, &ss);
 188     if (r < 0) {
 189       return r;
 190     }
 191
 192     r = _check_pool(mon->osdmon()->osdmap, metadata, true, force, &ss);
 193     if (r < 0) {
 194       return r;
 195     }
 196
 197     mon->osdmon()->do_application_enable(data,
 198                                          pg_pool_t::APPLICATION_NAME_CEPHFS);
 199     mon->osdmon()->do_application_enable(metadata,
 200                                          pg_pool_t::APPLICATION_NAME_CEPHFS);
 201
 202     // All checks passed, go ahead and create.
 203     fsmap.create_filesystem(fs_name, metadata, data,
 204                             mon->get_quorum_con_features());
 205     ss << "new fs with metadata pool " << metadata << " and data pool " << data;
 206     return 0;
 207   }
 208
 209 private:
 210   Paxos *m_paxos;
 211 };
 212
 213 class SetHandler : public FileSystemCommandHandler
 214 {
 215 public:
 216   SetHandler()
 217     : FileSystemCommandHandler("fs set")
 218   {}
 219
 220   int handle(
 221       Monitor *mon,
 222       FSMap &fsmap,
 223       MonOpRequestRef op,
 224       map<string, cmd_vartype> &cmdmap,
 225       std::stringstream &ss) override
 226   {
 227     std::string fs_name;
 228     if (!cmd_getval(g_ceph_context, cmdmap, "fs_name", fs_name) || fs_name.empty()) {
 229       ss << "Missing filesystem name";
 230       return -EINVAL;
 231     }
 232
 233     auto fs = fsmap.get_filesystem(fs_name);
 234     if (fs == nullptr) {
 235       ss << "Not found: '" << fs_name << "'";
 236       return -ENOENT;
 237     }
 238
 239     string var;
 240     if (!cmd_getval(g_ceph_context, cmdmap, "var", var) || var.empty()) {
 241       ss << "Invalid variable";
 242       return -EINVAL;
 243     }
 244     string val;
 245     string interr;
 246     int64_t n = 0;
 247     if (!cmd_getval(g_ceph_context, cmdmap, "val", val)) {
 248       return -EINVAL;
 249     }
 250     // we got a string.  see if it contains an int.
 251     n = strict_strtoll(val.c_str(), 10, &interr);
 252     if (var == "max_mds") {
 253       // NOTE: see also "mds set_max_mds", which can modify the same field.
 254       if (interr.length()) {
 255         ss << interr;
 256         return -EINVAL;
 257       }
 258
 259       if (n <= 0) {
 260         ss << "You must specify at least one MDS";
 261         return -EINVAL;
 262       }
 263
 264       if (!fs->mds_map.allows_multimds() && n > fs->mds_map.get_max_mds() &&
 265           n > 1) {
 266         ss << "multi-MDS clusters are not enabled; set 'allow_multimds' to enable";
 267         return -EINVAL;
 268       }
 269       if (n > MAX_MDS) {
 270         ss << "may not have more than " << MAX_MDS << " MDS ranks";
 271         return -EINVAL;
 272       }
 273       fsmap.modify_filesystem(
 274           fs->fscid,
 275           [n](std::shared_ptr<Filesystem> fs)
 276       {
 277         fs->mds_map.set_max_mds(n);
 278       });
 279     } else if (var == "inline_data") {
 280       bool enable_inline = false;
 281       int r = parse_bool(val, &enable_inline, ss);
 282       if (r != 0) {
 283         return r;
 284       }
 285
 286       if (enable_inline) {
 287         string confirm;
 288         if (!cmd_getval(g_ceph_context, cmdmap, "confirm", confirm) ||
 289             confirm != "--yes-i-really-mean-it") {
 290           ss << EXPERIMENTAL_WARNING;
 291           return -EPERM;
 292         }
 293         ss << "inline data enabled";
 294
 295         fsmap.modify_filesystem(
 296             fs->fscid,
 297             [](std::shared_ptr<Filesystem> fs)
 298         {
 299           fs->mds_map.set_inline_data_enabled(true);
 300         });
 301
 302         // Update `compat`
 303         CompatSet c = fsmap.get_compat();
 304         c.incompat.insert(MDS_FEATURE_INCOMPAT_INLINE);
 305         fsmap.update_compat(c);
 306       } else {
 307         ss << "inline data disabled";
 308         fsmap.modify_filesystem(
 309             fs->fscid,
 310             [](std::shared_ptr<Filesystem> fs)
 311         {
 312           fs->mds_map.set_inline_data_enabled(false);
 313         });
 314       }
 315     } else if (var == "balancer") {
 316       if (val.empty()) {
 317         ss << "unsetting the metadata load balancer";
 318       } else {
 319         ss << "setting the metadata load balancer to " << val;
 320       }
 321       fsmap.modify_filesystem(
 322         fs->fscid,
 323         [val](std::shared_ptr<Filesystem> fs)
 324         {
 325           fs->mds_map.set_balancer(val);
 326         });
 327       return true;
 328     } else if (var == "max_file_size") {
 329       if (interr.length()) {
 330         ss << var << " requires an integer value";
 331         return -EINVAL;
 332       }
 333       if (n < CEPH_MIN_STRIPE_UNIT) {
 334         ss << var << " must at least " << CEPH_MIN_STRIPE_UNIT;
 335         return -ERANGE;
 336       }
 337       fsmap.modify_filesystem(
 338           fs->fscid,
 339           [n](std::shared_ptr<Filesystem> fs)
 340       {
 341         fs->mds_map.set_max_filesize(n);
 342       });
 343     } else if (var == "allow_new_snaps") {
 344       bool enable_snaps = false;
 345       int r = parse_bool(val, &enable_snaps, ss);
 346       if (r != 0) {
 347         return r;
 348       }
 349
 350       if (!enable_snaps) {
 351         fsmap.modify_filesystem(
 352             fs->fscid,
 353             [](std::shared_ptr<Filesystem> fs)
 354         {
 355           fs->mds_map.clear_snaps_allowed();
 356         });
 357         ss << "disabled new snapshots";
 358       } else {
 359         string confirm;
 360         if (!cmd_getval(g_ceph_context, cmdmap, "confirm", confirm) ||
 361             confirm != "--yes-i-really-mean-it") {
 362           ss << EXPERIMENTAL_WARNING;
 363           return -EPERM;
 364         }
 365         fsmap.modify_filesystem(
 366             fs->fscid,
 367             [](std::shared_ptr<Filesystem> fs)
 368         {
 369           fs->mds_map.set_snaps_allowed();
 370         });
 371         ss << "enabled new snapshots";
 372       }
 373     } else if (var == "allow_multimds") {
 374       bool enable_multimds = false;
 375       int r = parse_bool(val, &enable_multimds, ss);
 376       if (r != 0) {
 377         return r;
 378       }
 379
 380       if (!enable_multimds) {
 381         fsmap.modify_filesystem(fs->fscid,
 382              [](std::shared_ptr<Filesystem> fs)
 383                 {
 384                   fs->mds_map.clear_multimds_allowed();
 385                 });
 386         ss << "disallowed increasing the cluster size past 1";
 387       } else {
 388         fsmap.modify_filesystem(
 389             fs->fscid,
 390             [](std::shared_ptr<Filesystem> fs)
 391         {
 392           fs->mds_map.set_multimds_allowed();
 393         });
 394         ss << "enabled creation of more than 1 active MDS";
 395       }
 396     } else if (var == "allow_dirfrags") {
 397       bool enable_dirfrags = false;
 398       int r = parse_bool(val, &enable_dirfrags, ss);
 399       if (r != 0) {
 400         return r;
 401       }
 402
 403       if (!enable_dirfrags) {
 404         fsmap.modify_filesystem(fs->fscid,
 405              [](std::shared_ptr<Filesystem> fs)
 406                 {
 407                   fs->mds_map.clear_dirfrags_allowed();
 408                 });
 409         ss << "disallowed new directory fragmentation";
 410       } else {
 411         fsmap.modify_filesystem(
 412             fs->fscid,
 413             [](std::shared_ptr<Filesystem> fs)
 414         {
 415           fs->mds_map.set_dirfrags_allowed();
 416         });
 417         ss << "enabled directory fragmentation";
 418       }
 419     } else if (var == "cluster_down") {
 420       bool is_down = false;
 421       int r = parse_bool(val, &is_down, ss);
 422       if (r != 0) {
 423         return r;
 424       }
 425
 426       fsmap.modify_filesystem(
 427           fs->fscid,
 428           [is_down](std::shared_ptr<Filesystem> fs)
 429       {
 430         if (is_down) {
 431           fs->mds_map.set_flag(CEPH_MDSMAP_DOWN);
 432         } else {
 433           fs->mds_map.clear_flag(CEPH_MDSMAP_DOWN);
 434         }
 435       });
 436
 437       ss << "marked " << (is_down ? "down" : "up");
 438     } else if (var == "standby_count_wanted") {
 439       if (interr.length()) {
 440        ss << var << " requires an integer value";
 441        return -EINVAL;
 442       }
 443       if (n < 0) {
 444        ss << var << " must be non-negative";
 445        return -ERANGE;
 446       }
 447       fsmap.modify_filesystem(
 448           fs->fscid,
 449           [n](std::shared_ptr<Filesystem> fs)
 450       {
 451         fs->mds_map.set_standby_count_wanted(n);
 452       });
 453     } else {
 454       ss << "unknown variable " << var;
 455       return -EINVAL;
 456     }
 457
 458     return 0;
 459   }
 460 };
 461
 462 class AddDataPoolHandler : public FileSystemCommandHandler
 463 {
 464   public:
 465   AddDataPoolHandler(Paxos *paxos)
 466     : FileSystemCommandHandler("fs add_data_pool"), m_paxos(paxos)
 467   {}
 468
 469   bool batched_propose() override {
 470     return true;
 471   }
 472
 473   int handle(
 474       Monitor *mon,
 475       FSMap &fsmap,
 476       MonOpRequestRef op,
 477       map<string, cmd_vartype> &cmdmap,
 478       std::stringstream &ss) override
 479   {
 480     assert(m_paxos->is_plugged());
 481
 482     string poolname;
 483     cmd_getval(g_ceph_context, cmdmap, "pool", poolname);
 484
 485     std::string fs_name;
 486     if (!cmd_getval(g_ceph_context, cmdmap, "fs_name", fs_name)
 487         || fs_name.empty()) {
 488       ss << "Missing filesystem name";
 489       return -EINVAL;
 490     }
 491
 492     auto fs = fsmap.get_filesystem(fs_name);
 493     if (fs == nullptr) {
 494       ss << "Not found: '" << fs_name << "'";
 495       return -ENOENT;
 496     }
 497
 498     int64_t poolid = mon->osdmon()->osdmap.lookup_pg_pool_name(poolname);
 499     if (poolid < 0) {
 500       string err;
 501       poolid = strict_strtol(poolname.c_str(), 10, &err);
 502       if (err.length()) {
 503         ss << "pool '" << poolname << "' does not exist";
 504         return -ENOENT;
 505       }
 506     }
 507
 508     int r = _check_pool(mon->osdmon()->osdmap, poolid, false, false, &ss);
 509     if (r != 0) {
 510       return r;
 511     }
 512
 513     // no-op when the data_pool already on fs
 514     if (fs->mds_map.is_data_pool(poolid)) {
 515       ss << "data pool " << poolid << " is already on fs " << fs_name;
 516       return 0;
 517     }
 518
 519     mon->osdmon()->do_application_enable(poolid,
 520                                          pg_pool_t::APPLICATION_NAME_CEPHFS);
 521
 522     fsmap.modify_filesystem(
 523         fs->fscid,
 524         [poolid](std::shared_ptr<Filesystem> fs)
 525     {
 526       fs->mds_map.add_data_pool(poolid);
 527     });
 528
 529     ss << "added data pool " << poolid << " to fsmap";
 530
 531     return 0;
 532   }
 533
 534 private:
 535   Paxos *m_paxos;
 536 };
 537
 538 class SetDefaultHandler : public FileSystemCommandHandler
 539 {
 540   public:
 541   SetDefaultHandler()
 542     : FileSystemCommandHandler("fs set-default")
 543   {}
 544
 545   int handle(
 546       Monitor *mon,
 547       FSMap &fsmap,
 548       MonOpRequestRef op,
 549       map<string, cmd_vartype> &cmdmap,
 550       std::stringstream &ss) override
 551   {
 552     std::string fs_name;
 553     cmd_getval(g_ceph_context, cmdmap, "fs_name", fs_name);
 554     auto fs = fsmap.get_filesystem(fs_name);
 555     if (fs == nullptr) {
 556         ss << "filesystem '" << fs_name << "' does not exist";
 557         return -ENOENT;
 558     }
 559
 560     fsmap.set_legacy_client_fscid(fs->fscid);
 561     return 0;
 562   }
 563 };
 564
 565 class RemoveFilesystemHandler : public FileSystemCommandHandler
 566 {
 567   public:
 568   RemoveFilesystemHandler()
 569     : FileSystemCommandHandler("fs rm")
 570   {}
 571
 572   int handle(
 573       Monitor *mon,
 574       FSMap &fsmap,
 575       MonOpRequestRef op,
 576       map<string, cmd_vartype> &cmdmap,
 577       std::stringstream &ss) override
 578   {
 579     // Check caller has correctly named the FS to delete
 580     // (redundant while there is only one FS, but command
 581     //  syntax should apply to multi-FS future)
 582     string fs_name;
 583     cmd_getval(g_ceph_context, cmdmap, "fs_name", fs_name);
 584     auto fs = fsmap.get_filesystem(fs_name);
 585     if (fs == nullptr) {
 586         // Consider absence success to make deletes idempotent
 587         ss << "filesystem '" << fs_name << "' does not exist";
 588         return 0;
 589     }
 590
 591     // Check that no MDS daemons are active
 592     if (fs->mds_map.get_num_up_mds() > 0) {
 593       ss << "all MDS daemons must be inactive before removing filesystem";
 594       return -EINVAL;
 595     }
 596
 597     // Check for confirmation flag
 598     string sure;
 599     cmd_getval(g_ceph_context, cmdmap, "sure", sure);
 600     if (sure != "--yes-i-really-mean-it") {
 601       ss << "this is a DESTRUCTIVE operation and will make data in your filesystem permanently" \
 602             " inaccessible.  Add --yes-i-really-mean-it if you are sure you wish to continue.";
 603       return -EPERM;
 604     }
 605
 606     if (fsmap.get_legacy_client_fscid() == fs->fscid) {
 607       fsmap.set_legacy_client_fscid(FS_CLUSTER_ID_NONE);
 608     }
 609
 610     std::vector<mds_gid_t> to_fail;
 611     // There may be standby_replay daemons left here
 612     for (const auto &i : fs->mds_map.get_mds_info()) {
 613       assert(i.second.state == MDSMap::STATE_STANDBY_REPLAY);
 614       to_fail.push_back(i.first);
 615     }
 616
 617     for (const auto &gid : to_fail) {
 618       // Standby replays don't write, so it isn't important to
 619       // wait for an osdmap propose here: ignore return value.
 620       mon->mdsmon()->fail_mds_gid(gid);
 621     }
 622
 623     fsmap.erase_filesystem(fs->fscid);
 624
 625     return 0;
 626   }
 627 };
 628
 629 class ResetFilesystemHandler : public FileSystemCommandHandler
 630 {
 631   public:
 632   ResetFilesystemHandler()
 633     : FileSystemCommandHandler("fs reset")
 634   {}
 635
 636   int handle(
 637       Monitor *mon,
 638       FSMap &fsmap,
 639       MonOpRequestRef op,
 640       map<string, cmd_vartype> &cmdmap,
 641       std::stringstream &ss) override
 642   {
 643     string fs_name;
 644     cmd_getval(g_ceph_context, cmdmap, "fs_name", fs_name);
 645     auto fs = fsmap.get_filesystem(fs_name);
 646     if (fs == nullptr) {
 647         ss << "filesystem '" << fs_name << "' does not exist";
 648         // Unlike fs rm, we consider this case an error
 649         return -ENOENT;
 650     }
 651
 652     // Check that no MDS daemons are active
 653     if (fs->mds_map.get_num_up_mds() > 0) {
 654       ss << "all MDS daemons must be inactive before resetting filesystem: set the cluster_down flag"
 655             " and use `ceph mds fail` to make this so";
 656       return -EINVAL;
 657     }
 658
 659     // Check for confirmation flag
 660     string sure;
 661     cmd_getval(g_ceph_context, cmdmap, "sure", sure);
 662     if (sure != "--yes-i-really-mean-it") {
 663       ss << "this is a potentially destructive operation, only for use by experts in disaster recovery.  "
 664         "Add --yes-i-really-mean-it if you are sure you wish to continue.";
 665       return -EPERM;
 666     }
 667
 668     fsmap.reset_filesystem(fs->fscid);
 669
 670     return 0;
 671   }
 672 };
 673
 674 class RemoveDataPoolHandler : public FileSystemCommandHandler
 675 {
 676   public:
 677   RemoveDataPoolHandler()
 678     : FileSystemCommandHandler("fs rm_data_pool")
 679   {}
 680
 681   int handle(
 682       Monitor *mon,
 683       FSMap &fsmap,
 684       MonOpRequestRef op,
 685       map<string, cmd_vartype> &cmdmap,
 686       std::stringstream &ss) override
 687   {
 688     string poolname;
 689     cmd_getval(g_ceph_context, cmdmap, "pool", poolname);
 690
 691     std::string fs_name;
 692     if (!cmd_getval(g_ceph_context, cmdmap, "fs_name", fs_name)
 693         || fs_name.empty()) {
 694       ss << "Missing filesystem name";
 695       return -EINVAL;
 696     }
 697
 698     auto fs = fsmap.get_filesystem(fs_name);
 699     if (fs == nullptr) {
 700       ss << "Not found: '" << fs_name << "'";
 701       return -ENOENT;
 702     }
 703
 704     int64_t poolid = mon->osdmon()->osdmap.lookup_pg_pool_name(poolname);
 705     if (poolid < 0) {
 706       string err;
 707       poolid = strict_strtol(poolname.c_str(), 10, &err);
 708       if (err.length()) {
 709         ss << "pool '" << poolname << "' does not exist";
 710         return -ENOENT;
 711       } else if (poolid < 0) {
 712         ss << "invalid pool id '" << poolid << "'";
 713         return -EINVAL;
 714       }
 715     }
 716
 717     assert(poolid >= 0);  // Checked by parsing code above
 718
 719     if (fs->mds_map.get_first_data_pool() == poolid) {
 720       ss << "cannot remove default data pool";
 721       return -EINVAL;
 722     }
 723
 724
 725     int r = 0;
 726     fsmap.modify_filesystem(fs->fscid,
 727         [&r, poolid](std::shared_ptr<Filesystem> fs)
 728     {
 729       r = fs->mds_map.remove_data_pool(poolid);
 730     });
 731     if (r == -ENOENT) {
 732       // It was already removed, succeed in silence
 733       return 0;
 734     } else if (r == 0) {
 735       // We removed it, succeed
 736       ss << "removed data pool " << poolid << " from fsmap";
 737       return 0;
 738     } else {
 739       // Unexpected error, bubble up
 740       return r;
 741     }
 742   }
 743 };
 744
 745
 746 /**
 747  * For commands that refer to a particular filesystem,
 748  * enable wrapping to implement the legacy version of
 749  * the command (like "mds add_data_pool" vs "fs add_data_pool")
 750  *
 751  * The wrapped handler must expect a fs_name argument in
 752  * its command map.
 753  */
 754 template<typename T>
 755 class LegacyHandler : public T
 756 {
 757   std::string legacy_prefix;
 758
 759   public:
 760   template <typename... Args>
 761   LegacyHandler(const std::string &new_prefix, Args&&... args)
 762     : T(std::forward<Args>(args)...)
 763   {
 764     legacy_prefix = new_prefix;
 765   }
 766
 767   std::string const &get_prefix() override {return legacy_prefix;}
 768
 769   int handle(
 770       Monitor *mon,
 771       FSMap &fsmap,
 772       MonOpRequestRef op,
 773       map<string, cmd_vartype> &cmdmap,
 774       std::stringstream &ss) override
 775   {
 776     auto fs = fsmap.get_legacy_filesystem();
 777     if (fs == nullptr) {
 778       ss << "No filesystem configured";
 779       return -ENOENT;
 780     }
 781     std::map<string, cmd_vartype> modified = cmdmap;
 782     modified["fs_name"] = fs->mds_map.get_fs_name();
 783     return T::handle(mon, fsmap, op, modified, ss);
 784   }
 785 };
 786
 787 /**
 788  * For commands with an alternative prefix
 789  */
 790 template<typename T>
 791 class AliasHandler : public T
 792 {
 793   std::string alias_prefix;
 794
 795   public:
 796   AliasHandler(const std::string &new_prefix)
 797     : T()
 798   {
 799     alias_prefix = new_prefix;
 800   }
 801
 802   std::string const &get_prefix() override {return alias_prefix;}
 803
 804   int handle(
 805       Monitor *mon,
 806       FSMap &fsmap,
 807       MonOpRequestRef op,
 808       map<string, cmd_vartype> &cmdmap,
 809       std::stringstream &ss) override
 810   {
 811     return T::handle(mon, fsmap, op, cmdmap, ss);
 812   }
 813 };
 814
 815
 816 std::list<std::shared_ptr<FileSystemCommandHandler> >
 817 FileSystemCommandHandler::load(Paxos *paxos)
 818 {
 819   std::list<std::shared_ptr<FileSystemCommandHandler> > handlers;
 820
 821   handlers.push_back(std::make_shared<SetHandler>());
 822   handlers.push_back(std::make_shared<LegacyHandler<SetHandler> >("mds set"));
 823   handlers.push_back(std::make_shared<FlagSetHandler>());
 824   handlers.push_back(std::make_shared<AddDataPoolHandler>(paxos));
 825   handlers.push_back(std::make_shared<LegacyHandler<AddDataPoolHandler> >(
 826         "mds add_data_pool", paxos));
 827   handlers.push_back(std::make_shared<RemoveDataPoolHandler>());
 828   handlers.push_back(std::make_shared<LegacyHandler<RemoveDataPoolHandler> >(
 829         "mds remove_data_pool"));
 830   handlers.push_back(std::make_shared<LegacyHandler<RemoveDataPoolHandler> >(
 831         "mds rm_data_pool"));
 832   handlers.push_back(std::make_shared<FsNewHandler>(paxos));
 833   handlers.push_back(std::make_shared<RemoveFilesystemHandler>());
 834   handlers.push_back(std::make_shared<ResetFilesystemHandler>());
 835
 836   handlers.push_back(std::make_shared<SetDefaultHandler>());
 837   handlers.push_back(std::make_shared<AliasHandler<SetDefaultHandler> >(
 838         "fs set_default"));
 839
 840   return handlers;
 841 }
 842
 843 int FileSystemCommandHandler::parse_bool(
 844       const std::string &bool_str,
 845       bool *result,
 846       std::ostream &ss)
 847 {
 848   assert(result != nullptr);
 849
 850   string interr;
 851   int64_t n = strict_strtoll(bool_str.c_str(), 10, &interr);
 852
 853   if (bool_str == "false" || bool_str == "no"
 854       || (interr.length() == 0 && n == 0)) {
 855     *result = false;
 856     return 0;
 857   } else if (bool_str == "true" || bool_str == "yes"
 858       || (interr.length() == 0 && n == 1)) {
 859     *result = true;
 860     return 0;
 861   } else {
 862     ss << "value must be false|no|0 or true|yes|1";
 863     return -EINVAL;
 864   }
 865 }
 866
 867 int FileSystemCommandHandler::_check_pool(
 868     OSDMap &osd_map,
 869     const int64_t pool_id,
 870     bool metadata,
 871     bool force,
 872     std::stringstream *ss) const
 873 {
 874   assert(ss != NULL);
 875
 876   const pg_pool_t *pool = osd_map.get_pg_pool(pool_id);
 877   if (!pool) {
 878     *ss << "pool id '" << pool_id << "' does not exist";
 879     return -ENOENT;
 880   }
 881
 882   const string& pool_name = osd_map.get_pool_name(pool_id);
 883
 884   if (pool->is_erasure() && metadata) {
 885       *ss << "pool '" << pool_name << "' (id '" << pool_id << "')"
 886          << " is an erasure-coded pool.  Use of erasure-coded pools"
 887          << " for CephFS metadata is not permitted";
 888     return -EINVAL;
 889   } else if (pool->is_erasure() && !pool->allows_ecoverwrites()) {
 890     // non-overwriteable EC pools are only acceptable with a cache tier overlay
 891     if (!pool->has_tiers() || !pool->has_read_tier() || !pool->has_write_tier()) {
 892       *ss << "pool '" << pool_name << "' (id '" << pool_id << "')"
 893          << " is an erasure-coded pool, with no overwrite support";
 894       return -EINVAL;
 895     }
 896
 897     // That cache tier overlay must be writeback, not readonly (it's the
 898     // write operations like modify+truncate we care about support for)
 899     const pg_pool_t *write_tier = osd_map.get_pg_pool(
 900         pool->write_tier);
 901     assert(write_tier != NULL);  // OSDMonitor shouldn't allow DNE tier
 902     if (write_tier->cache_mode == pg_pool_t::CACHEMODE_FORWARD
 903         || write_tier->cache_mode == pg_pool_t::CACHEMODE_READONLY) {
 904       *ss << "EC pool '" << pool_name << "' has a write tier ("
 905           << osd_map.get_pool_name(pool->write_tier)
 906           << ") that is configured "
 907              "to forward writes.  Use a cache mode such as 'writeback' for "
 908              "CephFS";
 909       return -EINVAL;
 910     }
 911   }
 912
 913   if (pool->is_tier()) {
 914     *ss << " pool '" << pool_name << "' (id '" << pool_id
 915       << "') is already in use as a cache tier.";
 916     return -EINVAL;
 917   }
 918
 919   if (!force && !pool->application_metadata.empty() &&
 920       pool->application_metadata.count(
 921         pg_pool_t::APPLICATION_NAME_CEPHFS) == 0) {
 922     *ss << " pool '" << pool_name << "' (id '" << pool_id
 923         << "') has a non-CephFS application enabled.";
 924     return -EINVAL;
 925   }
 926
 927   // Nothing special about this pool, so it is permissible
 928   return 0;
 929 }
 930