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