Merge remote-tracking branch 'gh/wip-osd-commands'

Reviewed-by: Dan Mick <dan.mick@inktank.com>
Reviewed-by: Joao Luis <joao.luis@inktank.com>

8 files changed, 275 insertions, 5 deletions

diff --git a/qa/workunits/mon/crush_ops.sh b/qa/workunits/mon/crush_ops.sh
new file mode 100755
index 00000000000..735646b5ca0
--- /dev/null
+++ b/qa/workunits/mon/crush_ops.sh
@@ -0,0 +1,23 @@
+#!/bin/sh -x
+
+set -e
+
+ceph osd crush dump
+ceph osd crush rule dump
+ceph osd crush rule ls
+ceph osd crush rule list
+
+ceph osd crush rule create-simple foo default host
+ceph osd crush rule create-simple foo default host
+ceph osd crush rule create-simple bar default host
+
+ceph osd crush rule ls | grep foo
+
+ceph osd crush rule rm foo
+ceph osd crush rule rm foo  # idempotent
+ceph osd crush rule rm bar
+
+# can't delete in-use rules, tho:
+ceph osd crush rule rm data && exit 1 || true
+
+echo OK
diff --git a/src/crush/CrushWrapper.cc b/src/crush/CrushWrapper.cc
index 45e4fb53de6..a22f23509c9 100644
--- a/src/crush/CrushWrapper.cc
+++ b/src/crush/CrushWrapper.cc
@@ -488,6 +488,61 @@ void CrushWrapper::reweight(CephContext *cct)
   }
 }
 
+int CrushWrapper::add_simple_rule(string name, string root_name, string failure_domain_name)
+{
+  if (rule_exists(name))
+    return -EEXIST;
+  if (!name_exists(root_name.c_str()))
+    return -ENOENT;
+  int root = get_item_id(root_name.c_str());
+  int type = 0;
+  if (failure_domain_name.length()) {
+    type = get_type_id(failure_domain_name.c_str());
+    if (type <= 0) // bah, returns 0 on error; but its ok, device isn't a domain really
+      return -EINVAL;
+  }
+
+  int ruleset = 0;
+  for (int i = 0; i < get_max_rules(); i++) {
+    if (rule_exists(i) &&
+	get_rule_mask_ruleset(i) >= ruleset) {
+      ruleset = get_rule_mask_ruleset(i) + 1;
+    }
+  }
+
+  crush_rule *rule = crush_make_rule(3, ruleset, 1 /* pg_pool_t::TYPE_REP */, 1, 10);
+  assert(rule);
+  crush_rule_set_step(rule, 0, CRUSH_RULE_TAKE, root, 0);
+  if (type)
+    crush_rule_set_step(rule, 1,
+			CRUSH_RULE_CHOOSE_LEAF_FIRSTN,
+			CRUSH_CHOOSE_N,
+			type);
+  else
+    crush_rule_set_step(rule, 1,
+			CRUSH_RULE_CHOOSE_FIRSTN,
+			CRUSH_CHOOSE_N,
+			0);
+  crush_rule_set_step(rule, 2, CRUSH_RULE_EMIT, 0, 0);
+  int rno = crush_add_rule(crush, rule, -1);
+  set_rule_name(rno, name.c_str());
+  have_rmaps = false;
+  return rno;
+}
+
+int CrushWrapper::remove_rule(int ruleno)
+{
+  if (ruleno >= (int)crush->max_rules)
+    return -ENOENT;
+  if (crush->rules[ruleno] == NULL)
+    return -ENOENT;
+  crush_destroy_rule(crush->rules[ruleno]);
+  crush->rules[ruleno] = NULL;
+  rule_name_map.erase(ruleno);
+  have_rmaps = false;
+  return 0;
+}
+
 void CrushWrapper::encode(bufferlist& bl, bool lean) const
 {
   assert(crush);
@@ -817,6 +872,12 @@ void CrushWrapper::dump(Formatter *f) const
   f->close_section();
 
   f->open_array_section("rules");
+  dump_rules(f);
+  f->close_section();
+}
+
+void CrushWrapper::dump_rules(Formatter *f) const
+{
   for (int i=0; i<get_max_rules(); i++) {
     if (!rule_exists(i))
       continue;
@@ -872,7 +933,15 @@ void CrushWrapper::dump(Formatter *f) const
     f->close_section();
     f->close_section();
   }
-  f->close_section();
+}
+
+void CrushWrapper::list_rules(Formatter *f) const
+{
+  for (int rule = 0; rule < get_max_rules(); rule++) {
+    if (!rule_exists(rule))
+      continue;
+    f->dump_string("name", get_rule_name(rule));
+  }
 }
 
 void CrushWrapper::generate_test_instances(list<CrushWrapper*>& o)
