mon: 'ceph osd crush link ...' to add a link to an existing bucket

Allow a second reference to an existing bucket to be added.  This lets
you create a DAG instead of a tree using the CLI.

Signed-off-by: Sage Weil <sage@inktank.com>

diff --git a/src/mon/OSDMonitor.cc b/src/mon/OSDMonitor.cc
index 47c6c1d8f69223805a8fc6b19c05bab54f883ebc..4f90345ec37cb6fbc6bc010d7ed93d19509ca00e 100644 (file)
--- a/src/mon/OSDMonitor.cc
+++ b/src/mon/OSDMonitor.cc
@@ -2409,6 +2409,46 @@ bool OSDMonitor::prepare_command(MMonCommand *m)
        }
       } while (false);
     }
+    else if (m->cmd.size() >= 5 && m->cmd[1] == "crush" && m->cmd[2] == "link") {
+      do {
+       // osd crush link <name> <loc1> [<loc2> ...]
+       string name = m->cmd[3];
+       map<string,string> loc;
+       parse_loc_map(m->cmd, 4, &loc);
+
+       dout(0) << "linking crush item name '" << name << "' at location " << loc << dendl;
+       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.name_exists(name.c_str())) {
+         err = -ENOENT;
+         ss << "item " << name << " does not exist";
+         break;
+       }
+       int id = newcrush.get_item_id(name.c_str());
+
+       if (!newcrush.check_item_loc(g_ceph_context, id, loc, (int *)NULL)) {
+         err = newcrush.link_bucket(g_ceph_context, id, loc);
+         if (err >= 0) {
+           ss << "linked item id " << id << " name '" << name << "' to location " << loc << " in crush map";
+           pending_inc.crush.clear();
+           newcrush.encode(pending_inc.crush);
+           getline(ss, rs);
+           wait_for_finished_proposal(new Monitor::C_Command(mon, m, 0, rs, get_version()));
+           return true;
+         }
+       } else {
+           ss << "no need to move item id " << id << " name '" << name << "' to location " << loc << " in crush map";
+       }
+      } while (false);
+    }
     else if (m->cmd.size() > 3 && m->cmd[1] == "crush" && (m->cmd[2] == "rm" || m->cmd[2] == "remove")) {
       do {
        // osd crush rm <id>

diff --git a/src/test/cli/ceph/help.t b/src/test/cli/ceph/help.t
index ca93dba39c43288df3ec75fcaefffcf9da4a871a..123a07464ebc18f12e190bee1e7d2b38026981f3 100644 (file)
--- a/src/test/cli/ceph/help.t
+++ b/src/test/cli/ceph/help.t
@@ -46,8 +46,9 @@
     ceph osd getmap -o <file>
     ceph osd crush set <osd-id> <weight> <loc1> [<loc2> ...]
     ceph osd crush add <osd-id> <weight> <loc1> [<loc2> ...]
-    ceph osd crush move <bucketname> <loc1> [<loc2> ...]
     ceph osd crush create-or-move <osd-id> <initial-weight> <loc1> [<loc2> ...]
+    ceph osd crush move <bucketname> <loc1> [<loc2> ...]
+    ceph osd crush link <bucketname> <loc1> [<loc2> ...]
     ceph osd crush reweight <name> <weight>
     ceph osd crush tunables <legacy|argonaut|bobtail|optimal|default>
     ceph osd create [<uuid>]

diff --git a/src/tools/ceph.cc b/src/tools/ceph.cc
index f93d445d61224f4a9386c4509376b0803d51f1a3..1dcec25463211120f14e01a5012fa5750995bb54 100644 (file)
--- a/src/tools/ceph.cc
+++ b/src/tools/ceph.cc
@@ -89,8 +89,9 @@ static void usage()
   cout << "  ceph osd getmap -o <file>\n";
   cout << "  ceph osd crush set <osd-id> <weight> <loc1> [<loc2> ...]\n";
   cout << "  ceph osd crush add <osd-id> <weight> <loc1> [<loc2> ...]\n";
-  cout << "  ceph osd crush move <bucketname> <loc1> [<loc2> ...]\n";
   cout << "  ceph osd crush create-or-move <osd-id> <initial-weight> <loc1> [<loc2> ...]\n";
+  cout << "  ceph osd crush move <bucketname> <loc1> [<loc2> ...]\n";
+  cout << "  ceph osd crush link <bucketname> <loc1> [<loc2> ...]\n";
   cout << "  ceph osd crush reweight <name> <weight>\n";
   cout << "  ceph osd crush tunables <legacy|argonaut|bobtail|optimal|default>\n";
   cout << "  ceph osd create [<uuid>]\n";