rgw, cls_rgw: trim olh log functionality

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

4 files changed, 89 insertions, 2 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;