rgw: bucket index log trim

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

7 files changed, 165 insertions, 5 deletions

diff --git a/src/cls/rgw/cls_rgw.cc b/src/cls/rgw/cls_rgw.cc
index 461e40a662a..464eb26038b 100644
--- a/src/cls/rgw/cls_rgw.cc
+++ b/src/cls/rgw/cls_rgw.cc
@@ -737,7 +737,7 @@ int bi_log_record_decode(bufferlist& bl, rgw_bi_log_entry& e)
   return 0;
 }
 
-static int bi_log_iterate_entries(cls_method_context_t hctx, const string& marker,
+static int bi_log_iterate_entries(cls_method_context_t hctx, const string& marker, const string& end_marker,
                               string& key_iter, uint32_t max_entries, bool *truncated,
                               int (*cb)(cls_method_context_t, const string&, rgw_bi_log_entry&, void *),
                               void *param)
@@ -763,10 +763,16 @@ static int bi_log_iterate_entries(cls_method_context_t hctx, const string& marke
     start_key = key_iter;
   }
 
-  end_key = BI_PREFIX_CHAR;
-  end_key.append(bucket_index_prefixes[BI_BUCKET_LAST_INDEX]);
+  if (end_marker.empty()) {
+    end_key = BI_PREFIX_CHAR;
+    end_key.append(bucket_index_prefixes[BI_BUCKET_LAST_INDEX]);
+  } else {
+    end_key = BI_PREFIX_CHAR;
+    end_key.append(bucket_index_prefixes[BI_BUCKET_LOG_INDEX]);
+    end_key.append(end_marker);
+  }
 
-  CLS_LOG(0, "bi_log_iterate_entries end_key=%s\n", end_key.c_str());
+  CLS_LOG(0, "bi_log_iterate_entries start_key=%s end_key=%s\n", start_key.c_str(), end_key.c_str());
 
   string filter;
 
@@ -824,7 +830,8 @@ static int bi_log_list_entries(cls_method_context_t hctx, const string& marker,
 			   uint32_t max, list<rgw_bi_log_entry>& entries, bool *truncated)
 {
   string key_iter;
-  int ret = bi_log_iterate_entries(hctx, marker,
+  string end_marker;
+  int ret = bi_log_iterate_entries(hctx, marker, end_marker,
                               key_iter, max, truncated,
                               bi_log_list_cb, &entries);
   return ret;
@@ -852,6 +859,70 @@ static int rgw_bi_log_list(cls_method_context_t hctx, bufferlist *in, bufferlist
   return 0;
 }
 
+static int bi_log_list_trim_cb(cls_method_context_t hctx, const string& key, rgw_bi_log_entry& info, void *param)
+{
+  list<rgw_bi_log_entry> *entries = (list<rgw_bi_log_entry> *)param;
+
+  entries->push_back(info);
+  return 0;
+}
+
+static int bi_log_remove_entry(cls_method_context_t hctx, rgw_bi_log_entry& entry)
+{
+  string key;
+  key = BI_PREFIX_CHAR;
+  key.append(bucket_index_prefixes[BI_BUCKET_LOG_INDEX]);
+  key.append(entry.id);
+  return cls_cxx_map_remove_key(hctx, key);
+}
+
+static int bi_log_list_trim_entries(cls_method_context_t hctx,
+                                    const string& start_marker, const string& end_marker,
+			            list<rgw_bi_log_entry>& entries, bool *truncated)
+{
+  string key_iter;
+#define MAX_TRIM_ENTRIES 1000 /* max entries to trim in a single operation */
+  int ret = bi_log_iterate_entries(hctx, start_marker, end_marker,
+                              key_iter, MAX_TRIM_ENTRIES, truncated,
+                              bi_log_list_trim_cb, &entries);
+  return ret;
+}
+
+static int rgw_bi_log_trim(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
+{
+  bufferlist::iterator in_iter = in->begin();
+
+  cls_rgw_bi_log_trim_op op;
+  try {
+    ::decode(op, in_iter);
+  } catch (buffer::error& err) {
+    CLS_LOG(1, "ERROR: rgw_bi_log_list(): failed to decode entry\n");
+    return -EINVAL;
+  }
+
+  cls_rgw_bi_log_list_ret op_ret;
+  list<rgw_bi_log_entry> entries;
+#define MAX_TRIM_ENTRIES 1000 /* don't do more than that in a single operation */
+  bool truncated;
+  int ret = bi_log_list_trim_entries(hctx, op.start_marker, op.end_marker, entries, &truncated);
+  if (ret < 0)
+    return ret;
+
+  if (entries.size() == 0)
+    return -ENODATA;
+
+  list<rgw_bi_log_entry>::iterator iter;
+  for (iter = entries.begin(); iter != entries.end(); ++iter) {
+    rgw_bi_log_entry& entry = *iter;
+
+    ret = bi_log_remove_entry(hctx, entry);
+    if (ret < 0)
+      return ret;
+  }
+
+  return 0;
+}
+
 static void usage_record_prefix_by_time(uint64_t epoch, string& key)
 {
   char buf[32];
@@ -1445,6 +1516,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, "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);
   cls_register_cxx_method(h_class, "dir_suggest_changes", CLS_METHOD_RD | CLS_METHOD_WR, rgw_dir_suggest_changes, &h_rgw_dir_suggest_changes);
 
   /* usage logging */
diff --git a/src/cls/rgw/cls_rgw_client.cc b/src/cls/rgw/cls_rgw_client.cc
index 1cd2cc96c7a..749e83dda96 100644
--- a/src/cls/rgw/cls_rgw_client.cc
+++ b/src/cls/rgw/cls_rgw_client.cc
@@ -183,6 +183,28 @@ int cls_rgw_bi_log_list(IoCtx& io_ctx, string& oid, string& marker, uint32_t max
  return r;
 }
 
+int cls_rgw_bi_log_trim(IoCtx& io_ctx, string& oid, string& start_marker, string& end_marker)
+{
+  int r;
+  do {
+    bufferlist in, out;
+    cls_rgw_bi_log_trim_op call;
+    call.start_marker = start_marker;
+    call.end_marker = end_marker;
+    ::encode(call, in);
+    r = io_ctx.exec(oid, "rgw", "bi_log_trim", in, out);
+
+    if (r == -ENODATA)
+      return 0;
+
+    if (r < 0)
+      return r;
+
+  } while (r != -ENODATA);
+
+ return 0;
+}
+
 int cls_rgw_usage_log_read(IoCtx& io_ctx, string& oid, string& user,
                            uint64_t start_epoch, uint64_t end_epoch, uint32_t max_entries,
                            string& read_iter, map<rgw_user_bucket, rgw_usage_log_entry>& usage,
diff --git a/src/cls/rgw/cls_rgw_client.h b/src/cls/rgw/cls_rgw_client.h
index 413ffe23a97..9b7bc9ed211 100644
--- a/src/cls/rgw/cls_rgw_client.h
+++ b/src/cls/rgw/cls_rgw_client.h
@@ -36,6 +36,7 @@ void cls_rgw_suggest_changes(librados::ObjectWriteOperation& o, bufferlist& upda
 
 int cls_rgw_bi_log_list(librados::IoCtx& io_ctx, string& oid, string& marker, uint32_t max,
                     list<rgw_bi_log_entry>& entries, bool *truncated);
+int cls_rgw_bi_log_trim(librados::IoCtx& io_ctx, string& oid, string& start_marker, string& end_marker);
 
 /* usage logging */
 int cls_rgw_usage_log_read(librados::IoCtx& io_ctx, string& oid, string& user,
diff --git a/src/cls/rgw/cls_rgw_ops.h b/src/cls/rgw/cls_rgw_ops.h
index a13e2dbb885..48b2d4c4257 100644
--- a/src/cls/rgw/cls_rgw_ops.h
+++ b/src/cls/rgw/cls_rgw_ops.h
@@ -406,6 +406,28 @@ struct cls_rgw_bi_log_list_op {
 };
 WRITE_CLASS_ENCODER(cls_rgw_bi_log_list_op)
 
+struct cls_rgw_bi_log_trim_op {
+  string start_marker;
+  string end_marker;
+
+  cls_rgw_bi_log_trim_op() {}
+
+  void encode(bufferlist& bl) const {
+    ENCODE_START(1, 1, bl);
+    ::encode(start_marker, bl);
+    ::encode(end_marker, bl);
+    ENCODE_FINISH(bl);
+  }
+
+  void decode(bufferlist::iterator& bl) {
+    DECODE_START(1, bl);
+    ::decode(start_marker, bl);
+    ::decode(end_marker, bl);
+    DECODE_FINISH(bl);
+  }
+};
+WRITE_CLASS_ENCODER(cls_rgw_bi_log_trim_op)
+
 struct cls_rgw_bi_log_list_ret {
   list<rgw_bi_log_entry> entries;
   bool truncated;
diff --git a/src/rgw/rgw_admin.cc b/src/rgw/rgw_admin.cc
index 7766e5196e5..da6098c9c42 100644
--- a/src/rgw/rgw_admin.cc
+++ b/src/rgw/rgw_admin.cc
@@ -91,6 +91,7 @@ void _usage()
   cerr << "  mdlog list                 list metadata log\n";
   cerr << "  mdlog trim                 trim metadata log\n";
   cerr << "  bilog list                 list bucket index log\n";
+  cerr << "  bilog trim                 trim bucket index log (use start-marker, end-marker)\n";
   cerr << "  datalog list               list data log\n";
   cerr << "options:\n";
   cerr << "   --uid=<id>                user id\n";
@@ -201,6 +202,7 @@ enum {
   OPT_MDLOG_LIST,
   OPT_MDLOG_TRIM,
   OPT_BILOG_LIST,
+  OPT_BILOG_TRIM,
   OPT_DATALOG_LIST,
 };
 
@@ -373,6 +375,8 @@ static int get_cmd(const char *cmd, const char *prev_cmd, bool *need_more)
   } else if (strcmp(prev_cmd, "bilog") == 0) {
     if (strcmp(cmd, "list") == 0)
       return OPT_BILOG_LIST;
+    if (strcmp(cmd, "trim") == 0)
+      return OPT_BILOG_TRIM;
   } else if (strcmp(prev_cmd, "datalog") == 0) {
     if (strcmp(cmd, "list") == 0)
       return OPT_DATALOG_LIST;
@@ -631,6 +635,8 @@ int main(int argc, char **argv)
   string metadata_key;
   RGWObjVersionTracker objv_tracker;
   string marker;
+  string start_marker;
+  string end_marker;
   int max_entries = -1;
 
   std::string val;
@@ -739,6 +745,10 @@ int main(int argc, char **argv)
       metadata_key = val;
     } else if (ceph_argparse_witharg(args, i, &val, "--marker", (char*)NULL)) {
       marker = val;
+    } else if (ceph_argparse_witharg(args, i, &val, "--start-marker", (char*)NULL)) {
+      start_marker = val;
+    } else if (ceph_argparse_witharg(args, i, &val, "--end-marker", (char*)NULL)) {
+      end_marker = val;
     } else {
       ++i;
     }
@@ -1828,6 +1838,23 @@ next:
     formatter->flush(cout);
   }
 
+  if (opt_cmd == OPT_BILOG_TRIM) {
+    if (bucket_name.empty()) {
+      cerr << "ERROR: bucket not specified" << std::endl;
+      return -EINVAL;
+    }
+    int ret = init_bucket(bucket_name, bucket);
+    if (ret < 0) {
+      cerr << "ERROR: could not init bucket: " << cpp_strerror(-ret) << std::endl;
+      return -ret;
+    }
+    ret = store->trim_bi_log_entries(bucket, start_marker, end_marker);
+    if (ret < 0) {
+      cerr << "ERROR: trim_bi_log_entries(): " << cpp_strerror(-ret) << std::endl;
+      return -ret;
+    }
+  }
+
   if (opt_cmd == OPT_DATALOG_LIST) {
     formatter->open_array_section("entries");
     bool truncated;
diff --git a/src/rgw/rgw_rados.cc b/src/rgw/rgw_rados.cc
index 6d6b6324331..1d5b6e5dd4a 100644
--- a/src/rgw/rgw_rados.cc
+++ b/src/rgw/rgw_rados.cc
@@ -4059,6 +4059,21 @@ int RGWRados::list_bi_log_entries(rgw_bucket& bucket, string& marker, uint32_t m
   return 0;
 }
 
+int RGWRados::trim_bi_log_entries(rgw_bucket& bucket, string& start_marker, string& end_marker)
+{
+  librados::IoCtx index_ctx;
+  string oid;
+  int r = open_bucket_index(bucket, index_ctx, oid);
+  if (r < 0)
+    return r;
+
+  int ret = cls_rgw_bi_log_trim(index_ctx, oid, start_marker, end_marker);
+  if (ret < 0)
+    return ret;
+
+  return 0;
+}
+
 int RGWRados::gc_operate(string& oid, librados::ObjectWriteOperation *op)
 {
   return gc_pool_ctx.operate(oid, op);
diff --git a/src/rgw/rgw_rados.h b/src/rgw/rgw_rados.h
index 231026e1f0d..175870dfa0c 100644
--- a/src/rgw/rgw_rados.h
+++ b/src/rgw/rgw_rados.h
@@ -979,6 +979,7 @@ public:
     return cls_obj_complete_cancel(bucket, tag, oid);
   }
   int list_bi_log_entries(rgw_bucket& bucket, string& marker, uint32_t max, std::list<rgw_bi_log_entry>& result, bool *truncated);
+  int trim_bi_log_entries(rgw_bucket& bucket, string& marker, string& end_marker);
 
   int cls_obj_usage_log_add(const string& oid, rgw_usage_log_info& info);
   int cls_obj_usage_log_read(string& oid, string& user, uint64_t start_epoch, uint64_t end_epoch, uint32_t max_entries,