rgw, cls_rgw: trim olh log functionality

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

6 files changed, 132 insertions, 5 deletions

diff --git a/src/cls/rgw/cls_rgw.cc b/src/cls/rgw/cls_rgw.cc
index 3320b992619..94ca47449fe 100644
--- a/src/cls/rgw/cls_rgw.cc
+++ b/src/cls/rgw/cls_rgw.cc
@@ -30,6 +30,7 @@ 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_read_olh_log;
+cls_method_handle_t h_rgw_bucket_trim_olh_log;
 cls_method_handle_t h_rgw_bi_log_list_op;
 cls_method_handle_t h_rgw_dir_suggest_changes;
 cls_method_handle_t h_rgw_user_usage_log_add;
@@ -1103,6 +1104,52 @@ static int rgw_bucket_read_olh_log(cls_method_context_t hctx, bufferlist *in, bu
   return 0;
 }
 
+static int rgw_bucket_trim_olh_log(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
+{
+  // decode request
+  rgw_cls_trim_olh_log_op op;
+  bufferlist::iterator iter = in->begin();
+  try {
+    ::decode(op, iter);
+  } catch (buffer::error& err) {
+    CLS_LOG(0, "ERROR: rgw_bucket_trim_olh_log(): failed to decode request\n");
+    return -EINVAL;
+  }
+
+  if (!op.olh.instance.empty()) {
+    CLS_LOG(1, "bad key passed in (non empty instance)");
+    return -EINVAL;
+  }
+
+  /* read olh entry */
+  struct rgw_bucket_olh_entry olh_data_entry;
+  string olh_data_key;
+  encode_olh_data_key(op.olh, &olh_data_key);
+  int ret = read_index_entry(hctx, olh_data_key, &olh_data_entry);
+  if (ret < 0 && ret != -ENOENT) {
+    CLS_LOG(0, "ERROR: read_index_entry() olh_key=%s ret=%d", olh_data_key.c_str(), ret);
+    return ret;
+  }
+
+  /* remove all versions up to and including ver from the pending map */
+  map<uint64_t, rgw_bucket_olh_log_entry>& log = olh_data_entry.pending_log;
+  map<uint64_t, rgw_bucket_olh_log_entry>::iterator liter = log.begin();
+  while (liter != log.end() && liter->first <= op.ver) {
+    map<uint64_t, rgw_bucket_olh_log_entry>::iterator rm_iter = liter;
+    ++liter;
+    log.erase(rm_iter);
+  }
+
+  /* write the olh data entry */
+  ret = write_entry(hctx, olh_data_entry, olh_data_key);
+  if (ret < 0) {
+    CLS_LOG(0, "ERROR: write_entry() olh_key=%s ret=%d", olh_data_key.c_str(), ret);
+    return ret;
+  }
+
+  return 0;
+}
+
 int rgw_dir_suggest_changes(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
 {
   CLS_LOG(1, "rgw_dir_suggest_changes()");
@@ -2030,6 +2077,7 @@ void __cls_init()
   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_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, rgw_bucket_trim_olh_log, &h_rgw_bucket_trim_olh_log);
 
   cls_register_cxx_method(h_class, "bi_log_list", CLS_METHOD_RD, rgw_bi_log_list, &h_rgw_bi_log_list_op);
   cls_register_cxx_method(h_class, "bi_log_trim", CLS_METHOD_RD | CLS_METHOD_WR, rgw_bi_log_trim, &h_rgw_bi_log_list_op);
diff --git a/src/cls/rgw/cls_rgw_client.cc b/src/cls/rgw/cls_rgw_client.cc
index e9c4b64a6cb..e7eeec0b578 100644
--- a/src/cls/rgw/cls_rgw_client.cc
+++ b/src/cls/rgw/cls_rgw_client.cc
@@ -139,6 +139,20 @@ int cls_rgw_get_olh_log(IoCtx& io_ctx, string& oid, const cls_rgw_obj_key& olh,
  return r;
 }
 
+int cls_rgw_trim_olh_log(IoCtx& io_ctx, string& oid, const cls_rgw_obj_key& olh, uint64_t ver)
+{
+  bufferlist in, out;
+  struct rgw_cls_trim_olh_log_op call;
+  call.olh = olh;
+  call.ver = ver;
+  ::encode(call, in);
+  int r = io_ctx.exec(oid, "rgw", "bucket_trim_olh_log", in, out);
+  if (r < 0)
+    return r;
+
+ return 0;
+}
+
 int cls_rgw_bucket_check_index_op(IoCtx& io_ctx, string& oid,
 				  rgw_bucket_dir_header *existing_header,
 				  rgw_bucket_dir_header *calculated_header)
diff --git a/src/cls/rgw/cls_rgw_client.h b/src/cls/rgw/cls_rgw_client.h
index 2f67c4b862c..317511fe004 100644
--- a/src/cls/rgw/cls_rgw_client.h
+++ b/src/cls/rgw/cls_rgw_client.h
@@ -35,6 +35,7 @@ int cls_rgw_bucket_link_olh(librados::IoCtx& io_ctx, const string& oid, const cl
                             bool delete_marker, 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);
 
 int cls_rgw_bucket_check_index_op(librados::IoCtx& io_ctx, string& oid,
 				  rgw_bucket_dir_header *existing_header,
diff --git a/src/cls/rgw/cls_rgw_ops.h b/src/cls/rgw/cls_rgw_ops.h
index 59476996f06..fc662b68c17 100644
--- a/src/cls/rgw/cls_rgw_ops.h
+++ b/src/cls/rgw/cls_rgw_ops.h
@@ -206,13 +206,13 @@ struct rgw_cls_read_olh_log_ret
   rgw_cls_read_olh_log_ret() : is_truncated(false) {}
 
   void encode(bufferlist &bl) const {
-    ENCODE_START(2, 2, bl);
+    ENCODE_START(1, 1, bl);
     ::encode(log, bl);
     ::encode(is_truncated, bl);
     ENCODE_FINISH(bl);
   }
   void decode(bufferlist::iterator &bl) {
-    DECODE_START_LEGACY_COMPAT_LEN(2, 2, 2, bl);
+    DECODE_START(1, bl);
     ::decode(log, bl);
     ::decode(is_truncated, bl);
     DECODE_FINISH(bl);
@@ -222,6 +222,30 @@ struct rgw_cls_read_olh_log_ret
 };
 WRITE_CLASS_ENCODER(rgw_cls_read_olh_log_ret)
 
+struct rgw_cls_trim_olh_log_op
+{
+  cls_rgw_obj_key olh;
+  uint64_t ver;
+
+  rgw_cls_trim_olh_log_op() : ver(0) {}
+
+  void encode(bufferlist &bl) const {
+    ENCODE_START(1, 1, bl);
+    ::encode(olh, bl);
+    ::encode(ver, bl);
+    ENCODE_FINISH(bl);
+  }
+  void decode(bufferlist::iterator &bl) {
+    DECODE_START(1, bl);
+    ::decode(olh, bl);
+    ::decode(ver, bl);
+    DECODE_FINISH(bl);
+  }
+  void dump(Formatter *f) const;
+#warning implement me
+};
+WRITE_CLASS_ENCODER(rgw_cls_trim_olh_log_op)
+
 struct rgw_cls_list_op
 {
   cls_rgw_obj_key start_obj;
diff --git a/src/rgw/rgw_rados.cc b/src/rgw/rgw_rados.cc
index ed37b0dca1c..b8fbd6f9629 100644
--- a/src/rgw/rgw_rados.cc
+++ b/src/rgw/rgw_rados.cc
@@ -5304,7 +5304,7 @@ int RGWRados::bucket_index_read_olh_log(rgw_obj& obj_instance, uint64_t ver_mark
     return ret;
   }
 
-  cls_rgw_obj_key key(obj_instance.get_index_key_name(), obj_instance.get_instance());
+  cls_rgw_obj_key key(obj_instance.get_index_key_name(), string());
   ret = cls_rgw_get_olh_log(index_ctx, oid, key, ver_marker, log, is_truncated);
   if (ret < 0) {
     return ret;
@@ -5313,6 +5313,32 @@ int RGWRados::bucket_index_read_olh_log(rgw_obj& obj_instance, uint64_t ver_mark
   return 0;
 }
 
+int RGWRados::bucket_index_trim_olh_log(rgw_obj& obj_instance, uint64_t ver)
+{
+  rgw_rados_ref ref;
+  rgw_bucket bucket;
+  int r = get_obj_ref(obj_instance, &ref, &bucket);
+  if (r < 0) {
+    return r;
+  }
+
+  librados::IoCtx index_ctx;
+  string oid;
+
+  int ret = open_bucket_index(bucket, index_ctx, oid);
+  if (ret < 0) {
+    return ret;
+  }
+
+  cls_rgw_obj_key key(obj_instance.get_index_key_name(), string());
+  ret = cls_rgw_trim_olh_log(index_ctx, oid, key, ver);
+  if (ret < 0) {
+    return ret;
+  }
+
+  return 0;
+}
+
 static void op_setxattr(librados::ObjectWriteOperation& op, const char *name, const string& val)
 {
   bufferlist bl;
@@ -5321,7 +5347,8 @@ static void op_setxattr(librados::ObjectWriteOperation& op, const char *name, co
 }
 
 int RGWRados::apply_olh_log(void *ctx, const string& bucket_owner, rgw_obj& obj,
-                            const string& obj_tag, map<uint64_t, rgw_bucket_olh_log_entry>& log)
+                            const string& obj_tag, map<uint64_t, rgw_bucket_olh_log_entry>& log,
+                            uint64_t *plast_ver)
 {
   if (log.empty()) {
     return 0;
@@ -5330,6 +5357,7 @@ int RGWRados::apply_olh_log(void *ctx, const string& bucket_owner, rgw_obj& obj,
   librados::ObjectWriteOperation op;
 
   uint64_t last_ver = log.rbegin()->first;
+  *plast_ver = last_ver;
 
   map<uint64_t, rgw_bucket_olh_log_entry>::iterator iter = log.begin();
 
@@ -5396,6 +5424,15 @@ int RGWRados::apply_olh_log(void *ctx, const string& bucket_owner, rgw_obj& obj,
 
   /* update olh object */
   r = ref.ioctx.operate(ref.oid, &op);
+  if (r < 0) {
+    ldout(cct, 0) << "ERROR: could not apply olh update, r=" << r << dendl;
+    return r;
+  }
+
+  r = bucket_index_trim_olh_log(obj, last_ver);
+  if (r < 0) {
+    ldout(cct, 0) << "ERROR: could not trim olh log, r=" << r << dendl;
+  }
   return r;
 }
 
diff --git a/src/rgw/rgw_rados.h b/src/rgw/rgw_rados.h
index 1af9240d292..70eb026ab16 100644
--- a/src/rgw/rgw_rados.h
+++ b/src/rgw/rgw_rados.h
@@ -1753,8 +1753,11 @@ public:
   int bucket_index_link_olh(rgw_obj& obj_instance, bool delete_marker, const string& op_tag);
   int bucket_index_read_olh_log(rgw_obj& obj_instance, uint64_t ver_marker,
                                 map<uint64_t, rgw_bucket_olh_log_entry> *log, bool *is_truncated);
+  int bucket_index_trim_olh_log(rgw_obj& obj_instance, uint64_t ver);
   int apply_olh_log(void *ctx, const string& bucket_owner, rgw_obj& obj,
-                    const string& obj_tag, map<uint64_t, rgw_bucket_olh_log_entry>& log);
+                    const string& obj_tag, map<uint64_t, rgw_bucket_olh_log_entry>& log,
+                    uint64_t *plast_ver);
+  int update_olh(void *ctx, const string& bucket_owner, rgw_obj& obj, const string& obj_tag);
 
   int follow_olh(map<string, bufferlist>& attrset, rgw_obj& target);
   int get_olh(rgw_obj& obj, RGWOLHInfo *olh);