rgw, cls_rgw: fix update of olh to reflect non existing object

need to update olh appropriately when removing the last reference of an
object.

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 4e5e26ea42a2b853b014f1116e2ecbbc3564b1b6..1f669eb01ac7efff526aed8bf5b53e86a3410586 100644 (file)
--- a/src/cls/rgw/cls_rgw.cc
+++ b/src/cls/rgw/cls_rgw.cc
@@ -1235,6 +1235,11 @@ public:
     update_olh_log(olh_data_entry, op, op_tag, key, delete_marker);
   }
 
+  bool exists() { return olh_data_entry.exists; }
+
+  void set_exists(bool exists) {
+    olh_data_entry.exists = exists;
+  }
 };
 
 /*
@@ -1341,13 +1346,15 @@ static int rgw_bucket_link_olh(cls_method_context_t hctx, bufferlist *in, buffer
   }
 
   if (olh_found) {
-    /* found olh, previous instance is no longer the latest, need to update */
-    BIVerObjEntry old_obj(hctx, olh.get_entry().key);
+    if (olh.exists()) {
+      /* found olh, previous instance is no longer the latest, need to update */
+      BIVerObjEntry old_obj(hctx, olh.get_entry().key);
 
-    ret = old_obj.demote_current();
-    if (ret < 0) {
-      CLS_LOG(0, "ERROR: could not demote current on previous key ret=%d", ret);
-      return ret;
+      ret = old_obj.demote_current();
+      if (ret < 0) {
+        CLS_LOG(0, "ERROR: could not demote current on previous key ret=%d", ret);
+        return ret;
+      }
     }
   } else {
     cls_rgw_obj_key key(op.key.name);
@@ -1367,6 +1374,8 @@ static int rgw_bucket_link_olh(cls_method_context_t hctx, bufferlist *in, buffer
 
   olh.update(op.key, op.delete_marker);
 
+  olh.set_exists(true);
+
   ret = olh.write();
   if (ret < 0) {
     CLS_LOG(0, "ERROR: failed to update olh ret=%d", ret);
@@ -1446,8 +1455,12 @@ static int rgw_bucket_unlink_instance(cls_method_context_t hctx, bufferlist *in,
               next_key.name.c_str(), next_key.instance.c_str(), (int)next.is_delete_marker());
 
       olh.update(next_key, next.is_delete_marker());
-
       olh.update_log(CLS_RGW_OLH_OP_LINK_OLH, op.op_tag, next_key, next.is_delete_marker());
+    } else {
+      /* next_key is empty */
+      olh.update(next_key, false);
+      olh.update_log(CLS_RGW_OLH_OP_UNLINK_OLH, op.op_tag, next_key, false);
+      olh.set_exists(false);
     }
   }
 

diff --git a/src/cls/rgw/cls_rgw_types.cc b/src/cls/rgw/cls_rgw_types.cc
index 3dad42dd6a8c316cc87a4caa24d3a186baab0c88..74469bc6f83f04498ffecc53363e756f59f5f12b 100644 (file)
--- a/src/cls/rgw/cls_rgw_types.cc
+++ b/src/cls/rgw/cls_rgw_types.cc
@@ -228,6 +228,8 @@ void rgw_bucket_olh_entry::dump(Formatter *f) const
   encode_json("delete_marker", delete_marker, f);
   encode_json("epoch", epoch, f);
   encode_json("pending_log", pending_log, f);
+  encode_json("tag", tag, f);
+  encode_json("exists", exists, f);
 }
 
 void rgw_bucket_olh_log_entry::dump(Formatter *f) const

diff --git a/src/cls/rgw/cls_rgw_types.h b/src/cls/rgw/cls_rgw_types.h
index 01e3fd1e524a07fbd26b7f8574321cffcdc0a8fe..923c59c620b854b31040cf59ff67ebf27a562898 100644 (file)
--- a/src/cls/rgw/cls_rgw_types.h
+++ b/src/cls/rgw/cls_rgw_types.h
@@ -368,7 +368,8 @@ WRITE_CLASS_ENCODER(rgw_cls_bi_entry)
 
 enum OLHLogOp {
   CLS_RGW_OLH_OP_LINK_OLH        = 1,
-  CLS_RGW_OLH_OP_REMOVE_INSTANCE = 2,
+  CLS_RGW_OLH_OP_UNLINK_OLH      = 2, /* object does not exist */
+  CLS_RGW_OLH_OP_REMOVE_INSTANCE = 3,
 };
 
 struct rgw_bucket_olh_log_entry {
@@ -411,8 +412,9 @@ struct rgw_bucket_olh_entry {
   uint64_t epoch;
   map<uint64_t, struct rgw_bucket_olh_log_entry> pending_log;
   string tag;
+  bool exists;
 
-  rgw_bucket_olh_entry() : delete_marker(false), epoch(0) {}
+  rgw_bucket_olh_entry() : delete_marker(false), epoch(0), exists(false) {}
 
   void encode(bufferlist &bl) const {
     ENCODE_START(1, 1, bl);
@@ -421,6 +423,7 @@ struct rgw_bucket_olh_entry {
     ::encode(epoch, bl);
     ::encode(pending_log, bl);
     ::encode(tag, bl);
+    ::encode(exists, bl);
     ENCODE_FINISH(bl);
   }
   void decode(bufferlist::iterator &bl) {
@@ -430,6 +433,7 @@ struct rgw_bucket_olh_entry {
     ::decode(epoch, bl);
     ::decode(pending_log, bl);
     ::decode(tag, bl);
+    ::decode(exists, bl);
     DECODE_FINISH(bl);
   }
   void dump(Formatter *f) const;

diff --git a/src/rgw/rgw_rados.cc b/src/rgw/rgw_rados.cc
index fc41ac1e1dad8eb3a5cbaf218c1b72bfcc47d67a..0b1d772519a82b24751bf0fc78bc794052afa46a 100644 (file)
--- a/src/rgw/rgw_rados.cc
+++ b/src/rgw/rgw_rados.cc
@@ -5645,6 +5645,12 @@ int RGWRados::apply_olh_log(RGWObjectCtx& obj_ctx, const string& bucket_owner, r
       key = entry.key;
       delete_marker = entry.delete_marker;
       break;
+    case CLS_RGW_OLH_OP_UNLINK_OLH:
+      /* treat this as linking into a delete marker */
+      need_to_link = true;
+      key = entry.key;
+      delete_marker = true;
+      break;
     default:
       ldout(cct, 0) << "ERROR: apply_olh_log: invalid op: " << (int)entry.op << dendl;
       return -EIO;