Merge pull request ceph#54957 from jzhu116-bloomberg/wip-63799

rgw: revert PR ceph#41897 to allow multiple delete markers to be created

Reviewed-by: Casey Bodley <[email protected]>

Author: cbodley

qa/workunits/rgw/test_rgw_versioning.py

@@ -58,9 +58,21 @@ def main():
     assert 'DeleteMarker' in resp, 'DeleteMarker key not present in response'
     assert resp['DeleteMarker'], 'DeleteMarker value not True in response'
     assert 'VersionId' in resp, 'VersionId key not present in response'
-    version_id = resp['VersionId']
-    bucket.Object(key).delete()
-    connection.ObjectVersion(bucket.name, key, version_id).delete()
+    version_id_1 = resp['VersionId']
+
+    resp = bucket.Object(key).delete()
+    assert 'DeleteMarker' in resp, 'DeleteMarker key not present in response'
+    assert resp['DeleteMarker'], 'DeleteMarker value not True in response'
+    assert 'VersionId' in resp, 'VersionId key not present in response'
+    version_id_2 = resp['VersionId']
+
+    connection.ObjectVersion(bucket.name, key, version_id_2).delete()
+    # bucket index should only include entries for an object version
+    out = exec_cmd(f'radosgw-admin bi list --bucket {BUCKET_NAME}')
+    json_out = json.loads(out.replace(b'\x80', b'0x80'))
+    assert len(json_out) == 4, 'bucket index did not only include entries for an object version'
+
+    connection.ObjectVersion(bucket.name, key, version_id_1).delete()
     # bucket index should now be empty
     out = exec_cmd(f'radosgw-admin bi list --bucket {BUCKET_NAME}')
     json_out = json.loads(out.replace(b'\x80', b'0x80'))

src/cls/rgw/cls_rgw.cc

@@ -1811,28 +1811,6 @@ static int rgw_bucket_link_olh(cls_method_context_t hctx, bufferlist *in, buffer
     return ret;
   }
 
-  BIOLHEntry olh(hctx, op.key);
-  bool olh_read_attempt = false;
-  bool olh_found = false;
-  if (!existed && op.delete_marker) {
-    /* read olh */
-    ret = olh.init(&olh_found);
-    if (ret < 0) {
-      return ret;
-    }
-    olh_read_attempt = true;
-
-    // if we're deleting (i.e., adding a delete marker, and the OLH
-    // indicates it already refers to a delete marker, error out)
-    if (olh_found && olh.get_entry().delete_marker) {
-      CLS_LOG(10,
-	      "%s: delete marker received for \"%s\" although OLH"
-	      " already refers to a delete marker",
-	      __func__, escape_str(op.key.to_string()).c_str());
-      return -ENOENT;
-    }
-  }
-
   if (existed && !real_clock::is_zero(op.unmod_since)) {
     timespec mtime = ceph::real_clock::to_timespec(obj.mtime());
     timespec unmod = ceph::real_clock::to_timespec(op.unmod_since);
@@ -1885,12 +1863,11 @@ static int rgw_bucket_link_olh(cls_method_context_t hctx, bufferlist *in, buffer
   }
 
   /* read olh */
-  if (!olh_read_attempt) { // only read if we didn't attempt earlier
-    ret = olh.init(&olh_found);
-    if (ret < 0) {
-      return ret;
-    }
-    olh_read_attempt = true;
+  BIOLHEntry olh(hctx, op.key);
+  bool olh_found = false;
+  ret = olh.init(&olh_found);
+  if (ret < 0) {
+    return ret;
   }
 
   const uint64_t prev_epoch = olh.get_epoch();

src/test/rgw/rgw_multi/tests.py

@@ -1102,6 +1102,8 @@ def test_delete_marker_full_sync():
         # create a delete marker
         key2 = new_key(zone, bucket, 'obj')
         key2.delete()
+        key2.delete()
+        key2.delete()
 
     # wait for full sync
     for _, bucket in zone_bucket:
@@ -1126,11 +1128,40 @@ def test_suspended_delete_marker_full_sync():
         # create a delete marker
         key2 = new_key(zone, bucket, 'obj')
         key2.delete()
+        key2.delete()
+        key2.delete()
 
     # wait for full sync
     for _, bucket in zone_bucket:
         zonegroup_bucket_checkpoint(zonegroup_conns, bucket.name)
 
+def test_concurrent_delete_markers_incremental_sync():
+    zonegroup = realm.master_zonegroup()
+    zonegroup_conns = ZonegroupConns(zonegroup)
+    zone = zonegroup_conns.rw_zones[0]
+
+    # create a versioned bucket
+    bucket = zone.create_bucket(gen_bucket_name())
+    log.debug('created bucket=%s', bucket.name)
+    bucket.configure_versioning(True)
+
+    zonegroup_meta_checkpoint(zonegroup)
+
+    obj = 'obj'
+
+    # upload a dummy object and wait for sync. this forces each zone to finish
+    # a full sync and switch to incremental
+    new_key(zone, bucket, obj).set_contents_from_string('')
+    zonegroup_bucket_checkpoint(zonegroup_conns, bucket.name)
+
+    # create several concurrent delete markers on each zone and let them race to sync
+    for i in range(2):
+        for zone_conn in zonegroup_conns.rw_zones:
+            key = new_key(zone_conn, bucket, obj)
+            key.delete()
+
+    zonegroup_bucket_checkpoint(zonegroup_conns, bucket.name)
+
 def test_bucket_versioning():
     buckets, zone_bucket = create_bucket_per_zone_in_realm()
     for _, bucket in zone_bucket: