cls/rgw: log owner, display name when creating delete marker

In the case of delete marker, we need to log the owner and the owner
display name so that we can create it correctly when syncing to a
different zone. We don't need it in other cases so we don't store it.

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 d109819489c07aeb76adc3c1d69d61c8284ed350..50faf0086c293b7b766cd1bd4f401389a0397b7f 100644 (file)
--- a/src/cls/rgw/cls_rgw.cc
+++ b/src/cls/rgw/cls_rgw.cc
@@ -142,7 +142,7 @@ static void bi_log_index_key(cls_method_context_t hctx, string& key, string& id,
 static int log_index_operation(cls_method_context_t hctx, cls_rgw_obj_key& obj_key, RGWModifyOp op,
                                string& tag, utime_t& timestamp,
                                rgw_bucket_entry_ver& ver, RGWPendingState state, uint64_t index_ver,
-                               string& max_marker, uint16_t bilog_flags)
+                               string& max_marker, uint16_t bilog_flags, string *owner, string *owner_display_name)
 {
   bufferlist bl;
 
@@ -157,6 +157,12 @@ static int log_index_operation(cls_method_context_t hctx, cls_rgw_obj_key& obj_k
   entry.index_ver = index_ver;
   entry.tag = tag;
   entry.bilog_flags = bilog_flags;
+  if (owner) {
+    entry.owner = *owner;
+  }
+  if (owner_display_name) {
+    entry.owner_display_name = *owner_display_name;
+  }
 
   string key;
   bi_log_index_key(hctx, key, entry.id, index_ver);
@@ -686,7 +692,7 @@ int rgw_bucket_prepare_op(cls_method_context_t hctx, bufferlist *in, bufferlist
 
   if (op.log_op) {
     rc = log_index_operation(hctx, op.key, op.op, op.tag, entry.meta.mtime,
-                             entry.ver, info.state, header.ver, header.max_marker, op.bilog_flags);
+                             entry.ver, info.state, header.ver, header.max_marker, op.bilog_flags, NULL, NULL);
     if (rc < 0)
       return rc;
   }
@@ -842,7 +848,7 @@ int rgw_bucket_complete_op(cls_method_context_t hctx, bufferlist *in, bufferlist
   if (cancel) {
     if (op.log_op) {
       rc = log_index_operation(hctx, op.key, op.op, op.tag, entry.meta.mtime, entry.ver,
-                               CLS_RGW_STATE_COMPLETE, header.ver, header.max_marker, op.bilog_flags);
+                               CLS_RGW_STATE_COMPLETE, header.ver, header.max_marker, op.bilog_flags, NULL, NULL);
       if (rc < 0)
         return rc;
     }
@@ -902,7 +908,7 @@ int rgw_bucket_complete_op(cls_method_context_t hctx, bufferlist *in, bufferlist
 
   if (op.log_op) {
     rc = log_index_operation(hctx, op.key, op.op, op.tag, entry.meta.mtime, entry.ver,
-                             CLS_RGW_STATE_COMPLETE, header.ver, header.max_marker, op.bilog_flags);
+                             CLS_RGW_STATE_COMPLETE, header.ver, header.max_marker, op.bilog_flags, NULL, NULL);
     if (rc < 0)
       return rc;
   }
@@ -927,7 +933,7 @@ int rgw_bucket_complete_op(cls_method_context_t hctx, bufferlist *in, bufferlist
 
     if (op.log_op) {
       rc = log_index_operation(hctx, remove_key, CLS_RGW_OP_DEL, op.tag, remove_entry.meta.mtime,
-                               remove_entry.ver, CLS_RGW_STATE_COMPLETE, header.ver, header.max_marker, op.bilog_flags);
+                               remove_entry.ver, CLS_RGW_STATE_COMPLETE, header.ver, header.max_marker, op.bilog_flags, NULL, NULL);
       if (rc < 0)
         continue;
     }
@@ -1500,10 +1506,19 @@ static int rgw_bucket_link_olh(cls_method_context_t hctx, bufferlist *in, buffer
     rgw_bucket_entry_ver ver;
     ver.epoch = (op.olh_epoch ? op.olh_epoch : olh.get_epoch());
 
+    string *powner = NULL;
+    string *powner_display_name = NULL;
+
+    if (op.delete_marker) {
+      powner = &entry.meta.owner;
+      powner_display_name = &entry.meta.owner_display_name;
+    }
+
     RGWModifyOp operation = (op.delete_marker ? CLS_RGW_OP_LINK_OLH_DM : CLS_RGW_OP_LINK_OLH);
     ret = log_index_operation(hctx, op.key, operation, op.op_tag,
                               entry.meta.mtime, ver,
-                              CLS_RGW_STATE_COMPLETE, header.ver, header.max_marker, op.bilog_flags | RGW_BILOG_FLAG_VERSIONED_OP);
+                              CLS_RGW_STATE_COMPLETE, header.ver, header.max_marker, op.bilog_flags | RGW_BILOG_FLAG_VERSIONED_OP,
+                              powner, powner_display_name);
     if (ret < 0)
       return ret;
   }
@@ -1634,7 +1649,7 @@ static int rgw_bucket_unlink_instance(cls_method_context_t hctx, bufferlist *in,
     ret = log_index_operation(hctx, op.key, CLS_RGW_OP_UNLINK_INSTANCE, op.op_tag,
                               mtime, ver,
                               CLS_RGW_STATE_COMPLETE, header.ver, header.max_marker,
-                              op.bilog_flags | RGW_BILOG_FLAG_VERSIONED_OP);
+                              op.bilog_flags | RGW_BILOG_FLAG_VERSIONED_OP, NULL, NULL);
     if (ret < 0)
       return ret;
   }

diff --git a/src/cls/rgw/cls_rgw_types.cc b/src/cls/rgw/cls_rgw_types.cc
index 7d92304e8b60f3c4eb927c396c882ec20a747c58..e14abde5c4b766b8bba8dd334e5ea288f95326d2 100644 (file)
--- a/src/cls/rgw/cls_rgw_types.cc
+++ b/src/cls/rgw/cls_rgw_types.cc
@@ -348,6 +348,8 @@ void rgw_bi_log_entry::decode_json(JSONObj *obj)
   JSONDecoder::decode_json("bilog_flags", f, obj);
   JSONDecoder::decode_json("ver", ver, obj);
   bilog_flags = (uint16_t)f;
+  JSONDecoder::decode_json("owner", owner, obj);
+  JSONDecoder::decode_json("owner_display_name", owner_display_name, obj);
 }
 
 void rgw_bi_log_entry::dump(Formatter *f) const
