rgw: is_olh() predicate should use user.rgw.olh.ver xattr

The predicate which determines whether an object is an OLH object was previously based off of the
user.rgw.olh.info xattr, but this xattr is not set when the OLH object is first created. This
means that in cases where an error prevents that xattr from being set later, the OLH object is
incorrectly interpreted as an empty unversioned object. This change modifies the is_olh function
to use the user.rgw.olh.ver xattr. This xattr is set at the time of OLH creation and therefore
ensures consistency.

Fixes: https://tracker.ceph.com/issues/61359
Signed-off-by: Cory Snyder <csnyder@1111systems.com>
(cherry picked from commit 74ab3eb94a4177eacdac4fe48528542ba355ecfe)

diff --git a/src/rgw/driver/rados/rgw_rados.cc b/src/rgw/driver/rados/rgw_rados.cc
index a8c26a45f8c2890e33b76ab969eeeaae7b5a9601..21f22214bab48d40def13798694bea1b99d636a3 100644 (file)
--- a/src/rgw/driver/rados/rgw_rados.cc
+++ b/src/rgw/driver/rados/rgw_rados.cc
@@ -5465,7 +5465,7 @@ static void generate_fake_tag(const DoutPrefixProvider *dpp, RGWRados* store, ma
 
 static bool is_olh(map<string, bufferlist>& attrs)
 {
-  map<string, bufferlist>::iterator iter = attrs.find(RGW_ATTR_OLH_INFO);
+  map<string, bufferlist>::iterator iter = attrs.find(RGW_ATTR_OLH_VER);
   return (iter != attrs.end());
 }
 
@@ -6769,6 +6769,18 @@ int RGWRados::olh_init_modification_impl(const DoutPrefixProvider *dpp, const RG
   if (has_tag) {
     /* guard against racing writes */
     bucket_index_guard_olh_op(dpp, state, op);
+  } else if (state.exists) {
+    // This is the case where a null versioned object already exists for this key
+    // but it hasn't been initialized as an OLH object yet. We immediately add
+    // the RGW_ATTR_OLH_INFO attr so that the OLH points back to itself and
+    // therefore effectively makes this an unobservable modification.
+    op.cmpxattr(RGW_ATTR_OLH_ID_TAG, CEPH_OSD_CMPXATTR_OP_EQ, bufferlist());
+    RGWOLHInfo info;
+    info.target = olh_obj;
+    info.removed = false;
+    bufferlist bl;
+    encode(info, bl);
+    op.setxattr(RGW_ATTR_OLH_INFO, bl);
   }
 
   if (!has_tag) {
@@ -7677,7 +7689,7 @@ int RGWRados::get_olh(const DoutPrefixProvider *dpp, RGWBucketInfo& bucket_info,
     return r;
   }
 
-  auto iter = attrset.find(RGW_ATTR_OLH_INFO);
+  auto iter = attrset.find(RGW_ATTR_OLH_VER);
   if (iter == attrset.end()) { /* not an olh */
     return -EINVAL;
   }
@@ -7780,10 +7792,14 @@ int RGWRados::follow_olh(const DoutPrefixProvider *dpp, RGWBucketInfo& bucket_in
     }
   }
 
-  auto iter = state->attrset.find(RGW_ATTR_OLH_INFO);
+  auto iter = state->attrset.find(RGW_ATTR_OLH_VER);
   if (iter == state->attrset.end()) {
     return -EINVAL;
   }
+  iter = state->attrset.find(RGW_ATTR_OLH_INFO);
+  if (iter == state->attrset.end()) {
+    return -ENOENT;
+  }
 
   RGWOLHInfo olh;
   int ret = decode_olh_info(dpp, cct, iter->second, &olh);