Merge PR ceph#56429 into main

* refs/pull/56429/head:
	mds: fix rank root doesn't insert root ino into its subtree map when starting
	mds: flush mds log before finishing STATE_STARTING
	mds/FSMap: go back to STARTING state when rank doesn't make it pass STARTING

Reviewed-by: Venky Shankar <[email protected]>
Reviewed-by: Patrick Donnelly <[email protected]>

Author: vshankar

src/mds/FSMap.cc

@@ -1011,6 +1011,12 @@ void FSMap::erase(mds_gid_t who, epoch_t blocklist_epoch)
         // the rank ever existed so that next time it's handed out
         // to a gid it'll go back into CREATING.
         fs.mds_map.in.erase(info.rank);
+      } else if (info.state == MDSMap::STATE_STARTING) {
+        // If this gid didn't make it past STARTING, then forget
+        // the rank ever existed so that next time it's handed out
+        // to a gid it'll go back into STARTING.
+        fs.mds_map.in.erase(info.rank);
+        fs.mds_map.stopped.insert(info.rank);
       } else {
         // Put this rank into the failed list so that the next available
         // STANDBY will pick it up.

src/mds/MDCache.cc

@@ -640,7 +640,24 @@ void MDCache::open_root_inode(MDSContext *c)
   if (mds->get_nodeid() == mds->mdsmap->get_root()) {
     CInode *in;
     in = create_system_inode(CEPH_INO_ROOT, S_IFDIR|0755);  // initially inaccurate!
-    in->fetch(c);
+    if (mds->is_starting()) {
+      in->fetch(
+          new MDSInternalContextWrapper(mds,
+            new LambdaContext([this, c](int r) {
+              if (r < 0) {
+                c->complete(r);
+                return;
+              }
+              CDir *rootdir = root->get_or_open_dirfrag(this, frag_t());
+              ceph_assert(rootdir);
+              adjust_subtree_auth(rootdir, mds->get_nodeid());
+              rootdir->fetch(c);
+            })
+          )
+        );
+    } else {
+      in->fetch(c);
+    }
   } else {
     discover_base_ino(CEPH_INO_ROOT, c, mds->mdsmap->get_root());
   }

src/mds/MDSRank.cc

@@ -1724,7 +1724,10 @@ void MDSRank::boot_start(BootStep step, int r)
       } else {
         dout(2) << "Booting: " << step << ": positioning at end of old mds log" << dendl;
         mdlog->append();
-        starting_done();
+        auto sle = mdcache->create_subtree_map();
+        mdlog->submit_entry(sle);
+        mdlog->flush();
+        mdlog->wait_for_safe(new C_MDS_VoidFn(this, &MDSRank::starting_done));
       }
       break;
     case MDS_BOOT_REPLAY_DONE:
@@ -1771,9 +1774,6 @@ void MDSRank::starting_done()
   ceph_assert(is_starting());
   request_state(MDSMap::STATE_ACTIVE);
 
-  auto sle = mdcache->create_subtree_map();
-  mdlog->submit_entry(sle);
-
   // sync snaptable cache
   snapclient->sync(new C_MDSInternalNoop);
 }