rgw: datalog trim

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

4 files changed, 46 insertions, 3 deletions

diff --git a/src/cls/rgw/cls_rgw_client.cc b/src/cls/rgw/cls_rgw_client.cc
index 749e83dda96..c2c6cca85c2 100644
--- a/src/cls/rgw/cls_rgw_client.cc
+++ b/src/cls/rgw/cls_rgw_client.cc
@@ -195,14 +195,14 @@ int cls_rgw_bi_log_trim(IoCtx& io_ctx, string& oid, string& start_marker, string
     r = io_ctx.exec(oid, "rgw", "bi_log_trim", in, out);
 
     if (r == -ENODATA)
-      return 0;
+      break;
 
     if (r < 0)
       return r;
 
-  } while (r != -ENODATA);
+  } while (1);
 
- return 0;
+  return 0;
 }
 
 int cls_rgw_usage_log_read(IoCtx& io_ctx, string& oid, string& user,
diff --git a/src/rgw/rgw_admin.cc b/src/rgw/rgw_admin.cc
index da6098c9c42..9ea6c2b790c 100644
--- a/src/rgw/rgw_admin.cc
+++ b/src/rgw/rgw_admin.cc
@@ -93,6 +93,7 @@ void _usage()
   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 << "  datalog trim               trim data log\n";
   cerr << "options:\n";
   cerr << "   --uid=<id>                user id\n";
   cerr << "   --subuser=<name>          subuser name\n";
@@ -204,6 +205,7 @@ enum {
   OPT_BILOG_LIST,
   OPT_BILOG_TRIM,
   OPT_DATALOG_LIST,
+  OPT_DATALOG_TRIM,
 };
 
 static int get_cmd(const char *cmd, const char *prev_cmd, bool *need_more)
@@ -380,6 +382,8 @@ static int get_cmd(const char *cmd, const char *prev_cmd, bool *need_more)
   } else if (strcmp(prev_cmd, "datalog") == 0) {
     if (strcmp(cmd, "list") == 0)
       return OPT_DATALOG_LIST;
+    if (strcmp(cmd, "trim") == 0)
+      return OPT_DATALOG_TRIM;
   }
 
   return -EINVAL;
@@ -1896,5 +1900,23 @@ next:
     formatter->flush(cout);
   }
   
+  if (opt_cmd == OPT_DATALOG_TRIM) {
+    utime_t start_time, end_time;
+
+    int ret = parse_date_str(start_date, start_time);
+    if (ret < 0)
+      return -ret;
+
+    ret = parse_date_str(end_date, end_time);
+    if (ret < 0)
+      return -ret;
+
+    RGWDataChangesLog *log = store->data_log;
+    ret = log->trim_entries(start_time, end_time);
+    if (ret < 0) {
+      cerr << "ERROR: trim_entries(): " << cpp_strerror(-ret) << std::endl;
+      return -ret;
+    }
+  }
   return 0;
 }
diff --git a/src/rgw/rgw_bucket.cc b/src/rgw/rgw_bucket.cc
index 308678d42d8..809c653de14 100644
--- a/src/rgw/rgw_bucket.cc
+++ b/src/rgw/rgw_bucket.cc
@@ -946,6 +946,7 @@ void rgw_data_change::dump(Formatter *f) const
   }
   encode_json("entity_type", type, f);
   encode_json("key", key, f);
+  encode_json("timestamp", timestamp, f);
 }
 
 
@@ -980,6 +981,7 @@ int RGWDataChangesLog::renew_entries()
     bufferlist bl;
     change.entity_type = ENTITY_TYPE_BUCKET;
     change.key = bucket.name;
+    change.timestamp = ut;
     ::encode(change, bl);
 
     store->time_log_prepare_entry(entry, ut, section, bucket.name, bl);
@@ -1100,6 +1102,7 @@ int RGWDataChangesLog::add_entry(rgw_bucket& bucket) {
     rgw_data_change change;
     change.entity_type = ENTITY_TYPE_BUCKET;
     change.key = bucket.name;
+    change.timestamp = now;
     ::encode(change, bl);
     string section;
 
@@ -1180,6 +1183,20 @@ int RGWDataChangesLog::list_entries(utime_t& start_time, utime_t& end_time, int
   return 0;
 }
 
+int RGWDataChangesLog::trim_entries(utime_t& start_time, utime_t& end_time)
+{
+  for (int shard = 0; shard < num_shards; shard++) {
+    int ret = store->time_log_trim(oids[shard], start_time, end_time);
+    if (ret == -ENOENT) {
+      continue;
+    }
+    if (ret < 0)
+      return ret;
+  }
+
+  return 0;
+}
+
 bool RGWDataChangesLog::going_down()
 {
   return (down_flag.read() != 0);
diff --git a/src/rgw/rgw_bucket.h b/src/rgw/rgw_bucket.h
index 8a849b32201..a6ac836ace7 100644
--- a/src/rgw/rgw_bucket.h
+++ b/src/rgw/rgw_bucket.h
@@ -229,12 +229,14 @@ enum DataLogEntityType {
 struct rgw_data_change {
   DataLogEntityType entity_type;
   string key;
+  utime_t timestamp;
 
   void encode(bufferlist& bl) const {
     ENCODE_START(1, 1, bl);
     uint8_t t = (uint8_t)entity_type;
     ::encode(t, bl);
     ::encode(key, bl);
+    ::encode(timestamp, bl);
     ENCODE_FINISH(bl);
   }
 
@@ -244,6 +246,7 @@ struct rgw_data_change {
      ::decode(t, bl);
      entity_type = (DataLogEntityType)t;
      ::decode(key, bl);
+     ::decode(timestamp, bl);
      DECODE_FINISH(bl);
   }
 
@@ -333,6 +336,7 @@ public:
   int renew_entries();
   int list_entries(int shard, utime_t& start_time, utime_t& end_time, int max_entries,
                list<rgw_data_change>& entries, string& marker, bool *truncated);
+  int trim_entries(utime_t& start_time, utime_t& end_time);
 
   struct LogMarker {
     int shard;