Merge PR #24603 into master

* refs/pull/24603/head:
	crush: get "ceph osd crush class create/rm" back

Reviewed-by: Sage Weil <sage@redhat.com>

qa/standalone/crush/crush-classes.sh

@@ -147,6 +147,17 @@ function TEST_mon_classes() {
     test "$(get_osds_up rbd SOMETHING)" == "1 2 0" || return 1
     add_something $dir SOMETHING || return 1
 
+    # test create and remove class
+    ceph osd crush class create CLASS || return 1
+    ceph osd crush class create CLASS || return 1 # idempotent
+    ceph osd crush class ls | grep CLASS  || return 1
+    ceph osd crush class rename CLASS TEMP || return 1
+    ceph osd crush class ls | grep TEMP || return 1
+    ceph osd crush class rename TEMP CLASS || return 1
+    ceph osd crush class ls | grep CLASS  || return 1
+    ceph osd crush class rm CLASS || return 1
+    expect_failure $dir ENOENT ceph osd crush class rm CLASS || return 1
+
     # test rm-device-class
     ceph osd crush set-device-class aaa osd.0 || return 1
     ceph osd tree | grep -q 'aaa' || return 1

src/crush/CrushWrapper.cc

@@ -1536,6 +1536,39 @@ void CrushWrapper::get_subtree_of_type(int type, vector<int> *subtrees)
   }
 }
 
+bool CrushWrapper::class_is_in_use(int class_id, ostream *ss)
+{
+  list<unsigned> rules;
+  for (unsigned i = 0; i < crush->max_rules; ++i) {
+    crush_rule *r = crush->rules[i];
+    if (!r)
+      continue;
+    for (unsigned j = 0; j < r->len; ++j) {
+      if (r->steps[j].op == CRUSH_RULE_TAKE) {
+        int root = r->steps[j].arg1;
+        for (auto &p : class_bucket) {
+          auto& q = p.second;
+          if (q.count(class_id) && q[class_id] == root) {
+            rules.push_back(i);
+          }
+        }
+      }
+    }
+  }
+  if (rules.empty()) {
+    return false;
+  }
+  if (ss) {
+    ostringstream os;
+    for (auto &p: rules) {
+      os << "'" << get_rule_name(p) <<"',";
+    }
+    string out(os.str());
+    out.resize(out.size() - 1); // drop last ','
+    *ss << "still referenced by crush_rule(s): " << out;
+  }
+  return true;
+}
 
 int CrushWrapper::rename_class(const string& srcname, const string& dstname)
 {

src/crush/CrushWrapper.h

@@ -1290,6 +1290,7 @@ public:
     const std::set<int32_t>& used_ids,
     int *clone,
     map<int,map<int,vector<int>>> *cmap_item_weight);
+  bool class_is_in_use(int class_id, ostream *ss = nullptr);
   int rename_class(const string& srcname, const string& dstname);
   int populate_classes(
     const std::map<int32_t, map<int32_t, int32_t>>& old_class_bucket);

src/mon/MonCommands.h

@@ -551,6 +551,14 @@ COMMAND("osd crush add " \
 COMMAND("osd crush set-all-straw-buckets-to-straw2",
         "convert all CRUSH current straw buckets to use the straw2 algorithm",
 	"osd", "rw", "cli,rest")
+COMMAND("osd crush class create " \
+        "name=class,type=CephString,goodchars=[A-Za-z0-9-_]", \
+        "create crush device class <class>", \
+        "osd", "rw", "cli,rest")
+COMMAND("osd crush class rm " \
+        "name=class,type=CephString,goodchars=[A-Za-z0-9-_]", \
+        "remove crush device class <class>", \
+        "osd", "rw", "cli,rest")
 COMMAND("osd crush set-device-class " \
         "name=class,type=CephString " \
 	"name=ids,type=CephString,n=N", \

src/mon/OSDMonitor.cc

@@ -8553,6 +8553,88 @@ bool OSDMonitor::prepare_command_impl(MonOpRequestRef op,
         new Monitor::C_Command(mon,op, 0, rs, get_last_committed() + 1));
       return true;
     }
+  } else if (prefix == "osd crush class create") {
+    string device_class;
+    if (!cmd_getval(g_ceph_context, cmdmap, "class", device_class)) {
+      err = -EINVAL; // no value!
+      goto reply;
+    }
+    if (osdmap.require_osd_release < CEPH_RELEASE_LUMINOUS) {
+      ss << "you must complete the upgrade and 'ceph osd require-osd-release "
+         << "luminous' before using crush device classes";
+      err = -EPERM;
+      goto reply;
+    }
+    if (!_have_pending_crush() &&
+        _get_stable_crush().class_exists(device_class)) {
+      ss << "class '" << device_class << "' already exists";
+      goto reply;
+    }
+     CrushWrapper newcrush;
+    _get_pending_crush(newcrush);
+     if (newcrush.class_exists(device_class)) {
+      ss << "class '" << device_class << "' already exists";
+      goto update;
+    }
+    int class_id = newcrush.get_or_create_class_id(device_class);
+    pending_inc.crush.clear();
+    newcrush.encode(pending_inc.crush, mon->get_quorum_con_features());
+    ss << "created class " << device_class << " with id " << class_id
+       << " to crush map";
+    goto update;
+  } else if (prefix == "osd crush class rm") {
+    string device_class;
+    if (!cmd_getval(g_ceph_context, cmdmap, "class", device_class)) {
+       err = -EINVAL; // no value!
+       goto reply;
+     }
+     if (osdmap.require_osd_release < CEPH_RELEASE_LUMINOUS) {
+       ss << "you must complete the upgrade and 'ceph osd require-osd-release "
+         << "luminous' before using crush device classes";
+       err = -EPERM;
+       goto reply;
+     }
+
+     CrushWrapper newcrush;
+     _get_pending_crush(newcrush);
+     if (!newcrush.class_exists(device_class)) {
+       err = -ENOENT;
+       ss << "class '" << device_class << "' does not exist";
+       goto reply;
+     }
+     int class_id = newcrush.get_class_id(device_class);
+     stringstream ts;
+     if (newcrush.class_is_in_use(class_id, &ts)) {
+       err = -EBUSY;
+       ss << "class '" << device_class << "' " << ts.str();
+       goto reply;
+     }
+
+     set<int> osds;
+     newcrush.get_devices_by_class(device_class, &osds);
+     for (auto& p: osds) {
+       err = newcrush.remove_device_class(g_ceph_context, p, &ss);
+       if (err < 0) {
+         // ss has reason for failure
+         goto reply;
+       }
+     }
+
+     if (osds.empty()) {
+       // empty class, remove directly
+       err = newcrush.remove_class_name(device_class);
+       if (err < 0) {
+         ss << "class '" << device_class << "' cannot be removed '"
+            << cpp_strerror(err) << "'";
+         goto reply;
+       }
+     }
+
+     pending_inc.crush.clear();
+     newcrush.encode(pending_inc.crush, mon->get_quorum_con_features());
+     ss << "removed class " << device_class << " with id " << class_id
+        << " from crush map";
+     goto update;
   } else if (prefix == "osd crush class rename") {
     string srcname, dstname;
     if (!cmd_getval(cct, cmdmap, "srcname", srcname)) {