diff --git a/src/crush/CrushWrapper.h b/src/crush/CrushWrapper.h
index 7def6e4ab34..0b919cba3ec 100644
--- a/src/crush/CrushWrapper.h
+++ b/src/crush/CrushWrapper.h
@@ -221,12 +221,15 @@ public:
   }
 
   // rule names
-  int get_rule_id(const char *n) {
-    string name(n);
+  bool rule_exists(string name) {
+    build_rmaps();
+    return rule_name_rmap.count(name);
+  }
+  int get_rule_id(string name) {
     build_rmaps();
     if (rule_name_rmap.count(name))
       return rule_name_rmap[name];
-    return 0;  /* hrm */
+    return -ENOENT;
   }
   const char *get_rule_name(int t) const {
     std::map<int,string>::const_iterator p = rule_name_map.find(t);
@@ -527,6 +530,9 @@ public:
     return set_rule_step(ruleno, step, CRUSH_RULE_EMIT, 0, 0);
   }
 
+  int add_simple_rule(string name, string root_name, string failure_domain_type);
+
+  int remove_rule(int ruleno);
 
 
   /** buckets **/
@@ -735,6 +741,8 @@ public:
   void decode(bufferlist::iterator &blp);
   void decode_crush_bucket(crush_bucket** bptr, bufferlist::iterator &blp);
   void dump(Formatter *f) const;
+  void dump_rules(Formatter *f) const;
+  void list_rules(Formatter *f) const;
   static void generate_test_instances(list<CrushWrapper*>& o);
 };
 WRITE_CLASS_ENCODER(CrushWrapper)
diff --git a/src/crush/crush.c b/src/crush/crush.c
index 19a765228e9..1e83eb866bb 100644
--- a/src/crush/crush.c
+++ b/src/crush/crush.c
@@ -116,7 +116,7 @@ void crush_destroy(struct crush_map *map)
 	if (map->rules) {
 		__u32 b;
 		for (b = 0; b < map->max_rules; b++)
-			kfree(map->rules[b]);
+			crush_destroy_rule(map->rules[b]);
 		kfree(map->rules);
 	}
 
@@ -124,6 +124,11 @@ void crush_destroy(struct crush_map *map)
 	kfree(map);
 }
 
+void crush_destroy_rule(struct crush_rule *rule)
+{
+	kfree(rule);
+}
+
 // methods to check for safe arithmetic operations
 int crush_addition_is_unsafe(__u32 a, __u32 b)
 {
diff --git a/src/crush/crush.h b/src/crush/crush.h
index 9fd37e9e516..82d032879d9 100644
--- a/src/crush/crush.h
+++ b/src/crush/crush.h
@@ -185,6 +185,7 @@ extern void crush_destroy_bucket_list(struct crush_bucket_list *b);
 extern void crush_destroy_bucket_tree(struct crush_bucket_tree *b);
 extern void crush_destroy_bucket_straw(struct crush_bucket_straw *b);
 extern void crush_destroy_bucket(struct crush_bucket *b);
+extern void crush_destroy_rule(struct crush_rule *r);
 extern void crush_destroy(struct crush_map *map);
 
 static inline int crush_calc_tree_node(int i)
diff --git a/src/mon/OSDMonitor.cc b/src/mon/OSDMonitor.cc
index 507eed74c42..209bb982d86 100644
--- a/src/mon/OSDMonitor.cc
+++ b/src/mon/OSDMonitor.cc
@@ -1880,6 +1880,38 @@ bool OSDMonitor::preprocess_command(MMonCommand *m)
 	}
       }
     }
