mon/OSDMonitor: make destroy and purge idempotent

Signed-off-by: Joao Eduardo Luis <[email protected]>

src/mon/AuthMonitor.cc

@@ -792,6 +792,11 @@ int AuthMonitor::validate_osd_destroy(
     return -EINVAL;
   }
 
+  if (!mon->key_server.contains(cephx_entity) &&
+      !mon->key_server.contains(lockbox_entity)) {
+    return -ENOENT;
+  }
+
   return 0;
 }
 

src/mon/ConfigKeyService.cc

@@ -92,6 +92,22 @@ void ConfigKeyService::store_list(stringstream &ss)
   f.flush(ss);
 }
 
+bool ConfigKeyService::store_has_prefix(const string &prefix)
+{
+  KeyValueDB::Iterator iter =
+    mon->store->get_iterator(STORE_PREFIX);
+
+  while (iter->valid()) {
+    string key(iter->key());
+    size_t p = key.find(prefix);
+    if (p != string::npos && p == 0) {
+      return true;
+    }
+    iter->next();
+  }
+  return false;
+}
+
 void ConfigKeyService::store_dump(stringstream &ss)
 {
   KeyValueDB::Iterator iter =
@@ -252,6 +268,21 @@ string _get_dmcrypt_prefix(const uuid_d& uuid, const string k)
   return "dm-crypt/osd/" + stringify(uuid) + "/" + k;
 }
 
