mon: OSDMonitor: don't expose uncommitted state on 'osd crush link'

Fixes: #4635
Signed-off-by: Sage Weil <sage@inktank.com>
Reviewed-by: Joao Eduardo Luis <joao.luis@inktank.com>

diff --git a/src/mon/OSDMonitor.cc b/src/mon/OSDMonitor.cc
index 7b9258b17309e9d0b7df1c9a67eaa5de9c6ffc70..c2725c0c032f1e605fc1f904399fd70376195bed 100644 (file)
--- a/src/mon/OSDMonitor.cc
+++ b/src/mon/OSDMonitor.cc
@@ -2819,41 +2819,51 @@ bool OSDMonitor::prepare_command(MMonCommand *m)
     } while (false);
 
   } else if (prefix == "osd crush link") {
-    do {
-      // osd crush link <name> <loc1> [<loc2> ...]
-      string name;
-      cmd_getval(g_ceph_context, cmdmap, "name", name);
-      vector<string> argvec;
-      cmd_getval(g_ceph_context, cmdmap, "args", argvec);
-      map<string,string> loc;
-      parse_loc_map(argvec, &loc);
+    // osd crush link <name> <loc1> [<loc2> ...]
+    string name;
+    cmd_getval(g_ceph_context, cmdmap, "name", name);
+    vector<string> argvec;
+    cmd_getval(g_ceph_context, cmdmap, "args", argvec);
+    map<string,string> loc;
+    parse_loc_map(argvec, &loc);
 
-      dout(0) << "linking crush item name '" << name << "' at location " << loc << dendl;
-      CrushWrapper newcrush;
-      _get_pending_crush(newcrush);
+    if (!osdmap.crush->name_exists(name)) {
+      err = -ENOENT;
+      ss << "item " << name << " does not exist";
+      goto reply;
+    }
+    if (osdmap.crush->check_item_loc(g_ceph_context, id, loc, (int*) NULL)) {
+      ss << "no need to move item id " << id << " name '" << name
+        << "' to location " << loc << " in crush map";
+      err = 0;
+      goto reply;
+    }
 
-      if (!newcrush.name_exists(name)) {
-       err = -ENOENT;
-       ss << "item " << name << " does not exist";
-       break;
-      }
-      int id = newcrush.get_item_id(name);
+    dout(5) << "linking crush item name '" << name << "' at location " << loc << dendl;
+    CrushWrapper newcrush;
+    _get_pending_crush(newcrush);
 
+    if (!newcrush.name_exists(name)) {
+      err = -ENOENT;
+      ss << "item " << name << " does not exist";
+    } else {
+      int id = newcrush.get_item_id(name);
       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";
+         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_last_committed()));
-         return true;
        }
       } else {
-       ss << "no need to move item id " << id << " name '" << name << "' to location " << loc << " in crush map";
+       ss << "no need to move item id " << id << " name '" << name
+           << "' to location " << loc << " in crush map";
        err = 0;
       }
-    } while (false);
+    }
+    wait_for_finished_proposal(new Monitor::C_Command(mon, m, err, ss.str(), get_last_committed()));
+    return true;
   } else if (prefix == "osd crush rm" ||
             prefix == "osd crush remove" ||
             prefix == "osd crush unlink") {