+    else if (m->cmd[1] == "find") {
+      if (m->cmd.size() < 3) {
+	ss << "usage: osd find <osd-id>";
+	r = -EINVAL;
+	goto out;
+      }
+      long osd = parse_osd_id(m->cmd[2].c_str(), &ss);
+      if (osd < 0) {
+	r = -EINVAL;
+	goto out;
+      }
+      if (!osdmap.exists(osd)) {
+	ss << "osd." << osd << " does not exist";
+	r = -ENOENT;
+	goto out;
+      }
+      JSONFormatter jf(true);
+      jf.open_object_section("osd_location");
+      jf.dump_int("osd", osd);
+      jf.dump_stream("ip") << osdmap.get_addr(osd);
+      jf.open_object_section("crush_location");
+      map<string,string> loc = osdmap.crush->get_full_location(osd);
+      for (map<string,string>::iterator p = loc.begin(); p != loc.end(); ++p)
+	jf.dump_string(p->first.c_str(), p->second);
+      jf.close_section();
+      jf.close_section();
+      ostringstream rs;
+      jf.flush(rs);
+      rs << "\n";
+      rdata.append(rs.str());
+      r = 0;
+    }
     else if (m->cmd[1] == "map" && m->cmd.size() == 4) {
       int64_t pool = osdmap.lookup_pg_pool_name(m->cmd[2].c_str());
       if (pool < 0) {
@@ -1964,6 +1996,40 @@ bool OSDMonitor::preprocess_command(MMonCommand *m)
       ss << "listed " << osdmap.blacklist.size() << " entries";
       r = 0;
     }
+    else if (m->cmd.size() >= 4 && m->cmd[1] == "crush" && m->cmd[2] == "rule" && (m->cmd[3] == "list" ||
+										   m->cmd[3] == "ls")) {
+      JSONFormatter jf(true);
+      jf.open_array_section("rules");
+      osdmap.crush->list_rules(&jf);
+      jf.close_section();
+      ostringstream rs;
+      jf.flush(rs);
+      rs << "\n";
+      rdata.append(rs.str());
+      r = 0;
+    }
+    else if (m->cmd.size() >= 4 && m->cmd[1] == "crush" && m->cmd[2] == "rule" && m->cmd[3] == "dump") {
+      JSONFormatter jf(true);
+      jf.open_array_section("rules");
+      osdmap.crush->dump_rules(&jf);
+      jf.close_section();
+      ostringstream rs;
+      jf.flush(rs);
+      rs << "\n";
+      rdata.append(rs.str());
+      r = 0;
+    }
+    else if (m->cmd.size() == 3 && m->cmd[1] == "crush" && m->cmd[2] == "dump") {
+      JSONFormatter jf(true);
+      jf.open_object_section("crush_map");
+      osdmap.crush->dump(&jf);
+      jf.close_section();
+      ostringstream rs;
+      jf.flush(rs);
+      rs << "\n";
+      rdata.append(rs.str());
+      r = 0;
+    }
   }
  out:
   if (r != -1) {
@@ -2380,6 +2446,94 @@ bool OSDMonitor::prepare_command(MMonCommand *m)
 	return true;
       }
     }
