rgw: convert plain object to versioned (with null version) when removing

Fixes #15243

When removing a plain null versioned object (was created prior to bucket versioning
enabled), need to convert the bucket index representation to a versioned one. This
is needed so that all the versioning mechanics play together.

Signed-off-by: Yehuda Sadeh <yehuda@redhat.com>

diff --git a/src/cls/rgw/cls_rgw.cc b/src/cls/rgw/cls_rgw.cc
index 4920d46ac100a98515bb3d62ac89bbfe36a634a2..5655006649619ef0d3980525bdef8dea610e5ead 100644 (file)
--- a/src/cls/rgw/cls_rgw.cc
+++ b/src/cls/rgw/cls_rgw.cc
@@ -1073,6 +1073,9 @@ public:
     initialized = true;
   }
 
+  void set_epoch(uint64_t epoch) {
+    instance_entry.versioned_epoch = epoch;
+  }
 
   int unlink_list_entry() {
     string list_idx;
@@ -1574,12 +1577,27 @@ static int rgw_bucket_unlink_instance(cls_method_context_t hctx, bufferlist *in,
     return ret;
   }
 
-  ret = olh.init(NULL);
+  bool olh_found;
+  ret = olh.init(&olh_found);
   if (ret < 0) {
     CLS_LOG(0, "ERROR: olh.init() returned ret=%d", ret);
     return ret;
   }
 
+  if (!olh_found) {
+    bool instance_only = false;
+    cls_rgw_obj_key key(dest_key.name);
+    ret = convert_plain_entry_to_versioned(hctx, key, true, instance_only);
+    if (ret < 0) {
+      CLS_LOG(0, "ERROR: convert_plain_entry_to_versioned ret=%d", ret);
+      return ret;
+    }
+    olh.update(dest_key, false);
+    olh.set_tag(op.olh_tag);
+
+    obj.set_epoch(1);
+  }
+
   if (!olh.start_modify(op.olh_epoch)) {
     ret = obj.unlink_list_entry();
     if (ret < 0) {

diff --git a/src/cls/rgw/cls_rgw_client.cc b/src/cls/rgw/cls_rgw_client.cc
index 40041aa484f6264f7a3d4f01e28a8ea8177f80d4..1cf1156385bb22d35690ada8f76673fca2738a9f 100644 (file)
--- a/src/cls/rgw/cls_rgw_client.cc
+++ b/src/cls/rgw/cls_rgw_client.cc
@@ -332,13 +332,14 @@ int cls_rgw_bucket_link_olh(librados::IoCtx& io_ctx, const string& oid, const cl
 
 int cls_rgw_bucket_unlink_instance(librados::IoCtx& io_ctx, const string& oid,
                                    const cls_rgw_obj_key& key, const string& op_tag,
-                                   uint64_t olh_epoch, bool log_op)
+                                   const string& olh_tag, uint64_t olh_epoch, bool log_op)
 {
   bufferlist in, out;
   struct rgw_cls_unlink_instance_op call;
   call.key = key;
   call.op_tag = op_tag;
   call.olh_epoch = olh_epoch;
+  call.olh_tag = olh_tag;
   call.log_op = log_op;
   ::encode(call, in);
   int r = io_ctx.exec(oid, "rgw", "bucket_unlink_instance", in, out);

diff --git a/src/cls/rgw/cls_rgw_client.h b/src/cls/rgw/cls_rgw_client.h
index 130a009f5fff02db148cdde33261eb224478564d..1b02a5eabf4e6c786b3d764a54afec36792fd789 100644 (file)
--- a/src/cls/rgw/cls_rgw_client.h
+++ b/src/cls/rgw/cls_rgw_client.h
@@ -333,7 +333,7 @@ int cls_rgw_bucket_link_olh(librados::IoCtx& io_ctx, const string& oid, const cl
                             bool delete_marker, const string& op_tag, struct rgw_bucket_dir_entry_meta *meta,
                             uint64_t olh_epoch, ceph::real_time unmod_since, bool high_precision_time, bool log_op);
 int cls_rgw_bucket_unlink_instance(librados::IoCtx& io_ctx, const string& oid, const cls_rgw_obj_key& key, const string& op_tag,
-                                   uint64_t olh_epoch, bool log_op);
+                                   const string& olh_tag, uint64_t olh_epoch, bool log_op);
 int cls_rgw_get_olh_log(librados::IoCtx& io_ctx, string& oid, librados::ObjectReadOperation& op, const cls_rgw_obj_key& olh, uint64_t ver_marker,
                         const string& olh_tag,
                         map<uint64_t, vector<struct rgw_bucket_olh_log_entry> > *log, bool *is_truncated);

diff --git a/src/cls/rgw/cls_rgw_ops.h b/src/cls/rgw/cls_rgw_ops.h
index 3f2c8d9798e4f5f1852b81a5f68e71d1b9be9af0..e8a7661d090e37850bc48d5b864c60edd8116ae1 100644 (file)
--- a/src/cls/rgw/cls_rgw_ops.h
+++ b/src/cls/rgw/cls_rgw_ops.h
@@ -222,26 +222,31 @@ struct rgw_cls_unlink_instance_op {
   uint64_t olh_epoch;
   bool log_op;
   uint16_t bilog_flags;
+  string olh_tag;
 
   rgw_cls_unlink_instance_op() : olh_epoch(0), log_op(false), bilog_flags(0) {}
 
   void encode(bufferlist& bl) const {
-    ENCODE_START(1, 1, bl);
+    ENCODE_START(2, 1, bl);
     ::encode(key, bl);
     ::encode(op_tag, bl);
     ::encode(olh_epoch, bl);
     ::encode(log_op, bl);
     ::encode(bilog_flags, bl);
+    ::encode(olh_tag, bl);
     ENCODE_FINISH(bl);
   }
 
   void decode(bufferlist::iterator& bl) {
-    DECODE_START(1, bl);
+    DECODE_START(2, bl);
     ::decode(key, bl);
     ::decode(op_tag, bl);
     ::decode(olh_epoch, bl);
     ::decode(log_op, bl);
     ::decode(bilog_flags, bl);
+    if (struct_v >= 2) {
+      ::decode(olh_tag, bl);
+    }
     DECODE_FINISH(bl);
   }
 

diff --git a/src/rgw/rgw_rados.cc b/src/rgw/rgw_rados.cc
index 6d5856498d314e5cb83b19d57e46d15138465388..9c565af6285f1cb7ab1f0334bfeeb70e17793325 100644 (file)
--- a/src/rgw/rgw_rados.cc
+++ b/src/rgw/rgw_rados.cc
@@ -9403,7 +9403,7 @@ void RGWRados::bucket_index_guard_olh_op(RGWObjState& olh_state, ObjectOperation
   op.cmpxattr(RGW_ATTR_OLH_ID_TAG, CEPH_OSD_CMPXATTR_OP_EQ, olh_state.olh_tag);
 }
 
-int RGWRados::bucket_index_unlink_instance(rgw_obj& obj_instance, const string& op_tag, uint64_t olh_epoch)
+int RGWRados::bucket_index_unlink_instance(rgw_obj& obj_instance, const string& op_tag, const string& olh_tag, uint64_t olh_epoch)
 {
   rgw_rados_ref ref;
   rgw_bucket bucket;
@@ -9420,7 +9420,7 @@ int RGWRados::bucket_index_unlink_instance(rgw_obj& obj_instance, const string&
   }
 
   cls_rgw_obj_key key(obj_instance.get_index_key_name(), obj_instance.get_instance());
-  ret = cls_rgw_bucket_unlink_instance(bs.index_ctx, bs.bucket_obj, key, op_tag, olh_epoch, get_zone().log_data);
+  ret = cls_rgw_bucket_unlink_instance(bs.index_ctx, bs.bucket_obj, key, op_tag, olh_tag, olh_epoch, get_zone().log_data);
   if (ret < 0) {
     return ret;
   }
@@ -9760,7 +9760,9 @@ int RGWRados::unlink_obj_instance(RGWObjectCtx& obj_ctx, RGWBucketInfo& bucket_i
       return ret;
     }
 
-    ret = bucket_index_unlink_instance(target_obj, op_tag, olh_epoch);
+    string olh_tag(state->olh_tag.c_str(), state->olh_tag.length());
+
+    ret = bucket_index_unlink_instance(target_obj, op_tag, olh_tag, olh_epoch);
     if (ret < 0) {
       ldout(cct, 20) << "bucket_index_link_olh() target_obj=" << target_obj << " returned " << ret << dendl;
       if (ret == -ECANCELED) {

diff --git a/src/rgw/rgw_rados.h b/src/rgw/rgw_rados.h
index 8a2db376d29bd7ce24d90314927cca1b6b3e52b2..2a7a0be68284d2ad336bf243bf49fa1df579f8de 100644 (file)
--- a/src/rgw/rgw_rados.h
+++ b/src/rgw/rgw_rados.h
@@ -2643,7 +2643,7 @@ public:
                             const string& op_tag, struct rgw_bucket_dir_entry_meta *meta,
                             uint64_t olh_epoch,
                             ceph::real_time unmod_since, bool high_precision_time);
-  int bucket_index_unlink_instance(rgw_obj& obj_instance, const string& op_tag, uint64_t olh_epoch);
+  int bucket_index_unlink_instance(rgw_obj& obj_instance, const string& op_tag, const string& olh_tag, uint64_t olh_epoch);
   int bucket_index_read_olh_log(RGWObjState& state, rgw_obj& obj_instance, uint64_t ver_marker,
                                 map<uint64_t, vector<rgw_bucket_olh_log_entry> > *log, bool *is_truncated);
   int bucket_index_trim_olh_log(RGWObjState& obj_state, rgw_obj& obj_instance, uint64_t ver);