cls_rgw: unlink object instance

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 b31845a1bf555249c7146cb3fc3a7eac1fa24a85..c9f4aa683a764b0fe2b107b2f6b600c40ebaf33c 100644 (file)
--- a/src/cls/rgw/cls_rgw.cc
+++ b/src/cls/rgw/cls_rgw.cc
@@ -29,6 +29,7 @@ cls_method_handle_t h_rgw_bucket_rebuild_index;
 cls_method_handle_t h_rgw_bucket_prepare_op;
 cls_method_handle_t h_rgw_bucket_complete_op;
 cls_method_handle_t h_rgw_bucket_link_olh;
+cls_method_handle_t h_rgw_bucket_unlink_instance_op;
 cls_method_handle_t h_rgw_bucket_read_olh_log;
 cls_method_handle_t h_rgw_bucket_trim_olh_log;
 cls_method_handle_t h_rgw_obj_remove;
@@ -942,7 +943,9 @@ static int read_olh(cls_method_context_t hctx,cls_rgw_obj_key& obj_key, struct r
     CLS_LOG(0, "ERROR: read_index_entry() olh_key=%s ret=%d", olh_key.name.c_str(), ret);
     return ret;
   }
-  *found = (ret != -ENOENT);
+  if (found) {
+    *found = (ret != -ENOENT);
+  }
   return 0;
 }
 
@@ -1032,15 +1035,13 @@ public:
 
 
   int unlink_list_entry() {
-    if (instance_entry.ver.epoch > 0) {
-      string list_idx;
-      /* this instance has a previous list entry, remove that entry */
-      get_list_index_key(instance_entry, &list_idx);
-      int ret = cls_cxx_map_remove_key(hctx, list_idx);
-      if (ret < 0) {
-        CLS_LOG(0, "ERROR: cls_cxx_map_remove_key() list_idx=%s ret=%d", list_idx.c_str(), ret);
-        return ret;
-      }
+    string list_idx;
+    /* this instance has a previous list entry, remove that entry */
+    get_list_index_key(instance_entry, &list_idx);
+    int ret = cls_cxx_map_remove_key(hctx, list_idx);
+    if (ret < 0) {
+      CLS_LOG(0, "ERROR: cls_cxx_map_remove_key() list_idx=%s ret=%d", list_idx.c_str(), ret);
+      return ret;
     }
     return 0;
   }
@@ -1081,6 +1082,48 @@ public:
   int demote_current() {
     return write_entries(0, RGW_BUCKET_DIRENT_FLAG_CURRENT);
   }
+
+  bool is_delete_marker() {
+    return instance_entry.is_delete_marker();
+  }
+
+  int find_next_key(cls_rgw_obj_key *next_key, bool *found) {
+    string list_idx;
+    /* this instance has a previous list entry, remove that entry */
+    get_list_index_key(instance_entry, &list_idx);
+    /* this is the current head, need to update! */
+    map<string, bufferlist> keys;
+    string filter = key.name; /* list key starts with key name, filter it to avoid a case where we cross to
+                                 different namespace */
+    int ret = cls_cxx_map_get_vals(hctx, list_idx, filter, 1, &keys);
+    if (ret < 0) {
+      return ret;
+    }
+
+    if (keys.size() < 1) {
+      *found = false;
+      return 0;
+    }
+
+    rgw_bucket_dir_entry next_entry;
+
+    map<string, bufferlist>::reverse_iterator last = keys.rbegin();
+    try {
+      bufferlist::iterator iter = last->second.begin();
+      ::decode(next_entry, iter);
+    } catch (buffer::error& err) {
+      CLS_LOG(0, "ERROR; failed to decode entry: %s", last->first.c_str());
+      return -EIO;
+    }
+
+    *found = (key.name == next_entry.key.name);
+    if (*found) {
+      *next_key = next_entry.key;
+    }
+
+    return 0;
+  }
+
 };
 
 
@@ -1119,10 +1162,12 @@ public:
     return olh_data_entry;
   }
 