+int ConfigKeyService::validate_osd_destroy(
+    const int32_t id,
+    const uuid_d& uuid)
+{
+  string dmcrypt_prefix = _get_dmcrypt_prefix(uuid, "");
+  string daemon_prefix =
+    "daemon-private/osd." + stringify(id) + "/";
+
+  if (!store_has_prefix(dmcrypt_prefix) &&
+      !store_has_prefix(daemon_prefix)) {
+    return -ENOENT;
+  }
+  return 0;
+}
+
 void ConfigKeyService::do_osd_destroy(int32_t id, uuid_d& uuid)
 {
   string dmcrypt_prefix = _get_dmcrypt_prefix(uuid, "");

src/mon/ConfigKeyService.h

@@ -37,6 +37,7 @@ class ConfigKeyService : public QuorumService
   void store_list(stringstream &ss);
   void store_dump(stringstream &ss);
   bool store_exists(const string &key);
+  bool store_has_prefix(const string &prefix);
 
   static const string STORE_PREFIX;
 
@@ -66,6 +67,7 @@ class ConfigKeyService : public QuorumService
   void cleanup() override { }
   void service_tick() override { }
 
+  int validate_osd_destroy(const int32_t id, const uuid_d& uuid);
   void do_osd_destroy(int32_t id, uuid_d& uuid);
   int validate_osd_new(
       const uuid_d& uuid,

src/mon/OSDMonitor.cc

@@ -6913,29 +6913,62 @@ int OSDMonitor::prepare_command_osd_destroy(
     stringstream& ss)
 {
   assert(paxos->is_plugged());
+
+  // we check if the osd exists for the benefit of `osd purge`, which may
+  // have previously removed the osd. If the osd does not exist, return
+  // -ENOENT to convey this, and let the caller deal with it.
+  //
+  // we presume that all auth secrets and config keys were removed prior
+  // to this command being called. if they exist by now, we also assume
+  // they must have been created by some other command and do not pertain
+  // to this non-existent osd.
+  if (!osdmap.exists(id)) {
+    dout(10) << __func__ << " osd." << id << " does not exist." << dendl;
+    return -ENOENT;
+  }
+
   uuid_d uuid = osdmap.get_uuid(id);
   dout(10) << __func__ << " destroying osd." << id
            << " uuid " << uuid << dendl;
 
+  // if it has been destroyed, we assume our work here is done.
   if (osdmap.is_destroyed(id)) {
     ss << "destroyed osd." << id;
     return 0;
   }
 
   EntityName cephx_entity, lockbox_entity;
+  bool idempotent_auth = false, idempotent_cks = false;
 
   int err = mon->authmon()->validate_osd_destroy(id, uuid,
                                                  cephx_entity,
                                                  lockbox_entity,
                                                  ss);
   if (err < 0) {
-    return err;
+    if (err == -ENOENT) {
+      idempotent_auth = true;
+      err = 0;
+    } else {
+      return err;
+    }
   }
 
-  err = mon->authmon()->do_osd_destroy(cephx_entity, lockbox_entity);
-  assert(0 == err);
+  ConfigKeyService *svc = (ConfigKeyService*)mon->config_key_service;
+  err = svc->validate_osd_destroy(id, uuid);
+  if (err < 0) {
+    assert(err == -ENOENT);
+    err = 0;
+    idempotent_cks = true;
+  }
+
+  if (!idempotent_auth) {
+    err = mon->authmon()->do_osd_destroy(cephx_entity, lockbox_entity);
+    assert(0 == err);
+  }
 
-  ((ConfigKeyService*)mon->config_key_service)->do_osd_destroy(id, uuid);
+  if (!idempotent_cks) {
+    svc->do_osd_destroy(id, uuid);
+  }
 
   pending_inc.new_state[id] = CEPH_OSD_DESTROYED;
   pending_inc.new_uuid[id] = uuid_d();
@@ -6977,9 +7010,12 @@ int OSDMonitor::prepare_command_osd_purge(
   CrushWrapper newcrush;
   _get_pending_crush(newcrush);
 
+  bool may_be_idempotent = false;
+
   int err = _prepare_command_osd_crush_remove(newcrush, id, 0, false, false);
   if (err == -ENOENT) {
     err = 0;
+    may_be_idempotent = true;
   } else if (err < 0) {
     ss << "error removing osd." << id << " from crush";
     return err;
@@ -6989,11 +7025,23 @@ int OSDMonitor::prepare_command_osd_purge(
   if (!osdmap.is_destroyed(id)) {
     err = prepare_command_osd_destroy(id, ss);
     if (err < 0) {
-      return err;
+      if (err == -ENOENT) {
+        err = 0;
+      } else {
+        return err;
+      }
+    } else {
+      may_be_idempotent = false;
     }
     assert(0 == err);
   }
 
+  if (may_be_idempotent && !osdmap.exists(id)) {
+    dout(10) << __func__ << " osd." << id << " does not exist and "
+             << "we are idempotent." << dendl;
+    return -ENOENT;
+  }
+
   err = prepare_command_osd_remove(id);
   // we should not be busy, as we should have made sure this id is not up.
   assert(0 == err);
@@ -8859,7 +8907,7 @@ bool OSDMonitor::prepare_command_impl(MonOpRequestRef op,
          << "really do.";
       err = -EPERM;
       goto reply;
-    } else if (!osdmap.exists(id)) {
+    } else if (is_destroy && !osdmap.exists(id)) {
       ss << "osd." << id << " does not exist";
       err = -ENOENT;
       goto reply;
@@ -8873,15 +8921,24 @@ bool OSDMonitor::prepare_command_impl(MonOpRequestRef op,
       goto reply;
     }
 
+    bool goto_reply = false;
+
     paxos->plug();
     if (is_destroy) {
       err = prepare_command_osd_destroy(id, ss);
+      // we checked above that it should exist.
+      assert(err != -ENOENT);
     } else {
       err = prepare_command_osd_purge(id, ss);
+      if (err == -ENOENT) {
+        err = 0;
+        ss << "osd." << id << " does not exist.";
+        goto_reply = true;
+      }
     }
     paxos->unplug();
 
-    if (err < 0) {
+    if (err < 0 || goto_reply) {
       goto reply;
     }
 
@@ -8890,6 +8947,7 @@ bool OSDMonitor::prepare_command_impl(MonOpRequestRef op,
     } else {
       ss << "purged osd." << id;
     }
+
     getline(ss, rs);
     wait_for_finished_proposal(op,
         new Monitor::C_Command(mon, op, 0, rs, get_last_committed() + 1));
@@ -8924,10 +8982,6 @@ bool OSDMonitor::prepare_command_impl(MonOpRequestRef op,
 
     if (err < 0) {
       goto reply;
-    } else if (err == EEXIST) {
-      // idempotent operation
-      err = 0;
-      goto reply;
     }
 
     if (f) {
@@ -8936,6 +8990,12 @@ bool OSDMonitor::prepare_command_impl(MonOpRequestRef op,
       rdata.append(ss);
     }
 
+    if (err == EEXIST) {
+      // idempotent operation
+      err = 0;
+      goto reply;
+    }
+
     wait_for_finished_proposal(op,
         new Monitor::C_Command(mon, op, 0, rs, rdata,
                                get_last_committed() + 1));