crimson/osd: check existing superblock when mkfs

in case mkfs on an existing store.

this change mirrors the behavior of classic osd, also addresses the
assert failure when BlueStore tries to create a collection when it
already contains a colloection with the same collection id.

Signed-off-by: Kefu Chai <kchai@redhat.com>

src/crimson/osd/osd.cc

@@ -169,17 +169,37 @@ seastar::future<> OSD::mkfs(uuid_d osd_uuid, uuid_d cluster_fsid)
 
 seastar::future<> OSD::_write_superblock()
 {
-  logger().info(
-    "{} writing superblock cluster_fsid {} osd_fsid {}",
-    __func__,
-    superblock.cluster_fsid,
-    superblock.osd_fsid);
-  return store->create_new_collection(coll_t::meta()).then([this] (auto ch) {
-    meta_coll = make_unique<OSDMeta>(ch , store.get());
-    ceph::os::Transaction t;
-    meta_coll->create(t);
-    meta_coll->store_superblock(t, superblock);
-    return store->do_transaction(meta_coll->collection(), std::move(t));
+  return store->open_collection(coll_t::meta()).then([this] (auto ch) {
+    if (ch) {
+      // if we already have superblock, check if it matches
+      meta_coll = make_unique<OSDMeta>(ch, store.get());
+      return meta_coll->load_superblock().then([this](OSDSuperblock&& sb) {
+        if (sb.cluster_fsid != superblock.cluster_fsid) {
+          logger().error("provided cluster fsid {} != superblock's {}",
+                         sb.cluster_fsid, superblock.cluster_fsid);
+          throw std::invalid_argument("mismatched fsid");
+        }
+        if (sb.whoami != superblock.whoami) {
+          logger().error("provided osd id {} != superblock's {}",
+                         sb.whoami, superblock.whoami);
+          throw std::invalid_argument("mismatched osd id");
+        }
+      });
+    } else {
+      // meta collection does not yet, create superblock
+      logger().info(
+        "{} writing superblock cluster_fsid {} osd_fsid {}",
+        __func__,
+        superblock.cluster_fsid,
+        superblock.osd_fsid);
+      return store->create_new_collection(coll_t::meta()).then([this] (auto ch) {
+        meta_coll = make_unique<OSDMeta>(ch , store.get());
+        ceph::os::Transaction t;
+        meta_coll->create(t);
+        meta_coll->store_superblock(t, superblock);
+        return store->do_transaction(meta_coll->collection(), std::move(t));
+      });
+    }
   });
 }