-  int update(cls_rgw_obj_key& key, bool delete_marker) {
+  void update(cls_rgw_obj_key& key, bool delete_marker) {
     olh_data_entry.delete_marker = delete_marker;
     olh_data_entry.key = key;
+  }
 
+  int write() {
     /* write the olh data entry */
     int ret = write_entry(hctx, olh_data_entry, olh_data_idx);
     if (ret < 0) {
@@ -1156,7 +1201,6 @@ static int rgw_bucket_link_olh(cls_method_context_t hctx, bufferlist *in, buffer
 {
   string olh_data_idx;
   string instance_idx;
-  struct rgw_bucket_olh_entry olh_data_entry;
 
   // decode request
   rgw_cls_link_olh_op op;
@@ -1221,7 +1265,12 @@ static int rgw_bucket_link_olh(cls_method_context_t hctx, bufferlist *in, buffer
     }
   }
 
-  ret = olh.update(op.key, op.delete_marker);
+  /* update the olh log */
+  olh.update_log(CLS_RGW_OLH_OP_LINK_OLH, op.op_tag, op.key, op.delete_marker);
+
+  olh.update(op.key, op.delete_marker);
+
+  ret = olh.write();
   if (ret < 0) {
     CLS_LOG(0, "ERROR: failed to update olh ret=%d", ret);
     return ret;
@@ -1233,9 +1282,79 @@ static int rgw_bucket_link_olh(cls_method_context_t hctx, bufferlist *in, buffer
     return ret;
   }
 
-  /* update the olh log */
+  return 0;
+}
 
-  olh.update_log(CLS_RGW_OLH_OP_LINK_OLH, op.op_tag, op.key, op.delete_marker);
+static int rgw_bucket_unlink_instance(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
+{
+  string olh_data_idx;
+  string instance_idx;
+
+  // decode request
+  rgw_cls_unlink_instance_op op;
+  bufferlist::iterator iter = in->begin();
+  try {
+    ::decode(op, iter);
+  } catch (buffer::error& err) {
+    CLS_LOG(0, "ERROR: rgw_bucket_rm_obj_instance_op(): failed to decode request\n");
+    return -EINVAL;
+  }
+
+  BIVerObjEntry obj(hctx, op.key);
+  BIOLHEntry olh(hctx, op.key);
+
+  int ret = obj.init();
+  if (ret == -ENOENT) {
+    return 0; /* already removed */
+  }
+  if (ret < 0) {
+    CLS_LOG(0, "ERROR: obj.init() returned ret=%d", ret);
+    return ret;
+  }
+
+  ret = olh.init(NULL);
+  if (ret < 0) {
+    CLS_LOG(0, "ERROR: olh.init() returned ret=%d", ret);
+    return ret;
+  }
+
+  if (olh.get_entry().key == op.key) {
+    /* this is the current head, need to update! */
+    cls_rgw_obj_key next_key;
+    bool found;
+    ret = obj.find_next_key(&next_key, &found);
+    if (ret < 0) {
+      CLS_LOG(0, "ERROR: obj.find_next_key() returned ret=%d", ret);
+      return ret;
+    }
+
+    if (found) {
+      BIVerObjEntry next(hctx, next_key);
+      ret = next.set_current(olh.get_epoch());
+      if (ret < 0) {
+        CLS_LOG(0, "ERROR: next.set_current() returned ret=%d", ret);
+        return ret;
+      }
+
+      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());
+    }
+  }
+
+  if (!obj.is_delete_marker()) {
+    olh.update_log(CLS_RGW_OLH_OP_REMOVE_INSTANCE, op.op_tag, op.key, false);
+  }
+
+  ret = obj.unlink_list_entry();
+  if (ret < 0) {
+    return ret;
+  }
+
+  ret = olh.write();
+  if (ret < 0) {
+    return ret;
+  }
 
   return 0;
 }
@@ -2317,6 +2436,7 @@ void __cls_init()
   cls_register_cxx_method(h_class, "bucket_prepare_op", CLS_METHOD_RD | CLS_METHOD_WR, rgw_bucket_prepare_op, &h_rgw_bucket_prepare_op);
   cls_register_cxx_method(h_class, "bucket_complete_op", CLS_METHOD_RD | CLS_METHOD_WR, rgw_bucket_complete_op, &h_rgw_bucket_complete_op);
   cls_register_cxx_method(h_class, "bucket_link_olh", CLS_METHOD_RD | CLS_METHOD_WR, rgw_bucket_link_olh, &h_rgw_bucket_link_olh);
+  cls_register_cxx_method(h_class, "bucket_unlink_instance", CLS_METHOD_RD | CLS_METHOD_WR, rgw_bucket_unlink_instance, &h_rgw_bucket_unlink_instance_op);
   cls_register_cxx_method(h_class, "bucket_read_olh_log", CLS_METHOD_RD, rgw_bucket_read_olh_log, &h_rgw_bucket_read_olh_log);
   cls_register_cxx_method(h_class, "bucket_trim_olh_log", CLS_METHOD_RD | CLS_METHOD_WR, rgw_bucket_trim_olh_log, &h_rgw_bucket_trim_olh_log);
 

diff --git a/src/cls/rgw/cls_rgw_client.cc b/src/cls/rgw/cls_rgw_client.cc
index be9fe0e942eece4b6184ea22eb0dc4e0339b0902..9d337d1bbb13fc9e289a1749b1d52496d5768bf2 100644 (file)
--- a/src/cls/rgw/cls_rgw_client.cc
+++ b/src/cls/rgw/cls_rgw_client.cc
@@ -119,6 +119,21 @@ int cls_rgw_bucket_link_olh(librados::IoCtx& io_ctx, const string& oid, const cl
   return 0;
 }
 
+int cls_rgw_bucket_unlink_instance(librados::IoCtx& io_ctx, const string& oid,
+                                   const cls_rgw_obj_key& key, const string& op_tag)
+{
+  bufferlist in, out;
+  struct rgw_cls_unlink_instance_op call;
+  call.key = key;
+  call.op_tag = op_tag;
+  ::encode(call, in);
+  int r = io_ctx.exec(oid, "rgw", "bucket_unlink_instance", in, out);
+  if (r < 0)
+    return r;
+
+  return 0;
+}
+
 int cls_rgw_get_olh_log(IoCtx& io_ctx, string& oid, const cls_rgw_obj_key& olh, uint64_t ver_marker,
                         map<uint64_t, struct rgw_bucket_olh_log_entry> *log, bool *is_truncated)
 {

diff --git a/src/cls/rgw/cls_rgw_client.h b/src/cls/rgw/cls_rgw_client.h
index 1d9105321c11eda7e716ab553f126c0b3bbc30ef..6f80e6280d842f3b461468d31933194e6d3b7dc2 100644 (file)
--- a/src/cls/rgw/cls_rgw_client.h
+++ b/src/cls/rgw/cls_rgw_client.h
@@ -35,6 +35,7 @@ void cls_rgw_remove_obj(librados::ObjectWriteOperation& o, list<string>& keep_at
 
 int cls_rgw_bucket_link_olh(librados::IoCtx& io_ctx, const string& oid, const cls_rgw_obj_key& key,
                             bool delete_marker, const string& op_tag);
+int cls_rgw_bucket_unlink_instance(librados::IoCtx& io_ctx, const string& oid, const cls_rgw_obj_key& key, const string& op_tag);
 int cls_rgw_get_olh_log(librados::IoCtx& io_ctx, string& oid, const cls_rgw_obj_key& olh, uint64_t ver_marker,
                         map<uint64_t, struct rgw_bucket_olh_log_entry> *log, bool *is_truncated);
 int cls_rgw_trim_olh_log(librados::IoCtx& io_ctx, string& oid, const cls_rgw_obj_key& olh, uint64_t ver);

diff --git a/src/cls/rgw/cls_rgw_ops.h b/src/cls/rgw/cls_rgw_ops.h
index 3795845586b7cc73be1c605d2f8094d092c3fb51..a80b499db6d1cc6bba352d37a9697dc10cee9e6e 100644 (file)
--- a/src/cls/rgw/cls_rgw_ops.h
+++ b/src/cls/rgw/cls_rgw_ops.h
@@ -173,6 +173,31 @@ struct rgw_cls_link_olh_op {
 };
 WRITE_CLASS_ENCODER(rgw_cls_link_olh_op)
 
+struct rgw_cls_unlink_instance_op {
+  cls_rgw_obj_key key;
+  string op_tag;
+
+  rgw_cls_unlink_instance_op() {}
+
+  void encode(bufferlist& bl) const {
+    ENCODE_START(1, 1, bl);
+    ::encode(key, bl);
+    ::encode(op_tag, bl);
+    ENCODE_FINISH(bl);
+  }
+
+  void decode(bufferlist::iterator& bl) {
+    DECODE_START(1, bl);
+    ::decode(key, bl);
+    ::decode(op_tag, bl);
+    DECODE_FINISH(bl);
+  }
+
+  void dump(Formatter *f) const;
+#warning implement me
+};
+WRITE_CLASS_ENCODER(rgw_cls_unlink_instance_op)
+
 struct rgw_cls_read_olh_log_op
 {
   cls_rgw_obj_key olh;