From b91490d35319b6e215cc9fc82e7ca5af55e2e309 Mon Sep 17 00:00:00 2001
From: Patrick Donnelly <pdonnell@redhat.com>
Date: Thu, 14 Jan 2021 19:55:43 -0800
Subject: [PATCH] mds: fix alternate_name durability

This is a collection of fixes to Xiubo's prior work. Namely:

- Add new mds_alternate_name_max option to limit the size of
  alternate_name. Otherwise a Client could trick the MDS into creating
  an alternate_name of any size!

- Clean up how alternate_name is assigned to CDentry. In the general
  case, this should be assigned as part of creating the dentry. We want
  this value to be immutable for the life of the dentry. Even for the
  very special case of rename(2) where the destination dentry already
  exists.  We explicitly check (after discussion with Jeff) that the
  target dentry alternate_name already matches what the rename RPC is
  giving.

- The MDS is now properly journaling the alternate_name.

- The MDS rejoin phase is properly transmitting each dentry's
  alternate_name. I've discovered that this MMDSCacheRejoin message
  actually wasn't versioned which I've raised in a tracker [1]. In the
  mean time, we'll just bump CEPH_MDS_PROTOCOL as usual.

[1] https://tracker.ceph.com/issues/48886

Signed-off-by: Patrick Donnelly <pdonnell@redhat.com>
---
 src/common/options.cc          |  5 ++
 src/mds/CDentry.cc             | 18 ++++----
 src/mds/CDentry.h              | 25 +++++++---
 src/mds/CDir.cc                | 45 +++++++++---------
 src/mds/CDir.h                 |  3 +-
 src/mds/Locker.cc              |  4 +-
 src/mds/MDCache.cc             | 34 +++++++++-----
 src/mds/MDSDaemon.h            |  2 +-
 src/mds/MDSRank.cc             |  1 +
 src/mds/Migrator.cc            | 13 +++---
 src/mds/Server.cc              | 83 +++++++++++++++++++++++++++-------
 src/mds/Server.h               |  5 +-
 src/mds/events/EMetaBlob.h     | 46 +++++++++++--------
 src/mds/journal.cc             | 26 ++++++++---
 src/messages/MClientReply.h    | 15 ++++--
 src/messages/MClientRequest.h  |  2 +
 src/messages/MMDSCacheRejoin.h | 29 ++++++------
 src/messages/MMDSPeerRequest.h |  3 ++
 18 files changed, 238 insertions(+), 121 deletions(-)