+    else if (m->cmd.size() == 7 &&
+	     m->cmd[1] == "crush" &&
+	     m->cmd[2] == "rule" &&
+	     m->cmd[3] == "create-simple") {
+      string name = m->cmd[4];
+      string root = m->cmd[5];
+      string type = m->cmd[6];
+
+      if (osdmap.crush->rule_exists(name)) {
+	ss << "rule " << name << " already exists";
+	err = 0;
+	goto out;
+      }
+
+      bufferlist bl;
+      if (pending_inc.crush.length())
+	bl = pending_inc.crush;
+      else
+	osdmap.crush->encode(bl);
+      CrushWrapper newcrush;
+      bufferlist::iterator p = bl.begin();
+      newcrush.decode(p);
+
+      if (newcrush.rule_exists(name)) {
+	ss << "rule " << name << " already exists";
+      } else {
+	int rule = newcrush.add_simple_rule(name, root, type);
+	if (rule < 0) {
+	  err = rule;
+	  goto out;
+	}
+
+	pending_inc.crush.clear();
+	newcrush.encode(pending_inc.crush);
+      }
+      getline(ss, rs);
+      paxos->wait_for_commit(new Monitor::C_Command(mon, m, 0, rs, paxos->get_version()));
+      return true;
+    }
+    else if (m->cmd.size() == 5 &&
+	     m->cmd[1] == "crush" &&
+	     m->cmd[2] == "rule" &&
+	     m->cmd[3] == "rm") {
+      string name = m->cmd[4];
+
+      if (!osdmap.crush->rule_exists(name)) {
+	ss << "rule " << name << " does not exist";
+	err = 0;
+	goto out;
+      }
+
+      bufferlist bl;
+      if (pending_inc.crush.length())
+	bl = pending_inc.crush;
+      else
+	osdmap.crush->encode(bl);
+      CrushWrapper newcrush;
+      bufferlist::iterator p = bl.begin();
+      newcrush.decode(p);
+
+      if (!newcrush.rule_exists(name)) {
+	ss << "rule " << name << " does not exist";
+      } else {
+	int ruleno = newcrush.get_rule_id(name);
+	assert(ruleno >= 0);
+
+	// make sure it is not in use.
+	// FIXME: this is ok in some situations, but let's not bother with that
+	// complexity now.
+	int ruleset = newcrush.get_rule_mask_ruleset(ruleno);
+	if (osdmap.crush_ruleset_in_use(ruleset)) {
+	  ss << "crush rule " << name << " ruleset " << ruleset << " is in use";
+	  err = -EBUSY;
+	  goto out;
+	}
+
+	err = newcrush.remove_rule(ruleno);
+	if (err < 0) {
+	  goto out;
+	}
+
+	pending_inc.crush.clear();
+	newcrush.encode(pending_inc.crush);
+      }
+      getline(ss, rs);
+      paxos->wait_for_commit(new Monitor::C_Command(mon, m, 0, rs, paxos->get_version()));
+      return true;
+    }
     else if (m->cmd[1] == "setmaxosd" && m->cmd.size() > 2) {
       int newmax = parse_pos_long(m->cmd[2].c_str(), &ss);
       if (newmax < 0) {
diff --git a/src/osd/OSDMap.cc b/src/osd/OSDMap.cc
index c7d044ac6fd..6b692d407a8 100644
--- a/src/osd/OSDMap.cc
+++ b/src/osd/OSDMap.cc
@@ -1654,6 +1654,15 @@ void OSDMap::print_summary(ostream& out) const
     out << " nearfull";
 }
 
+bool OSDMap::crush_ruleset_in_use(int ruleset) const
+{
+  for (map<int64_t,pg_pool_t>::const_iterator p = pools.begin(); p != pools.end(); ++p) {
+    if (p->second.crush_ruleset == ruleset)
+      return true;
+  }
+  return false;
+}
+
 void OSDMap::build_simple(CephContext *cct, epoch_t e, uuid_d &fsid,
 			  int nosd, int pg_bits, int pgp_bits)
 {
diff --git a/src/osd/OSDMap.h b/src/osd/OSDMap.h
index d161fa7436b..70ec263e4d8 100644
--- a/src/osd/OSDMap.h
+++ b/src/osd/OSDMap.h
@@ -553,6 +553,7 @@ public:
   static void build_simple_crush_map_from_conf(CephContext *cct, CrushWrapper& crush,
 					       map<int, const char*>& rulesets);
 
+  bool crush_ruleset_in_use(int ruleset) const;
 
 private:
   void print_osd_line(int cur, ostream *out, Formatter *f) const;