@@ -403,6 +405,8 @@ void rgw_bi_log_entry::dump(Formatter *f) const
   f->close_section();
   f->dump_int("bilog_flags", bilog_flags);
   f->dump_bool("versioned", (bilog_flags & RGW_BILOG_FLAG_VERSIONED_OP) != 0);
+  f->dump_string("owner", owner);
+  f->dump_string("owner_display_name", owner_display_name);
 }
 
 void rgw_bi_log_entry::generate_test_instances(list<rgw_bi_log_entry*>& ls)

diff --git a/src/cls/rgw/cls_rgw_types.h b/src/cls/rgw/cls_rgw_types.h
index 98fb6db70b341782210bf349f830fadcb59fbdf1..404aac650398d48a6cc423425edf658d16796e1f 100644 (file)
--- a/src/cls/rgw/cls_rgw_types.h
+++ b/src/cls/rgw/cls_rgw_types.h
@@ -479,11 +479,13 @@ struct rgw_bi_log_entry {
   uint64_t index_ver;
   string tag;
   uint16_t bilog_flags;
+  string owner; /* only being set if it's a delete marker */
+  string owner_display_name; /* only being set if it's a delete marker */
 
   rgw_bi_log_entry() : op(CLS_RGW_OP_UNKNOWN), state(CLS_RGW_STATE_PENDING_MODIFY), index_ver(0), bilog_flags(0) {}
 
   void encode(bufferlist &bl) const {
-    ENCODE_START(2, 1, bl);
+    ENCODE_START(3, 1, bl);
     ::encode(id, bl);
     ::encode(object, bl);
     ::encode(timestamp, bl);
@@ -496,6 +498,8 @@ struct rgw_bi_log_entry {
     encode_packed_val(index_ver, bl);
     ::encode(instance, bl);
     ::encode(bilog_flags, bl);
+    ::encode(owner, bl);
+    ::encode(owner_display_name, bl);
     ENCODE_FINISH(bl);
   }
   void decode(bufferlist::iterator &bl) {
@@ -515,6 +519,10 @@ struct rgw_bi_log_entry {
       ::decode(instance, bl);
       ::decode(bilog_flags, bl);
     }
+    if (struct_v >= 3) {
+      ::decode(owner, bl);
+      ::decode(owner_display_name, bl);
+    }
     DECODE_FINISH(bl);
   }
   void dump(Formatter *f) const;