rgw: guard versioned object removal from circular sync

check if marker instance already exists, only apply (and update bi log)
if does not exist. Propagate the versioning flag so that we apply the
appropriate operation on the object.

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

6 files changed, 24 insertions, 4 deletions

diff --git a/src/cls/rgw/cls_rgw.cc b/src/cls/rgw/cls_rgw.cc
index 50faf0086c2..0fd88905eb1 100644
--- a/src/cls/rgw/cls_rgw.cc
+++ b/src/cls/rgw/cls_rgw.cc
@@ -1183,6 +1183,9 @@ public:
     return 0;
   }
 
+  time_t mtime() {
+    return instance_entry.meta.mtime;
+  }
 };
 
 
@@ -1386,6 +1389,12 @@ static int rgw_bucket_link_olh(cls_method_context_t hctx, bufferlist *in, buffer
     return ret;
   }
 
+  if (existed && op.unmod_since > 0) {
+    if (obj.mtime() >= op.unmod_since) {
+      return 0; /* no need to set error, we just return 0 and avoid writing to the bi log */
+    }
+  }
+
   bool removing;
 
   /*
diff --git a/src/cls/rgw/cls_rgw_client.cc b/src/cls/rgw/cls_rgw_client.cc
index c2847741713..686cd6cbd67 100644
--- a/src/cls/rgw/cls_rgw_client.cc
+++ b/src/cls/rgw/cls_rgw_client.cc
@@ -306,7 +306,7 @@ int cls_rgw_bi_list(librados::IoCtx& io_ctx, const string oid,
 
 int cls_rgw_bucket_link_olh(librados::IoCtx& io_ctx, const string& oid, const cls_rgw_obj_key& key, bufferlist& olh_tag,
                             bool delete_marker, const string& op_tag, struct rgw_bucket_dir_entry_meta *meta,
-                            uint64_t olh_epoch, bool log_op)
+                            uint64_t olh_epoch, time_t unmod_since, bool log_op)
 {
   bufferlist in, out;
   struct rgw_cls_link_olh_op call;
@@ -319,6 +319,7 @@ int cls_rgw_bucket_link_olh(librados::IoCtx& io_ctx, const string& oid, const cl
   }
   call.olh_epoch = olh_epoch;
   call.log_op = log_op;
+  call.unmod_since = unmod_since;
   ::encode(call, in);
   int r = io_ctx.exec(oid, "rgw", "bucket_link_olh", in, out);
   if (r < 0)
diff --git a/src/cls/rgw/cls_rgw_client.h b/src/cls/rgw/cls_rgw_client.h
index 6d136a9202a..e1d1342e3e4 100644
--- a/src/cls/rgw/cls_rgw_client.h
+++ b/src/cls/rgw/cls_rgw_client.h
@@ -330,7 +330,7 @@ int cls_rgw_bi_list(librados::IoCtx& io_ctx, const string oid,
 
 int cls_rgw_bucket_link_olh(librados::IoCtx& io_ctx, const string& oid, const cls_rgw_obj_key& key, bufferlist& olh_tag,
                             bool delete_marker, const string& op_tag, struct rgw_bucket_dir_entry_meta *meta,
-                            uint64_t olh_epoch, bool log_op);
+                            uint64_t olh_epoch, time_t unmod_since, 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);
 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,
diff --git a/src/cls/rgw/cls_rgw_ops.cc b/src/cls/rgw/cls_rgw_ops.cc
index 23beb4ee429..38f57e0c134 100644
--- a/src/cls/rgw/cls_rgw_ops.cc
+++ b/src/cls/rgw/cls_rgw_ops.cc
@@ -175,6 +175,7 @@ void rgw_cls_link_olh_op::dump(Formatter *f) const
   ::encode_json("olh_epoch", olh_epoch, f);
   ::encode_json("log_op", log_op, f);
   ::encode_json("bilog_flags", (uint32_t)bilog_flags, f);
+  ::encode_json("unmod_since", unmod_since, f);
 }
 
 void rgw_cls_unlink_instance_op::generate_test_instances(list<rgw_cls_unlink_instance_op*>& o)
diff --git a/src/cls/rgw/cls_rgw_ops.h b/src/cls/rgw/cls_rgw_ops.h
index 1c8f96bff24..053d716e292 100644
--- a/src/cls/rgw/cls_rgw_ops.h
+++ b/src/cls/rgw/cls_rgw_ops.h
@@ -164,11 +164,12 @@ struct rgw_cls_link_olh_op {
   uint64_t olh_epoch;
   bool log_op;
   uint16_t bilog_flags;
+  time_t unmod_since; /* only create delete marker if newer then this */
 
   rgw_cls_link_olh_op() : delete_marker(false), 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(olh_tag, bl);
     ::encode(delete_marker, bl);
@@ -177,11 +178,12 @@ struct rgw_cls_link_olh_op {
     ::encode(olh_epoch, bl);
     ::encode(log_op, bl);
     ::encode(bilog_flags, bl);
+    ::encode(unmod_since, bl);
     ENCODE_FINISH(bl);
   }
 
   void decode(bufferlist::iterator& bl) {
-    DECODE_START(1, bl);
+    DECODE_START(2, bl);
     ::decode(key, bl);
     ::decode(olh_tag, bl);
     ::decode(delete_marker, bl);
@@ -190,6 +192,9 @@ struct rgw_cls_link_olh_op {
     ::decode(olh_epoch, bl);
     ::decode(log_op, bl);
     ::decode(bilog_flags, bl);
+    if (struct_v >= 2) {
+      ::decode(unmod_since, bl);
+    }
     DECODE_FINISH(bl);
   }
 
diff --git a/src/cls/rgw/cls_rgw_types.h b/src/cls/rgw/cls_rgw_types.h
index 404aac65039..3da746851b7 100644
--- a/src/cls/rgw/cls_rgw_types.h
+++ b/src/cls/rgw/cls_rgw_types.h
@@ -528,6 +528,10 @@ struct rgw_bi_log_entry {
   void dump(Formatter *f) const;
   void decode_json(JSONObj *obj);
   static void generate_test_instances(list<rgw_bi_log_entry*>& o);
+  
+  bool is_versioned() {
+    return ((bilog_flags & RGW_BILOG_FLAG_VERSIONED_OP) != 0);
+  }
 };
 WRITE_CLASS_ENCODER(rgw_bi_log_entry)