diff --git a/src/common/options.cc b/src/common/options.cc
index 71133bdfb81d7..d0a2104c45485 100644
--- a/src/common/options.cc
+++ b/src/common/options.cc
@@ -7953,6 +7953,11 @@ static std::vector<Option> get_immutable_object_cache_options() {
 
 std::vector<Option> get_mds_options() {
   return std::vector<Option>({
+    Option("mds_alternate_name_max", Option::TYPE_SIZE, Option::LEVEL_ADVANCED)
+    .set_default(8192)
+    .set_flag(Option::FLAG_RUNTIME)
+    .set_description("set the maximum length of alternate names for dentries"),
+
     Option("mds_numa_node", Option::TYPE_INT, Option::LEVEL_ADVANCED)
     .set_default(-1)
     .set_flag(Option::FLAG_STARTUP)
diff --git a/src/mds/CDentry.cc b/src/mds/CDentry.cc
index 9831b3dc9a488..7d8170f73167a 100644
--- a/src/mds/CDentry.cc
+++ b/src/mds/CDentry.cc
@@ -109,6 +109,10 @@ ostream& operator<<(ostream& out, const CDentry& dn)
     dn.print_pin_set(out);
   }
 
+  if (dn.get_alternate_name().size()) {
+    out << " altname=" << binstrprint(dn.get_alternate_name(), 16);
+  }
+
   out << " " << &dn;
   out << "]";
   return out;
@@ -235,12 +239,6 @@ void CDentry::make_path(filepath& fp, bool projected) const
   fp.push_dentry(get_name());
 }
 
-void CDentry::set_alternate_name(std::string_view _alternate_name)
-{
-  dout(10) << "setting alternate_name on " << *this << dendl;
-  alternate_name = _alternate_name;
-}
-
 /*
  * we only add ourselves to remote_parents when the linkage is
  * active (no longer projected).  if the passed dnl is projected,
@@ -317,9 +315,11 @@ CDentry::linkage_t *CDentry::pop_projected_linkage()
       linkage.inode = n.inode;
       linkage.inode->add_remote_parent(this);
     }
-  } else if (n.inode) {
-    dir->link_primary_inode(this, n.inode);
-    n.inode->pop_projected_parent();
+  } else {
+    if (n.inode) {
+      dir->link_primary_inode(this, n.inode);
+      n.inode->pop_projected_parent();
+    }
   }
 
   ceph_assert(n.inode == linkage.inode);
diff --git a/src/mds/CDentry.h b/src/mds/CDentry.h
index 5b60b838155cd..dff250bcac332 100644
--- a/src/mds/CDentry.h
+++ b/src/mds/CDentry.h
@@ -69,7 +69,7 @@ public:
     unsigned char get_remote_d_type() const { return remote_d_type; }
     std::string get_remote_d_type_string() const;
 
-    void set_remote(inodeno_t ino, unsigned char d_type) { 
+    void set_remote(inodeno_t ino, unsigned char d_type) {
       remote_ino = ino;
       remote_d_type = d_type;
       inode = 0;
@@ -100,22 +100,27 @@ public:
 
 
   CDentry(std::string_view n, __u32 h,
+          mempool::mds_co::string alternate_name,
 	  snapid_t f, snapid_t l) :
     hash(h),
     first(f), last(l),
     item_dirty(this),
     lock(this, &lock_type),
     versionlock(this, &versionlock_type),
-    name(n)
+    name(n),
+    alternate_name(std::move(alternate_name))
   {}
-  CDentry(std::string_view n, __u32 h, inodeno_t ino, unsigned char dt,
+  CDentry(std::string_view n, __u32 h,
+          mempool::mds_co::string alternate_name,
+          inodeno_t ino, unsigned char dt,
 	  snapid_t f, snapid_t l) :
     hash(h),
     first(f), last(l),
     item_dirty(this),
     lock(this, &lock_type),
     versionlock(this, &versionlock_type),
-    name(n)
+    name(n),
+    alternate_name(std::move(alternate_name))
   {
     linkage.remote_ino = ino;
     linkage.remote_d_type = dt;
@@ -151,9 +156,15 @@ public:
   const CDir *get_dir() const { return dir; }
   CDir *get_dir() { return dir; }
   std::string_view get_name() const { return std::string_view(name); }
-  void set_alternate_name(std::string_view _alternate_name);
-  std::string_view get_alternate_name() const { return std::string_view(alternate_name); }
-  void decode_alternate_name(bufferlist::const_iterator &bl) { decode(alternate_name, bl); }
+  std::string_view get_alternate_name() const {
+    return std::string_view(alternate_name);
+  }
+  void set_alternate_name(mempool::mds_co::string altn) {
+    alternate_name = std::move(altn);
+  }
+  void set_alternate_name(std::string_view altn) {
+    alternate_name = mempool::mds_co::string(altn);
+  }
 
   __u32 get_hash() const { return hash; }
 
diff --git a/src/mds/CDir.cc b/src/mds/CDir.cc
index 5458dadcf94f7..ceb9d46e0bd3e 100644
--- a/src/mds/CDir.cc
+++ b/src/mds/CDir.cc
@@ -345,7 +345,7 @@ CDentry* CDir::add_null_dentry(std::string_view dname,
   ceph_assert(lookup_exact_snap(dname, last) == 0);
    
   // create dentry
-  CDentry* dn = new CDentry(dname, inode->hash_dentry_name(dname), first, last);
+  CDentry* dn = new CDentry(dname, inode->hash_dentry_name(dname), "", first, last);
   if (is_auth()) 
     dn->state_set(CDentry::STATE_AUTH);
 
@@ -382,13 +382,14 @@ CDentry* CDir::add_null_dentry(std::string_view dname,
 
 
 CDentry* CDir::add_primary_dentry(std::string_view dname, CInode *in,
+                                  mempool::mds_co::string alternate_name,
 				  snapid_t first, snapid_t last) 
 {
   // primary
   ceph_assert(lookup_exact_snap(dname, last) == 0);
   
   // create dentry
-  CDentry* dn = new CDentry(dname, inode->hash_dentry_name(dname), first, last);
+  CDentry* dn = new CDentry(dname, inode->hash_dentry_name(dname), std::move(alternate_name), first, last);
   if (is_auth()) 
     dn->state_set(CDentry::STATE_AUTH);
   if (is_auth() || !inode->is_stray()) {
@@ -431,13 +432,14 @@ CDentry* CDir::add_primary_dentry(std::string_view dname, CInode *in,
 }
 
 CDentry* CDir::add_remote_dentry(std::string_view dname, inodeno_t ino, unsigned char d_type,
+                                 mempool::mds_co::string alternate_name,
 				 snapid_t first, snapid_t last) 
 {
   // foreign
   ceph_assert(lookup_exact_snap(dname, last) == 0);
 
   // create dentry
-  CDentry* dn = new CDentry(dname, inode->hash_dentry_name(dname), ino, d_type, first, last);
+  CDentry* dn = new CDentry(dname, inode->hash_dentry_name(dname), std::move(alternate_name), ino, d_type, first, last);
   if (is_auth()) 
     dn->state_set(CDentry::STATE_AUTH);
   mdcache->lru.lru_insert_mid(dn);
@@ -1769,8 +1771,6 @@ CDentry *CDir::_load_dentry(
     mempool::mds_co::string alternate_name;
 
     CDentry::decode_remote(type, ino, d_type, alternate_name, q);
-    if (alternate_name.length())
-      dn->set_alternate_name(std::move(alternate_name));
 
     if (stale) {
       if (!dn) {
@@ -1787,14 +1787,15 @@ CDentry *CDir::_load_dentry(
 	  dnl->is_remote() &&
 	  dn->is_dirty() &&
 	  ino == dnl->get_remote_ino() &&
-	  d_type == dnl->get_remote_d_type()) {
+	  d_type == dnl->get_remote_d_type() &&
+          alternate_name == dn->get_alternate_name()) {
 	// see comment below
 	dout(10) << "_fetched  had underwater dentry " << *dn << ", marking clean" << dendl;
 	dn->mark_clean();
       }
     } else {
       // (remote) link
-      dn = add_remote_dentry(dname, ino, d_type, first, last);
+      dn = add_remote_dentry(dname, ino, d_type, std::move(alternate_name), first, last);
 
       // link to inode?
       CInode *in = mdcache->get_inode(ino);   // we may or may not have it.
@@ -1808,12 +1809,14 @@ CDentry *CDir::_load_dentry(
   }
   else if (type == 'I' || type == 'i') {
     InodeStore inode_data;
+    mempool::mds_co::string alternate_name;
     // inode
     // Load inode data before looking up or constructing CInode
     if (type == 'i') {
       DECODE_START(2, q);
-      if (struct_v >= 2)
-        dn->decode_alternate_name(q);
+      if (struct_v >= 2) {
+        decode(alternate_name, q);
+      }
       inode_data.decode(q);
       DECODE_FINISH(q);
     } else {
@@ -1891,7 +1894,7 @@ CDentry *CDir::_load_dentry(
 
         if (!undef_inode) {
           mdcache->add_inode(in); // add
-          dn = add_primary_dentry(dname, in, first, last); // link
+          dn = add_primary_dentry(dname, in, std::move(alternate_name), first, last); // link
         }
         dout(12) << "_fetched  got " << *dn << " " << *in << dendl;
 
@@ -1904,7 +1907,7 @@ CDentry *CDir::_load_dentry(
         //num_new_inodes_loaded++;
       } else if (g_conf().get_val<bool>("mds_hack_allow_loading_invalid_metadata")) {
 	dout(20) << "hack: adding duplicate dentry for " << *in << dendl;
-	dn = add_primary_dentry(dname, in, first, last);
+	dn = add_primary_dentry(dname, in, std::move(alternate_name), first, last);
       } else {
         dout(0) << "_fetched  badness: got (but i already had) " << *in
                 << " mode " << in->get_inode()->mode
@@ -2233,7 +2236,6 @@ void CDir::_encode_primary_inode_base(dentry_commit_item &item, bufferlist &dfts
 
   encode(item.oldest_snap, bl);
   encode(item.damage_flags, bl);
-  encode(item.alternate_name, bl);
   ENCODE_FINISH(bl);
 }
 
@@ -2330,6 +2332,7 @@ void CDir::_omap_commit_ops(int r, int op_prio, int64_t metapool, version_t vers
       bl.append('i');         // inode
 
       ENCODE_START(2, 1, bl);
+      encode(item.alternate_name, bl);
       _encode_primary_inode_base(item, dfts, bl);
       ENCODE_FINISH(bl);
     }
@@ -2450,19 +2453,17 @@ void CDir::_parse_dentry(CDentry *dn, dentry_commit_item &item,
 
   item.first = dn->first;
 
-  // Only in very rare case the dn->alternate_name not empty,
-  // so it won't cost much to copy it here
-  item.alternate_name = dn->get_alternate_name();
-
   // primary or remote?
-  if (dn->linkage.is_remote()) {
+  auto& linkage = dn->linkage;
+  item.alternate_name = dn->get_alternate_name();
+  if (linkage.is_remote()) {
     item.is_remote = true;
-    item.ino = dn->linkage.get_remote_ino();
-    item.d_type = dn->linkage.get_remote_d_type();
+    item.ino = linkage.get_remote_ino();
+    item.d_type = linkage.get_remote_d_type();
     dout(14) << " dn '" << dn->get_name() << "' remote ino " << item.ino << dendl;
-  } else if (dn->linkage.is_primary()) {
+  } else if (linkage.is_primary()) {
     // primary link
-    CInode *in = dn->linkage.get_inode();
+    CInode *in = linkage.get_inode();
     ceph_assert(in);
 
     dout(14) << " dn '" << dn->get_name() << "' inode " << *in << dendl;
@@ -2491,7 +2492,7 @@ void CDir::_parse_dentry(CDentry *dn, dentry_commit_item &item,
     item.oldest_snap = in->oldest_snap;
     item.damage_flags = in->damage_flags;
   } else {
-    ceph_assert(!dn->linkage.is_null());
+    ceph_assert(!linkage.is_null());
   }
 }
 
diff --git a/src/mds/CDir.h b/src/mds/CDir.h
index 985f94cb06296..3bb1374113dfb 100644
--- a/src/mds/CDir.h
+++ b/src/mds/CDir.h
@@ -370,9 +370,10 @@ public:
 
   CDentry* add_null_dentry(std::string_view dname,
 			   snapid_t first=2, snapid_t last=CEPH_NOSNAP);
-  CDentry* add_primary_dentry(std::string_view dname, CInode *in,
+  CDentry* add_primary_dentry(std::string_view dname, CInode *in, mempool::mds_co::string alternate_name,
 			      snapid_t first=2, snapid_t last=CEPH_NOSNAP);
   CDentry* add_remote_dentry(std::string_view dname, inodeno_t ino, unsigned char d_type,
+                             mempool::mds_co::string alternate_name,
 			     snapid_t first=2, snapid_t last=CEPH_NOSNAP);
   void remove_dentry( CDentry *dn );         // delete dentry
   void link_remote_inode( CDentry *dn, inodeno_t ino, unsigned char d_type);
diff --git a/src/mds/Locker.cc b/src/mds/Locker.cc
index dc112871d8926..a2fbe76625ab2 100644
--- a/src/mds/Locker.cc
+++ b/src/mds/Locker.cc
@@ -4252,7 +4252,7 @@ void Locker::issue_client_lease(CDentry *dn, MDRequestRef &mdr, int mask,
     lstat.mask = CEPH_LEASE_VALID | mask;
     lstat.duration_ms = (uint32_t)(1000 * mdcache->client_lease_durations[pool]);
     lstat.seq = ++l->seq;
-    lstat.alternate_name = mempool::mds_co::string(dn->get_alternate_name());
+    lstat.alternate_name = std::string(dn->alternate_name);
     encode_lease(bl, session->info, lstat);
     dout(20) << "issue_client_lease seq " << lstat.seq << " dur " << lstat.duration_ms << "ms "
 	     << " on " << *dn << dendl;
@@ -4260,7 +4260,7 @@ void Locker::issue_client_lease(CDentry *dn, MDRequestRef &mdr, int mask,
     // null lease
     LeaseStat lstat;
     lstat.mask = mask;
-    lstat.alternate_name = mempool::mds_co::string(dn->get_alternate_name());
+    lstat.alternate_name = std::string(dn->alternate_name);
     encode_lease(bl, session->info, lstat);
     dout(20) << "issue_client_lease no/null lease on " << *dn << dendl;
   }
diff --git a/src/mds/MDCache.cc b/src/mds/MDCache.cc
index b699028ab2000..ad7a8f5be0cd6 100644
--- a/src/mds/MDCache.cc
+++ b/src/mds/MDCache.cc
@@ -505,7 +505,7 @@ void MDCache::create_mydir_hierarchy(MDSGather *gather)
     CDir *straydir = stray->get_or_open_dirfrag(this, frag_t());
     CachedStackStringStream css;
     *css << "stray" << i;
-    CDentry *sdn = mydir->add_primary_dentry(css->str(), stray);
+    CDentry *sdn = mydir->add_primary_dentry(css->str(), stray, "");
     sdn->_mark_dirty(mds->mdlog->get_current_segment());
 
     stray->_get_inode()->dirstat = straydir->get_fnode()->fragstat;
@@ -1664,8 +1664,7 @@ void MDCache::journal_cow_dentry(MutationImpl *mut, EMetaBlob *metablob,
 	dn->first = dir_follows+1;
 	if (realm->has_snaps_in_range(oldfirst, dir_follows)) {
 	  CDir *dir = dn->dir;
-	  CDentry *olddn = dir->add_remote_dentry(dn->get_name(), in->ino(), in->d_type(),
-						  oldfirst, dir_follows);
+	  CDentry *olddn = dir->add_remote_dentry(dn->get_name(), in->ino(), in->d_type(), dn->alternate_name, oldfirst, dir_follows);
 	  dout(10) << " olddn " << *olddn << dendl;
 	  ceph_assert(dir->is_projected());
 	  olddn->set_projected_version(dir->get_projected_version());
@@ -1739,7 +1738,7 @@ void MDCache::journal_cow_dentry(MutationImpl *mut, EMetaBlob *metablob,
       mut->add_cow_inode(oldin);
       if (pcow_inode)
 	*pcow_inode = oldin;
-      CDentry *olddn = dir->add_primary_dentry(dn->get_name(), oldin, oldfirst, follows);
+      CDentry *olddn = dir->add_primary_dentry(dn->get_name(), oldin, dn->alternate_name, oldfirst, follows);
       dout(10) << " olddn " << *olddn << dendl;
       bool need_snapflush = !oldin->client_snap_caps.empty();
       if (need_snapflush) {
@@ -1751,8 +1750,7 @@ void MDCache::journal_cow_dentry(MutationImpl *mut, EMetaBlob *metablob,
       mut->add_cow_dentry(olddn);
     } else {
       ceph_assert(dnl->is_remote());
-      CDentry *olddn = dir->add_remote_dentry(dn->get_name(), dnl->get_remote_ino(), dnl->get_remote_d_type(),
-						  oldfirst, follows);
+      CDentry *olddn = dir->add_remote_dentry(dn->get_name(), dnl->get_remote_ino(), dnl->get_remote_d_type(), dn->alternate_name, oldfirst, follows);
       dout(10) << " olddn " << *olddn << dendl;
 
       olddn->set_projected_version(dir->get_projected_version());
@@ -4332,7 +4330,8 @@ void MDCache::rejoin_walk(CDir *dir, const ref_t<MMDSCacheRejoin> &rejoin)
       }
 
       dout(15) << " add_strong_dentry " << *dn << dendl;
-      rejoin->add_strong_dentry(dir->dirfrag(), dn->get_name(), dn->first, dn->last,
+      rejoin->add_strong_dentry(dir->dirfrag(), dn->get_name(), dn->get_alternate_name(),
+                                dn->first, dn->last,
 				dnl->is_primary() ? dnl->get_inode()->ino():inodeno_t(0),
 				dnl->is_remote() ? dnl->get_remote_ino():inodeno_t(0),
 				dnl->is_remote() ? dnl->get_remote_d_type():0, 
@@ -4554,7 +4553,8 @@ void MDCache::handle_cache_rejoin_weak(const cref_t<MMDSCacheRejoin> &weak)
       unsigned dnonce = dn->add_replica(from);
       dout(10) << " have " << *dn << dendl;
       if (ack) 
-	ack->add_strong_dentry(dir->dirfrag(), dn->get_name(), dn->first, dn->last,
+	ack->add_strong_dentry(dir->dirfrag(), dn->get_name(), dn->get_alternate_name(),
+                               dn->first, dn->last,
 			       dnl->get_inode()->ino(), inodeno_t(0), 0, 
 			       dnonce, dn->lock.get_replica_state());
 
@@ -4805,13 +4805,13 @@ void MDCache::handle_cache_rejoin_strong(const cref_t<MMDSCacheRejoin> &strong)
         }
         if (!dn) {
 	  if (d.is_remote()) {
-	    dn = dir->add_remote_dentry(ss.name, d.remote_ino, d.remote_d_type, d.first, ss.snapid);
+	    dn = dir->add_remote_dentry(ss.name, d.remote_ino, d.remote_d_type, mempool::mds_co::string(d.alternate_name), d.first, ss.snapid);
 	  } else if (d.is_null()) {
 	    dn = dir->add_null_dentry(ss.name, d.first, ss.snapid);
 	  } else {
 	    CInode *in = get_inode(d.ino, ss.snapid);
 	    if (!in) in = rejoin_invent_inode(d.ino, ss.snapid);
-	    dn = dir->add_primary_dentry(ss.name, in, d.first, ss.snapid);
+	    dn = dir->add_primary_dentry(ss.name, in, mempool::mds_co::string(d.alternate_name), d.first, ss.snapid);
 	  }
 	  dout(10) << " invented " << *dn << dendl;
         }
@@ -6163,7 +6163,8 @@ void MDCache::rejoin_send_acks()
 	  auto it = acks.find(r.first);
 	  if (it == acks.end())
 	    continue;
-	  it->second->add_strong_dentry(dir->dirfrag(), dn->get_name(), dn->first, dn->last,
+	  it->second->add_strong_dentry(dir->dirfrag(), dn->get_name(), dn->get_alternate_name(),
+                                           dn->first, dn->last,
 					   dnl->is_primary() ? dnl->get_inode()->ino():inodeno_t(0),
 					   dnl->is_remote() ? dnl->get_remote_ino():inodeno_t(0),
 					   dnl->is_remote() ? dnl->get_remote_d_type():0,
@@ -10658,7 +10659,7 @@ void MDCache::encode_replica_dir(CDir *dir, mds_rank_t to, bufferlist& bl)
 
 void MDCache::encode_replica_dentry(CDentry *dn, mds_rank_t to, bufferlist& bl)
 {
-  ENCODE_START(1, 1, bl);
+  ENCODE_START(2, 1, bl);
   encode(dn->get_name(), bl);
   encode(dn->last, bl);
 
@@ -10670,6 +10671,7 @@ void MDCache::encode_replica_dentry(CDentry *dn, mds_rank_t to, bufferlist& bl)
   dn->lock.encode_state_for_replica(bl);
   bool need_recover = mds->get_state() < MDSMap::STATE_ACTIVE;
   encode(need_recover, bl);
+  encode(dn->alternate_name, bl);
   ENCODE_FINISH(bl);
 }
 
@@ -10774,11 +10776,19 @@ void MDCache::decode_replica_dentry(CDentry *&dn, bufferlist::const_iterator& p,
   bool need_recover;
   decode(need_recover, p);
 
+  mempool::mds_co::string alternate_name;
+  if (struct_v >= 2) {
+    decode(alternate_name, p);
+  }
+
   if (is_new) {
+    dn->set_alternate_name(std::move(alternate_name));
     if (rino)
       dir->link_remote_inode(dn, rino, rdtype);
     if (need_recover)
       dn->lock.mark_need_recover();
+  } else {
+    ceph_assert(dn->alternate_name == alternate_name);
   }
 
   dir->take_dentry_waiting(name, dn->first, dn->last, finished);
diff --git a/src/mds/MDSDaemon.h b/src/mds/MDSDaemon.h
index 2f97de4b6f361..d92658934bd85 100644
--- a/src/mds/MDSDaemon.h
+++ b/src/mds/MDSDaemon.h
@@ -35,7 +35,7 @@
 #include "MDSMap.h"
 #include "MDSRank.h"
 
-#define CEPH_MDS_PROTOCOL    35 /* cluster internal */
+#define CEPH_MDS_PROTOCOL    36 /* cluster internal */
 
 class Messenger;
 class MonClient;
diff --git a/src/mds/MDSRank.cc b/src/mds/MDSRank.cc
index 11a07d68c4e39..1244fdd2236db 100644
--- a/src/mds/MDSRank.cc
+++ b/src/mds/MDSRank.cc
@@ -3689,6 +3689,7 @@ const char** MDSRankDispatcher::get_tracked_conf_keys() const
     "mds_session_cap_acquisition_throttle",
     "mds_session_max_caps_throttle_ratio",
     "mds_cap_acquisition_throttle_retry_request_time",
+    "mds_alternate_name_max",
     NULL
   };
   return KEYS;
diff --git a/src/mds/Migrator.cc b/src/mds/Migrator.cc
index f3b653dd47b4f..13bd2652aab2a 100644
--- a/src/mds/Migrator.cc
+++ b/src/mds/Migrator.cc
@@ -1793,8 +1793,9 @@ void Migrator::encode_export_dir(bufferlist& exportbl,
     if (dn->get_linkage()->is_remote()) {
       inodeno_t ino = dn->get_linkage()->get_remote_ino();
       unsigned char d_type = dn->get_linkage()->get_remote_d_type();
+      auto& alternate_name = dn->alternate_name;
       // remote link
-      CDentry::encode_remote(ino, d_type, dn->get_alternate_name(), exportbl);
+      CDentry::encode_remote(ino, d_type, alternate_name, exportbl);
       continue;
     }
 
@@ -1804,7 +1805,7 @@ void Migrator::encode_export_dir(bufferlist& exportbl,
 
     ENCODE_START(2, 1, exportbl);
     encode_export_inode(in, exportbl, exported_client_map, exported_client_metadata_map);  // encode, and (update state for) export
-    encode(dn->get_alternate_name(), exportbl);
+    encode(dn->alternate_name, exportbl);
     ENCODE_FINISH(exportbl);
 
     // directory?
@@ -3450,13 +3451,13 @@ void Migrator::decode_import_dir(bufferlist::const_iterator& blp,
       mempool::mds_co::string alternate_name;
 
       CDentry::decode_remote(icode, ino, d_type, alternate_name, blp);
-      if (alternate_name.length())
-        dn->set_alternate_name(std::move(alternate_name));
 
       if (dn->get_linkage()->is_remote()) {
 	ceph_assert(dn->get_linkage()->get_remote_ino() == ino);
+        ceph_assert(dn->get_alternate_name() == alternate_name);
       } else {
 	dir->link_remote_inode(dn, ino, d_type);
+        dn->set_alternate_name(std::move(alternate_name));
       }
     }
     else if (icode == 'I' || icode == 'i') {
@@ -3466,8 +3467,8 @@ void Migrator::decode_import_dir(bufferlist::const_iterator& blp,
         DECODE_START(2, blp);
         decode_import_inode(dn, blp, oldauth, ls,
                             peer_exports, updated_scatterlocks);
-	if (struct_v >= 2)
-          dn->decode_alternate_name(blp);
+        ceph_assert(!dn->is_projected());
+        decode(dn->alternate_name, blp);
         DECODE_FINISH(blp);
       } else {
         decode_import_inode(dn, blp, oldauth, ls,
diff --git a/src/mds/Server.cc b/src/mds/Server.cc
index fb7c74f8ed345..ecf47bd2c717c 100644
--- a/src/mds/Server.cc
+++ b/src/mds/Server.cc
@@ -1239,6 +1239,9 @@ void Server::handle_conf_change(const std::set<std::string>& changed) {
   if (changed.count("mds_cap_acquisition_throttle_retry_request_timeout")) {
     caps_throttle_retry_request_timeout = g_conf().get_val<double>("mds_cap_acquisition_throttle_retry_request_timeout");
   }
+  if (changed.count("mds_alternate_name_max")) {
+    alternate_name_max  = g_conf().get_val<Option::size_t>("mds_alternate_name_max");
+  }
 }
 
 /*
@@ -4326,6 +4329,13 @@ void Server::handle_client_openc(MDRequestRef& mdr)
 
   ceph_assert(dnl->is_null());
 
+  if (req->get_alternate_name().size() > alternate_name_max) {
+    dout(10) << " alternate_name longer than " << alternate_name_max << dendl;
+    respond_to_request(mdr, -ENAMETOOLONG);
+    return;
+  }
+  dn->set_alternate_name(req->get_alternate_name());
+
   // set layout
   file_layout_t layout;
   if (mdr->dir_layout != file_layout_t())
@@ -4390,8 +4400,6 @@ void Server::handle_client_openc(MDRequestRef& mdr)
 				   req->head.args.open.mode | S_IFREG, &layout);
   ceph_assert(newi);
 
-  dn->set_alternate_name(req->get_alternate_name());
-
   // it's a file.
   dn->push_projected_linkage(newi);
 
@@ -6261,6 +6269,14 @@ void Server::handle_client_mknod(MDRequestRef& mdr)
   if (!check_fragment_space(mdr, dn->get_dir()))
     return;
 
+  ceph_assert(dn->get_projected_linkage()->is_null());
+  if (req->get_alternate_name().size() > alternate_name_max) {
+    dout(10) << " alternate_name longer than " << alternate_name_max << dendl;
+    respond_to_request(mdr, -ENAMETOOLONG);
+    return;
+  }
+  dn->set_alternate_name(req->get_alternate_name());
+
   // set layout
   file_layout_t layout;
   if (mdr->dir_layout != file_layout_t())
@@ -6271,7 +6287,6 @@ void Server::handle_client_mknod(MDRequestRef& mdr)
   CInode *newi = prepare_new_inode(mdr, dn->get_dir(), inodeno_t(req->head.ino), mode, &layout);
   ceph_assert(newi);
 
-  dn->set_alternate_name(req->get_alternate_name());
   dn->push_projected_linkage(newi);
 
   auto _inode = newi->_get_inode();
@@ -6353,6 +6368,14 @@ void Server::handle_client_mkdir(MDRequestRef& mdr)
   if (!check_fragment_space(mdr, dir))
     return;
 
+  ceph_assert(dn->get_projected_linkage()->is_null());
+  if (req->get_alternate_name().size() > alternate_name_max) {
+    dout(10) << " alternate_name longer than " << alternate_name_max << dendl;
+    respond_to_request(mdr, -ENAMETOOLONG);
+    return;
+  }
+  dn->set_alternate_name(req->get_alternate_name());
+
   // new inode
   unsigned mode = req->head.args.mkdir.mode;
   mode &= ~S_IFMT;
@@ -6360,8 +6383,6 @@ void Server::handle_client_mkdir(MDRequestRef& mdr)
   CInode *newi = prepare_new_inode(mdr, dir, inodeno_t(req->head.ino), mode);
   ceph_assert(newi);
 
-  dn->set_alternate_name(req->get_alternate_name());
-
   // it's a directory.
   dn->push_projected_linkage(newi);
 
@@ -6423,6 +6444,8 @@ void Server::handle_client_mkdir(MDRequestRef& mdr)
 
 void Server::handle_client_symlink(MDRequestRef& mdr)
 {
+  const auto& req = mdr->client_request;
+
   mdr->disable_lock_cache();
   CDentry *dn = rdlock_path_xlock_dentry(mdr, true);
   if (!dn)
@@ -6436,8 +6459,11 @@ void Server::handle_client_symlink(MDRequestRef& mdr)
   if (!check_fragment_space(mdr, dir))
     return;
 
-  const cref_t<MClientRequest> &req = mdr->client_request;
-
+  ceph_assert(dn->get_projected_linkage()->is_null());
+  if (req->get_alternate_name().size() > alternate_name_max) {
+    dout(10) << " alternate_name longer than " << alternate_name_max << dendl;
+    respond_to_request(mdr, -ENAMETOOLONG);
+  }
   dn->set_alternate_name(req->get_alternate_name());
 
   unsigned mode = S_IFLNK | 0777;
@@ -6514,7 +6540,6 @@ void Server::handle_client_link(MDRequestRef& mdr)
     destdn = rdlock_path_xlock_dentry(mdr, false);
     if (!destdn)
       return;
-
   } else {
     auto ret = rdlock_two_paths_xlock_destdn(mdr, false);
     destdn = ret.first;
@@ -6526,11 +6551,17 @@ void Server::handle_client_link(MDRequestRef& mdr)
       return;
     }
 
-    destdn->set_alternate_name(req->get_alternate_name());
-
     targeti = ret.second->get_projected_linkage()->get_inode();
   }
 
+  ceph_assert(destdn->get_projected_linkage()->is_null());
+  if (req->get_alternate_name().size() > alternate_name_max) {
+    dout(10) << " alternate_name longer than " << alternate_name_max << dendl;
+    respond_to_request(mdr, -ENAMETOOLONG);
+    return;
+  }
+  destdn->set_alternate_name(req->get_alternate_name());
+
   if (targeti->is_dir()) {
     dout(7) << "target is a dir, failing..." << dendl;
     respond_to_request(mdr, -EINVAL);
@@ -7915,7 +7946,7 @@ public:
  */
 void Server::handle_client_rename(MDRequestRef& mdr)
 {
-  const cref_t<MClientRequest> &req = mdr->client_request;
+  const auto& req = mdr->client_request;
   dout(7) << "handle_client_rename " << *req << dendl;
 
   filepath destpath = req->get_filepath();
@@ -7925,6 +7956,12 @@ void Server::handle_client_rename(MDRequestRef& mdr)
     return;
   }
 
+  if (req->get_alternate_name().size() > alternate_name_max) {
+    dout(10) << " alternate_name longer than " << alternate_name_max << dendl;
+    respond_to_request(mdr, -ENAMETOOLONG);
+    return;
+  }
+
   auto [destdn, srcdn] = rdlock_two_paths_xlock_destdn(mdr, true);
   if (!destdn)
     return;
@@ -7955,8 +7992,6 @@ void Server::handle_client_rename(MDRequestRef& mdr)
     return;
   }
 
-  destdn->set_alternate_name(req->get_alternate_name());
-
   // is this a stray migration, reintegration or merge? (sanity checks!)
   if (mdr->reqid.name.is_mds() &&
       !(MDS_INO_IS_STRAY(srcpath.get_ino()) &&
@@ -7993,6 +8028,11 @@ void Server::handle_client_rename(MDRequestRef& mdr)
       respond_to_request(mdr, 0);  // no-op.  POSIX makes no sense.
       return;
     }
+    if (destdn->get_alternate_name() != req->get_alternate_name()) {
+      /* the dentry exists but the alternate_names do not match, fail... */
+      respond_to_request(mdr, -EINVAL);
+      return;
+    }
   }
 
   vector<CDentry*>& srctrace = mdr->dn[1];
@@ -8319,7 +8359,7 @@ void Server::handle_client_rename(MDRequestRef& mdr)
   mdr->ls = mdlog->get_current_segment();
   EUpdate *le = new EUpdate(mdlog, "rename");
   mdlog->start_entry(le);
-  le->metablob.add_client_req(mdr->reqid, mdr->client_request->get_oldest_client_tid());
+  le->metablob.add_client_req(mdr->reqid, req->get_oldest_client_tid());
   if (!mdr->more()->witnessed.empty()) {
     dout(20) << " noting uncommitted_peers " << mdr->more()->witnessed << dendl;
     
@@ -8331,7 +8371,7 @@ void Server::handle_client_rename(MDRequestRef& mdr)
     mdr->more()->is_remote_frozen_authpin = false;
   }
   
-  _rename_prepare(mdr, &le->metablob, &le->client_map, srcdn, destdn, straydn);
+  _rename_prepare(mdr, &le->metablob, &le->client_map, srcdn, destdn, req->get_alternate_name(), straydn);
   if (le->client_map.length())
     le->cmapv = mds->sessionmap.get_projected();
 
@@ -8394,6 +8434,9 @@ void Server::_rename_finish(MDRequestRef& mdr, CDentry *srcdn, CDentry *destdn,
 bool Server::_rename_prepare_witness(MDRequestRef& mdr, mds_rank_t who, set<mds_rank_t> &witnesse,
 				     vector<CDentry*>& srctrace, vector<CDentry*>& dsttrace, CDentry *straydn)
 {
+  const auto& client_req = mdr->client_request;
+  ceph_assert(client_req);
+
   if (mds->is_cluster_degraded() &&
       !mds->mdsmap->is_clientreplay_or_active_or_stopping(who)) {
     dout(10) << "_rename_prepare_witness mds." << who << " is not active" << dendl;
@@ -8411,6 +8454,7 @@ bool Server::_rename_prepare_witness(MDRequestRef& mdr, mds_rank_t who, set<mds_
   req->destdnpath = filepath(dsttrace.front()->get_dir()->ino());
   for (auto dn : dsttrace)
     req->destdnpath.push_dentry(dn->get_name());
+  req->alternate_name = client_req->alternate_name;
   if (straydn)
     mdcache->encode_replica_stray(straydn, who, req->straybl);
 
@@ -8503,7 +8547,8 @@ bool Server::_need_force_journal(CInode *diri, bool empty)
 
 void Server::_rename_prepare(MDRequestRef& mdr,
 			     EMetaBlob *metablob, bufferlist *client_map_bl,
-			     CDentry *srcdn, CDentry *destdn, CDentry *straydn)
+			     CDentry *srcdn, CDentry *destdn, std::string_view alternate_name,
+                             CDentry *straydn)
 {
   dout(10) << "_rename_prepare " << *mdr << " " << *srcdn << " " << *destdn << dendl;
   if (straydn)
@@ -8579,6 +8624,10 @@ void Server::_rename_prepare(MDRequestRef& mdr,
   }
 
   // dest
+  if (destdnl->is_null()) {
+    /* handle_client_rename checks that alternate_name matches for existing destdn */
+    destdn->set_alternate_name(alternate_name);
+  }
   if (srcdnl->is_remote()) {
     if (!linkmerge) {
       // destdn
@@ -9295,7 +9344,7 @@ void Server::handle_peer_rename_prep(MDRequestRef& mdr)
   le->rollback = mdr->more()->rollback_bl;
   
   bufferlist blah;  // inode import data... obviously not used if we're the peer
-  _rename_prepare(mdr, &le->commit, &blah, srcdn, destdn, straydn);
+  _rename_prepare(mdr, &le->commit, &blah, srcdn, destdn, mdr->peer_request->alternate_name, straydn);
 
   if (le->commit.empty()) {
     dout(10) << " empty metablob, skipping journal" << dendl;
diff --git a/src/mds/Server.h b/src/mds/Server.h
index d03d22834ff83..5d9daf335c0a6 100644
--- a/src/mds/Server.h
+++ b/src/mds/Server.h
@@ -289,7 +289,8 @@ public:
   bool _need_force_journal(CInode *diri, bool empty);
   void _rename_prepare(MDRequestRef& mdr,
 		       EMetaBlob *metablob, bufferlist *client_map_bl,
-		       CDentry *srcdn, CDentry *destdn, CDentry *straydn);
+		       CDentry *srcdn, CDentry *destdn, std::string_view alternate_name,
+                       CDentry *straydn);
   /* set not_journaling=true if you're going to discard the results --
    * this bypasses the asserts to make sure we're journaling the right
    * things on the right nodes */
@@ -467,6 +468,8 @@ private:
   uint64_t cap_acquisition_throttle;
   double max_caps_throttle_ratio;
   double caps_throttle_retry_request_timeout;
+
+  size_t alternate_name_max = g_conf().get_val<Option::size_t>("mds_alternate_name_max");
 };
 
 static inline constexpr auto operator|(Server::RecallFlags a, Server::RecallFlags b) {
diff --git a/src/mds/events/EMetaBlob.h b/src/mds/events/EMetaBlob.h
index 7938a83242a58..d555627a3c638 100644
--- a/src/mds/events/EMetaBlob.h
+++ b/src/mds/events/EMetaBlob.h
@@ -23,6 +23,7 @@
 #include "../LogSegment.h"
 
 #include "include/interval_set.h"
+#include "common/strescape.h"
 
 class MDSRank;
 class MDLog;
@@ -65,6 +66,7 @@ public:
     static const int STATE_NEED_SNAPFLUSH = (1<<3);
     static const int STATE_EPHEMERAL_RANDOM = (1<<4);
     std::string  dn;         // dentry
+    std::string alternate_name;
     snapid_t dnfirst, dnlast;
     version_t dnv{0};
     CInode::inode_const_ptr inode;      // if it's not XXX should not be part of mempool; wait for std::pmr to simplify
@@ -76,12 +78,12 @@ public:
     __u8 state{0};
     CInode::old_inode_map_const_ptr old_inodes; // XXX should not be part of mempool; wait for std::pmr to simplify
 
-    fullbit(std::string_view d, snapid_t df, snapid_t dl, 
+    fullbit(std::string_view d, std::string_view an, snapid_t df, snapid_t dl,
 	    version_t v, const CInode::inode_const_ptr& i, const fragtree_t &dft,
 	    const CInode::xattr_map_const_ptr& xa, std::string_view sym,
 	    snapid_t os, const bufferlist &sbl, __u8 st,
 	    const CInode::old_inode_map_const_ptr& oi) :
-      dn(d), dnfirst(df), dnlast(dl), dnv(v), inode(i), xattrs(xa),
+      dn(d), alternate_name(an), dnfirst(df), dnlast(dl), dnv(v), inode(i), xattrs(xa),
       oldest_snap(os), state(st), old_inodes(oi)
     {
       if (i->is_symlink())
@@ -93,7 +95,7 @@ public:
     explicit fullbit(bufferlist::const_iterator &p) {
       decode(p);
     }
-    fullbit() {}
+    fullbit() = default;
     fullbit(const fullbit&) = delete;
     ~fullbit() {}
     fullbit& operator=(const fullbit&) = delete;
@@ -113,7 +115,11 @@ public:
     void print(ostream& out) const {
       out << " fullbit dn " << dn << " [" << dnfirst << "," << dnlast << "] dnv " << dnv
 	  << " inode " << inode->ino
-	  << " state=" << state << std::endl;
+	  << " state=" << state;
+      if (!alternate_name.empty()) {
+          out << " altn " << binstrprint(alternate_name, 8);
+      }
+      out << std::endl;
     }
     string state_string() const {
       string state_string;
@@ -136,24 +142,28 @@ public:
    */
   struct remotebit {
     std::string dn;
-    snapid_t dnfirst, dnlast;
-    version_t dnv;
-    inodeno_t ino;
-    unsigned char d_type;
-    bool dirty;
-
-    remotebit(std::string_view d, snapid_t df, snapid_t dl, version_t v, inodeno_t i, unsigned char dt, bool dr) : 
-      dn(d), dnfirst(df), dnlast(dl), dnv(v), ino(i), d_type(dt), dirty(dr) { }
+    std::string alternate_name;
+    snapid_t dnfirst = 0, dnlast = 0;
+    version_t dnv = 0;
+    inodeno_t ino = 0;
+    unsigned char d_type = '\0';
+    bool dirty = false;
+
+    remotebit(std::string_view d, std::string_view an, snapid_t df, snapid_t dl, version_t v, inodeno_t i, unsigned char dt, bool dr) : 
+      dn(d), alternate_name(an), dnfirst(df), dnlast(dl), dnv(v), ino(i), d_type(dt), dirty(dr) { }
     explicit remotebit(bufferlist::const_iterator &p) { decode(p); }
-    remotebit(): dnfirst(0), dnlast(0), dnv(0), ino(0),
-	d_type('\0'), dirty(false) {}
+    remotebit() = default;
 
     void encode(bufferlist& bl) const;
     void decode(bufferlist::const_iterator &bl);
     void print(ostream& out) const {
       out << " remotebit dn " << dn << " [" << dnfirst << "," << dnlast << "] dnv " << dnv
 	  << " ino " << ino
-	  << " dirty=" << dirty << std::endl;
+	  << " dirty=" << dirty;
+      if (!alternate_name.empty()) {
+        out << " altn " << binstrprint(alternate_name, 8);
+      }
+      out << std::endl;
     }
     void dump(Formatter *f) const;
     static void generate_test_instances(std::list<remotebit*>& ls);
@@ -425,7 +435,7 @@ private:
       rdt = dn->get_projected_linkage()->get_remote_d_type();
     }
     lump.nremote++;
-    lump.add_dremote(dn->get_name(), dn->first, dn->last,
+    lump.add_dremote(dn->get_name(), dn->get_alternate_name(), dn->first, dn->last,
 		     dn->get_projected_version(), rino, rdt, dirty);
   }
 
@@ -460,7 +470,7 @@ private:
       sr->encode(snapbl);
 
     lump.nfull++;
-    lump.add_dfull(dn->get_name(), dn->first, dn->last, dn->get_projected_version(),
+    lump.add_dfull(dn->get_name(), dn->get_alternate_name(), dn->first, dn->last, dn->get_projected_version(),
 		   pi, in->dirfragtree, in->get_projected_xattrs(), in->symlink,
 		   in->oldest_snap, snapbl, state, in->get_old_inodes());
 
@@ -518,7 +528,7 @@ private:
     }
 
     string empty;
-    roots.emplace_back(empty, in->first, in->last, 0, pi, pdft, px, in->symlink,
+    roots.emplace_back(empty, "", in->first, in->last, 0, pi, pdft, px, in->symlink,
 		       in->oldest_snap, snapbl, (dirty ? fullbit::STATE_DIRTY : 0),
 		       in->get_old_inodes());
   }
diff --git a/src/mds/journal.cc b/src/mds/journal.cc
index d9118ca73144b..e17874650663b 100644
--- a/src/mds/journal.cc
+++ b/src/mds/journal.cc
@@ -444,7 +444,7 @@ void EMetaBlob::update_segment(LogSegment *ls)
 // EMetaBlob::fullbit
 
 void EMetaBlob::fullbit::encode(bufferlist& bl, uint64_t features) const {
-  ENCODE_START(8, 5, bl);
+  ENCODE_START(9, 5, bl);
   encode(dn, bl);
   encode(dnfirst, bl);
   encode(dnlast, bl);
@@ -471,11 +471,12 @@ void EMetaBlob::fullbit::encode(bufferlist& bl, uint64_t features) const {
   if (!inode->is_dir())
     encode(snapbl, bl);
   encode(oldest_snap, bl);
+  encode(alternate_name, bl);
   ENCODE_FINISH(bl);
 }
 
 void EMetaBlob::fullbit::decode(bufferlist::const_iterator &bl) {
-  DECODE_START(8, bl);
+  DECODE_START(9, bl);
   decode(dn, bl);
   decode(dnfirst, bl);
   decode(dnlast, bl);
@@ -509,6 +510,9 @@ void EMetaBlob::fullbit::decode(bufferlist::const_iterator &bl) {
     decode(snapbl, bl);
   }
   decode(oldest_snap, bl);
+  if (struct_v >= 9) {
+    decode(alternate_name, bl);
+  }
   DECODE_FINISH(bl);
 }
 
@@ -553,6 +557,7 @@ void EMetaBlob::fullbit::dump(Formatter *f) const
     }
     f->close_section(); // old inodes
   }
+  f->dump_string("alternate_name", alternate_name);
 }
 
 void EMetaBlob::fullbit::generate_test_instances(std::list<EMetaBlob::fullbit*>& ls)
@@ -561,7 +566,7 @@ void EMetaBlob::fullbit::generate_test_instances(std::list<EMetaBlob::fullbit*>&
   fragtree_t fragtree;
   auto _xattrs = CInode::allocate_xattr_map();
   bufferlist empty_snapbl;
-  fullbit *sample = new fullbit("/testdn", 0, 0, 0,
+  fullbit *sample = new fullbit("/testdn", "", 0, 0, 0,
                                 _inode, fragtree, _xattrs, "", 0, empty_snapbl,
                                 false, NULL);
   ls.push_back(sample);
@@ -635,7 +640,7 @@ void EMetaBlob::fullbit::update_inode(MDSRank *mds, CInode *in)
 
 void EMetaBlob::remotebit::encode(bufferlist& bl) const
 {
-  ENCODE_START(2, 2, bl);
+  ENCODE_START(3, 2, bl);
   encode(dn, bl);
   encode(dnfirst, bl);
   encode(dnlast, bl);
@@ -643,12 +648,13 @@ void EMetaBlob::remotebit::encode(bufferlist& bl) const
   encode(ino, bl);
   encode(d_type, bl);
   encode(dirty, bl);
+  encode(alternate_name, bl);
   ENCODE_FINISH(bl);
 }
 
 void EMetaBlob::remotebit::decode(bufferlist::const_iterator &bl)
 {
-  DECODE_START_LEGACY_COMPAT_LEN(2, 2, 2, bl);
+  DECODE_START_LEGACY_COMPAT_LEN(3, 2, 2, bl);
   decode(dn, bl);
   decode(dnfirst, bl);
   decode(dnlast, bl);
@@ -656,6 +662,8 @@ void EMetaBlob::remotebit::decode(bufferlist::const_iterator &bl)
   decode(ino, bl);
   decode(d_type, bl);
   decode(dirty, bl);
+  if (struct_v >= 3)
+    decode(alternate_name, bl);
   DECODE_FINISH(bl);
 }
 
@@ -688,12 +696,15 @@ void EMetaBlob::remotebit::dump(Formatter *f) const
   }
   f->dump_string("d_type", type_string);
   f->dump_string("dirty", dirty ? "true" : "false");
+  f->dump_string("alternate_name", alternate_name);
 }
 
 void EMetaBlob::remotebit::
 generate_test_instances(std::list<EMetaBlob::remotebit*>& ls)
 {
-  remotebit *remote = new remotebit("/test/dn", 0, 10, 15, 1, IFTODT(S_IFREG), false);
+  remotebit *remote = new remotebit("/test/dn", "", 0, 10, 15, 1, IFTODT(S_IFREG), false);
+  ls.push_back(remote);
+  remote = new remotebit("/test/dn2", "foo", 0, 10, 15, 1, IFTODT(S_IFREG), false);
   ls.push_back(remote);
 }
 
@@ -1352,7 +1363,7 @@ void EMetaBlob::replay(MDSRank *mds, LogSegment *logseg, MDPeerUpdate *peerup)
     for (const auto& rb : lump.get_dremote()) {
       CDentry *dn = dir->lookup_exact_snap(rb.dn, rb.dnlast);
       if (!dn) {
-	dn = dir->add_remote_dentry(rb.dn, rb.ino, rb.d_type, rb.dnfirst, rb.dnlast);
+	dn = dir->add_remote_dentry(rb.dn, rb.ino, rb.d_type, mempool::mds_co::string(rb.alternate_name), rb.dnfirst, rb.dnlast);
 	dn->set_version(rb.dnv);
 	if (rb.dirty) dn->_mark_dirty(logseg);
 	dout(10) << "EMetaBlob.replay added " << *dn << dendl;
@@ -1368,6 +1379,7 @@ void EMetaBlob::replay(MDSRank *mds, LogSegment *logseg, MDPeerUpdate *peerup)
 	  }
 	  dir->unlink_inode(dn, false);
 	}
+        dn->set_alternate_name(mempool::mds_co::string(rb.alternate_name));
 	dir->link_remote_inode(dn, rb.ino, rb.d_type);
 	dn->set_version(rb.dnv);
 	if (rb.dirty) dn->_mark_dirty(logseg);
diff --git a/src/messages/MClientReply.h b/src/messages/MClientReply.h
index bb20a83b779a7..bd58e0883ab0d 100644
--- a/src/messages/MClientReply.h
+++ b/src/messages/MClientReply.h
@@ -24,6 +24,7 @@
 #include "msg/Message.h"
 #include "include/ceph_features.h"
 #include "common/errno.h"
+#include "common/strescape.h"
 
 /***
  *
@@ -48,12 +49,12 @@
 
 struct LeaseStat {
   // this matches ceph_mds_reply_lease
-  __u16 mask;
-  __u32 duration_ms;  
-  __u32 seq;
+  __u16 mask = 0;
+  __u32 duration_ms = 0;
+  __u32 seq = 0;
   std::string alternate_name;
 
-  LeaseStat() : mask(0), duration_ms(0), seq(0) {}
+  LeaseStat() = default;
   LeaseStat(__u16 msk, __u32 dur, __u32 sq) : mask{msk}, duration_ms{dur}, seq{sq} {}
 
   void decode(ceph::buffer::list::const_iterator &bl, const uint64_t features) {
@@ -76,7 +77,11 @@ struct LeaseStat {
 };
 
 inline std::ostream& operator<<(std::ostream& out, const LeaseStat& l) {
-  return out << "lease(mask " << l.mask << " dur " << l.duration_ms << ")";
+  out << "lease(mask " << l.mask << " dur " << l.duration_ms;
+  if (l.alternate_name.size()) {
+    out << " altn " << binstrprint(l.alternate_name, 128) << ")";
+  }
+  return out << ")";
 }
 
 struct DirStat {
diff --git a/src/messages/MClientRequest.h b/src/messages/MClientRequest.h
index 1463e26194feb..a02572367c069 100644
--- a/src/messages/MClientRequest.h
+++ b/src/messages/MClientRequest.h
@@ -297,6 +297,8 @@ public:
     }
     //if (!get_filepath().empty()) 
     out << " " << get_filepath();
+    if (alternate_name.size())
+      out << " (" << alternate_name << ") ";
     if (!get_filepath2().empty())
       out << " " << get_filepath2();
     if (stamp != utime_t())
diff --git a/src/messages/MMDSCacheRejoin.h b/src/messages/MMDSCacheRejoin.h
index 7bd4099994938..6f13ccd105572 100644
--- a/src/messages/MMDSCacheRejoin.h
+++ b/src/messages/MMDSCacheRejoin.h
@@ -86,15 +86,15 @@ public:
 
   struct dn_strong {
     snapid_t first;
-    inodeno_t ino;
-    inodeno_t remote_ino;
-    unsigned char remote_d_type;
-    uint32_t nonce;
-    int32_t lock;
-    dn_strong() : 
-      ino(0), remote_ino(0), remote_d_type(0), nonce(0), lock(0) {}
-    dn_strong(snapid_t f, inodeno_t pi, inodeno_t ri, unsigned char rdt, int n, int l) : 
-      first(f), ino(pi), remote_ino(ri), remote_d_type(rdt), nonce(n), lock(l) {}
+    std::string alternate_name;
+    inodeno_t ino = 0;
+    inodeno_t remote_ino = 0;
+    unsigned char remote_d_type = 0;
+    uint32_t nonce = 0;
+    int32_t lock = 0;
+    dn_strong() = default;
+    dn_strong(snapid_t f, std::string_view altn, inodeno_t pi, inodeno_t ri, unsigned char rdt, int n, int l) :
+      first(f), alternate_name(altn), ino(pi), remote_ino(ri), remote_d_type(rdt), nonce(n), lock(l) {}
     bool is_primary() const { return ino > 0; }
     bool is_remote() const { return remote_ino > 0; }
     bool is_null() const { return ino == 0 && remote_ino == 0; }
@@ -106,6 +106,7 @@ public:
       encode(remote_d_type, bl);
       encode(nonce, bl);
       encode(lock, bl);
+      encode(alternate_name, bl);
     }
     void decode(ceph::buffer::list::const_iterator &bl) {
       using ceph::decode;
@@ -115,14 +116,15 @@ public:
       decode(remote_d_type, bl);
       decode(nonce, bl);
       decode(lock, bl);
+      decode(alternate_name, bl);
     }
   };
   WRITE_CLASS_ENCODER(dn_strong)
 
   struct dn_weak {
     snapid_t first;
-    inodeno_t ino;
-    dn_weak() : ino(0) {}
+    inodeno_t ino = 0;
+    dn_weak() = default;
     dn_weak(snapid_t f, inodeno_t pi) : first(f), ino(pi) {}
     void encode(ceph::buffer::list &bl) const {
       using ceph::encode;
@@ -241,8 +243,9 @@ public:
   void add_weak_primary_dentry(inodeno_t dirino, std::string_view dname, snapid_t first, snapid_t last, inodeno_t ino) {
     weak[dirino][string_snap_t(dname, last)] = dn_weak(first, ino);
   }
-  void add_strong_dentry(dirfrag_t df, std::string_view dname, snapid_t first, snapid_t last, inodeno_t pi, inodeno_t ri, unsigned char rdt, int n, int ls) {
-    strong_dentries[df][string_snap_t(dname, last)] = dn_strong(first, pi, ri, rdt, n, ls);
+  void add_strong_dentry(dirfrag_t df, std::string_view dname, std::string_view altn, snapid_t first, snapid_t last, inodeno_t pi, inodeno_t ri, unsigned char rdt, int n, int ls) {
+    auto& m = strong_dentries[df];
+    m.insert_or_assign(string_snap_t(dname, last), dn_strong(first, altn, pi, ri, rdt, n, ls));
   }
   void add_dentry_authpin(dirfrag_t df, std::string_view dname, snapid_t last,
 			  const metareqid_t& ri, __u32 attempt) {
diff --git a/src/messages/MMDSPeerRequest.h b/src/messages/MMDSPeerRequest.h
index 68861211cf147..389f8b96bbd37 100644
--- a/src/messages/MMDSPeerRequest.h
+++ b/src/messages/MMDSPeerRequest.h
@@ -118,6 +118,7 @@ public:
   // for rename prep
   filepath srcdnpath;
   filepath destdnpath;
+  std::string alternate_name;
   std::set<mds_rank_t> witnesses;
   ceph::buffer::list inode_export;
   version_t inode_export_v;
@@ -192,6 +193,7 @@ public:
     encode(straybl, payload);
     encode(srci_snapbl, payload);
     encode(desti_snapbl, payload);
+    encode(alternate_name, payload);
   }
   void decode_payload() override {
     using ceph::decode;
@@ -213,6 +215,7 @@ public:
     decode(straybl, p);
     decode(srci_snapbl, p);
     decode(desti_snapbl, p);
+    decode(alternate_name, p);
   }
 
   std::string_view get_type_name() const override { return "peer_request"; }
-- 
2.39.5