rgw, cls_rgw: conditionally remove olh object

still not completely working, need to debug

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 7b613b964489ab2bdff1710681c3364468b6a175..c6cdbadf6782109e42cbcf37e3ecb61c92428622 100644 (file)
--- a/src/cls/rgw/cls_rgw.cc
+++ b/src/cls/rgw/cls_rgw.cc
@@ -35,6 +35,7 @@ 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_bucket_clear_olh;
 cls_method_handle_t h_rgw_obj_remove;
+cls_method_handle_t h_rgw_obj_check_attrs_prefix;
 cls_method_handle_t h_rgw_bi_get_op;
 cls_method_handle_t h_rgw_bi_put_op;
 cls_method_handle_t h_rgw_bi_list_op;
@@ -1941,6 +1942,49 @@ static int rgw_obj_remove(cls_method_context_t hctx, bufferlist *in, bufferlist
   return 0;
 }
 
+static int rgw_obj_check_attrs_prefix(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
+{
+  // decode request
+  rgw_cls_obj_check_attrs_prefix op;
+  bufferlist::iterator iter = in->begin();
+  try {
+    ::decode(op, iter);
+  } catch (buffer::error& err) {
+    CLS_LOG(0, "ERROR: %s(): failed to decode request", __func__);
+    return -EINVAL;
+  }
+
+  if (op.check_prefix.empty()) {
+    return -EINVAL;
+  }
+
+  map<string, bufferlist> attrset;
+  int ret = cls_cxx_getxattrs(hctx, &attrset);
+  if (ret < 0 && ret != -ENOENT) {
+    CLS_LOG(0, "ERROR: %s(): cls_cxx_getxattrs() returned %d", __func__, ret);
+    return ret;
+  }
+
+  bool exist = false;
+
+  for (map<string, bufferlist>::iterator aiter = attrset.lower_bound(op.check_prefix);
+       aiter != attrset.end(); ++aiter) {
+    const string& attr = aiter->first;
+
+    if (attr.substr(0, op.check_prefix.size()) > op.check_prefix) {
+      break;
+    }
+
+    exist = true;
+  }
+
+  if (exist == op.fail_if_exist) {
+    return -ECANCELED;
+  }
+
+  return 0;
+}
+
 static int rgw_bi_get_op(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
 {
   // decode request
@@ -2997,6 +3041,7 @@ void __cls_init()
   cls_register_cxx_method(h_class, "bucket_clear_olh", CLS_METHOD_RD | CLS_METHOD_WR, rgw_bucket_clear_olh, &h_rgw_bucket_clear_olh);
 
   cls_register_cxx_method(h_class, "obj_remove", CLS_METHOD_RD | CLS_METHOD_WR, rgw_obj_remove, &h_rgw_obj_remove);
+  cls_register_cxx_method(h_class, "obj_check_attrs_prefix", CLS_METHOD_RD, rgw_obj_check_attrs_prefix, &h_rgw_obj_check_attrs_prefix);
 
   cls_register_cxx_method(h_class, "bi_get", CLS_METHOD_RD, rgw_bi_get_op, &h_rgw_bi_get_op);
   cls_register_cxx_method(h_class, "bi_put", CLS_METHOD_RD | CLS_METHOD_WR, rgw_bi_put_op, &h_rgw_bi_put_op);

diff --git a/src/cls/rgw/cls_rgw_client.cc b/src/cls/rgw/cls_rgw_client.cc
index ab8b9a2981e6e402f74bcb7976ea1014ebd3b2df..be447172c2dcd3e465a53864c5671e0d1a735f47 100644 (file)
--- a/src/cls/rgw/cls_rgw_client.cc
+++ b/src/cls/rgw/cls_rgw_client.cc
@@ -103,6 +103,16 @@ void cls_rgw_remove_obj(librados::ObjectWriteOperation& o, list<string>& keep_at
   o.exec("rgw", "obj_remove", in);
 }
 
+void cls_rgw_obj_check_attrs_prefix(librados::ObjectOperation& o, const string& prefix, bool fail_if_exist)
+{
+  bufferlist in;
+  struct rgw_cls_obj_check_attrs_prefix call;
+  call.check_prefix = prefix;
+  call.fail_if_exist = fail_if_exist;
+  ::encode(call, in);
+  o.exec("rgw", "obj_check_attrs_prefix", in);
+}
+
 int cls_rgw_bi_get(librados::IoCtx& io_ctx, const string oid,
                    BIIndexType index_type, cls_rgw_obj_key& key,
                    rgw_cls_bi_entry *entry)

diff --git a/src/cls/rgw/cls_rgw_client.h b/src/cls/rgw/cls_rgw_client.h
index 4a445615748252c6fed92f4d636008585ac4de11..5af68855c0f287d4dd21ba4cf8642fe6d4a8fddd 100644 (file)
--- a/src/cls/rgw/cls_rgw_client.h
+++ b/src/cls/rgw/cls_rgw_client.h
@@ -32,6 +32,7 @@ int cls_rgw_list_op(librados::IoCtx& io_ctx, const string& oid,
                     rgw_bucket_dir *dir, bool *is_truncated);
 
 void cls_rgw_remove_obj(librados::ObjectWriteOperation& o, list<string>& keep_attr_prefixes);
+void cls_rgw_obj_check_attrs_prefix(librados::ObjectOperation& o, const string& prefix, bool fail_if_exist);
 
 int cls_rgw_bi_get(librados::IoCtx& io_ctx, const string oid,
                    BIIndexType index_type, cls_rgw_obj_key& key,

diff --git a/src/cls/rgw/cls_rgw_ops.h b/src/cls/rgw/cls_rgw_ops.h
index 7d50d42b1f2a77b90fe2c2df7cf571ade6631b0e..cd2467a07163d3d27337622816413baeac1b08a0 100644 (file)
--- a/src/cls/rgw/cls_rgw_ops.h
+++ b/src/cls/rgw/cls_rgw_ops.h
@@ -415,6 +415,28 @@ struct rgw_cls_obj_remove_op {
 };
 WRITE_CLASS_ENCODER(rgw_cls_obj_remove_op)
 
+struct rgw_cls_obj_check_attrs_prefix {
+  string check_prefix;
+  bool fail_if_exist;
+
+  rgw_cls_obj_check_attrs_prefix() : fail_if_exist(false) {}
+
+  void encode(bufferlist& bl) const {
+    ENCODE_START(1, 1, bl);
+    ::encode(check_prefix, bl);
+    ::encode(fail_if_exist, bl);
+    ENCODE_FINISH(bl);
+  }
+
+  void decode(bufferlist::iterator& bl) {
+    DECODE_START(1, bl);
+    ::decode(check_prefix, bl);
+    ::decode(fail_if_exist, bl);
+    DECODE_FINISH(bl);
+  }
+};
+WRITE_CLASS_ENCODER(rgw_cls_obj_check_attrs_prefix)
+
 struct rgw_cls_usage_log_add_op {
   rgw_usage_log_info info;
 

diff --git a/src/cls/rgw/cls_rgw_types.cc b/src/cls/rgw/cls_rgw_types.cc
index cf9df7db0bddc9cb291619d84c1e6376176b7863..6cc9c4d755cb35037183f288ba676eae18372d05 100644 (file)
--- a/src/cls/rgw/cls_rgw_types.cc
+++ b/src/cls/rgw/cls_rgw_types.cc
@@ -255,6 +255,9 @@ void rgw_bucket_olh_log_entry::dump(Formatter *f) const
     case CLS_RGW_OLH_OP_LINK_OLH:
       op_str = "link_olh";
       break;
+    case CLS_RGW_OLH_OP_UNLINK_OLH:
+      op_str = "unlink_olh";
+      break;
     case CLS_RGW_OLH_OP_REMOVE_INSTANCE:
       op_str = "remove_instance";
       break;

diff --git a/src/rgw/rgw_rados.cc b/src/rgw/rgw_rados.cc
index d2be902cbea1b22144b69887ebff03406c8b1c7e..2e904d3c830301219025502b38f66d1191dbdb54 100644 (file)
--- a/src/rgw/rgw_rados.cc
+++ b/src/rgw/rgw_rados.cc
@@ -3986,6 +3986,11 @@ void RGWRados::remove_rgw_head_obj(ObjectWriteOperation& op)
   cls_rgw_remove_obj(op, prefixes);
 }
 
+void RGWRados::cls_obj_check_prefix_exist(ObjectOperation& op, const string& prefix, bool fail_if_exist)
+{
+  cls_rgw_obj_check_attrs_prefix(op, prefix, fail_if_exist);
+}
+
 
 /**
  * Delete an object.
@@ -5839,9 +5844,7 @@ int RGWRados::apply_olh_log(RGWObjectCtx& obj_ctx, RGWObjState& state, RGWBucket
     return r;
   }
 
-  if (need_to_remove) {
-    op.remove();
-  } else if (need_to_link) {
+  if (need_to_link) {
     rgw_obj target(bucket, key.name);
     target.set_instance(key.instance);
     RGWOLHInfo info;
@@ -5867,7 +5870,7 @@ int RGWRados::apply_olh_log(RGWObjectCtx& obj_ctx, RGWObjState& state, RGWBucket
 
   /* update olh object */
   r = ref.ioctx.operate(ref.oid, &op);
-  if (need_to_remove && (r == -ENOENT || r == -ECANCELED)) {
+  if (r == -ECANCELED) {
     r = 0;
   }
   if (r < 0) {
@@ -5882,6 +5885,18 @@ int RGWRados::apply_olh_log(RGWObjectCtx& obj_ctx, RGWObjState& state, RGWBucket
   }
 
   if (need_to_remove) {
+    ObjectWriteOperation rm_op;
+
+    rm_op.cmpxattr(RGW_ATTR_OLH_ID_TAG, CEPH_OSD_CMPXATTR_OP_EQ, olh_tag);
+    rm_op.cmpxattr(RGW_ATTR_OLH_VER, CEPH_OSD_CMPXATTR_OP_GT, last_ver);
+    cls_obj_check_prefix_exist(rm_op, RGW_ATTR_OLH_PENDING_PREFIX, true); /* fail if found one of these, pending modification */
+    rm_op.remove();
+
+    r = ref.ioctx.operate(ref.oid, &rm_op);
+    if (r == -ECANCELED) {
+      return 0; /* someone else won this race */
+    }
+
     r = bucket_index_clear_olh(state, obj);
     if (r < 0) {
       ldout(cct, 0) << "ERROR: could not clear bucket index olh entries r=" << r << dendl;

diff --git a/src/rgw/rgw_rados.h b/src/rgw/rgw_rados.h
index f0b9436520a8f78b0c40f8f3e0d281ed1ab3aae6..eadd3ec36d242cb480511cf65260bcc2ccf6b5d5 100644 (file)
--- a/src/rgw/rgw_rados.h
+++ b/src/rgw/rgw_rados.h
@@ -1199,6 +1199,7 @@ class RGWRados
   int store_bucket_info(RGWBucketInfo& info, map<string, bufferlist> *pattrs, RGWObjVersionTracker *objv_tracker, bool exclusive);
 
   void remove_rgw_head_obj(librados::ObjectWriteOperation& op);
+  void cls_obj_check_prefix_exist(librados::ObjectOperation& op, const string& prefix, bool fail_if_exist);
 protected:
   CephContext *cct;
   librados::Rados *rados;