radosgw-admin, cls_rgw: list bucket index log

a new radosgw-admin command to list bucket index log.

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

9 files changed, 347 insertions, 19 deletions

diff --git a/src/cls/rgw/cls_rgw.cc b/src/cls/rgw/cls_rgw.cc
index 64b7e6912c3..0c0f4f7398e 100644
--- a/src/cls/rgw/cls_rgw.cc
+++ b/src/cls/rgw/cls_rgw.cc
@@ -27,6 +27,7 @@ cls_method_handle_t h_rgw_bucket_check_index;
 cls_method_handle_t h_rgw_bucket_rebuild_index;
 cls_method_handle_t h_rgw_bucket_prepare_op;
 cls_method_handle_t h_rgw_bucket_complete_op;
+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;
 cls_method_handle_t h_rgw_user_usage_log_read;
@@ -44,8 +45,13 @@ cls_method_handle_t h_rgw_gc_remove;
 #define BI_BUCKET_OBJS_INDEX 0
 #define BI_BUCKET_LOG_INDEX  1
 
+#define BI_BUCKET_LAST_INDEX  2
+
 static string bucket_index_prefixes[] = { "", /* special handling for the objs index */
-                                          "0_" };
+                                          "0_",
+
+                                          /* this must be the last index */
+                                          "9999_",};
 
 static uint64_t get_rounded_size(uint64_t size)
 {
@@ -91,14 +97,13 @@ static void get_index_ver_key(cls_method_context_t hctx, uint64_t index_ver, str
   *key = buf;
 }
 
-static void bi_log_index_key(cls_method_context_t hctx, string& key, uint64_t index_ver)
+static void bi_log_index_key(cls_method_context_t hctx, string& key, string& id, uint64_t index_ver)
 {
   key = BI_PREFIX_CHAR;
   key.append(bucket_index_prefixes[BI_BUCKET_LOG_INDEX]);
 
-  string k;
-  get_index_ver_key(hctx, index_ver, &k);
-  key.append(k);
+  get_index_ver_key(hctx, index_ver, &id);
+  key.append(id);
 }
 
 static int log_index_operation(cls_method_context_t hctx, string& obj, RGWModifyOp op, rgw_bucket_entry_ver& ver, RGWPendingState state, uint64_t index_ver)
@@ -113,11 +118,11 @@ static int log_index_operation(cls_method_context_t hctx, string& obj, RGWModify
   entry.ver = ver;
   entry.state = state;
   entry.index_ver = index_ver;
-  ::encode(entry, bl);
 
   string key;
+  bi_log_index_key(hctx, key, entry.id, index_ver);
 
-  bi_log_index_key(hctx, key, index_ver);
+  ::encode(entry, bl);
 
   return cls_cxx_map_set_val(hctx, key, &bl);
 }
@@ -715,6 +720,133 @@ int rgw_dir_suggest_changes(cls_method_context_t hctx, bufferlist *in, bufferlis
   return 0;
 }
 
+int bi_log_record_decode(bufferlist& bl, rgw_bi_log_entry& e)
+{
+  bufferlist::iterator iter = bl.begin();
+  try {
+    ::decode(e, iter);
+  } catch (buffer::error& err) {
+    CLS_LOG(0, "ERROR: failed to decode rgw_bi_log_entry");
+    return -EIO;
+  }
+  return 0;
+}
+
+static int bi_log_iterate_entries(cls_method_context_t hctx, const string& 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)
+{
+  CLS_LOG(10, "bi_log_iterate_range");
+
+  map<string, bufferlist> keys;
+  string filter_prefix, end_key;
+  uint32_t i = 0;
+  string key;
+
+  if (truncated)
+    *truncated = false;
+
+  string start_key;
+  if (key_iter.empty()) {
+    key = BI_PREFIX_CHAR;
+    key.append(bucket_index_prefixes[BI_BUCKET_LOG_INDEX]);
+    key.append(marker);
+
+    start_key = key;
+  } else {
+    start_key = key_iter;
+  }
+
+  end_key = BI_PREFIX_CHAR;
+  end_key.append(bucket_index_prefixes[BI_BUCKET_LAST_INDEX]);
+
+  CLS_LOG(0, "bi_log_iterate_entries end_key=%s\n", end_key.c_str());
+
+  string filter;
+
+  do {
+#define BI_NUM_KEYS 128
+    int ret = cls_cxx_map_get_vals(hctx, start_key, filter, BI_NUM_KEYS, &keys);
+    if (ret < 0)
+      return ret;
+
+
+    map<string, bufferlist>::iterator iter = keys.begin();
+    if (iter == keys.end())
+      break;
+
+    for (; iter != keys.end(); ++iter) {
+      const string& key = iter->first;
+      rgw_bi_log_entry e;
+
+      CLS_LOG(0, "bi_log_iterate_entries key=%s bl.length=%d\n", key.c_str(), (int)iter->second.length());
+
+      if (key.compare(end_key) >= 0)
+        return 0;
+
+      ret = bi_log_record_decode(iter->second, e);
+      if (ret < 0)
+        return ret;
+
+      if (max_entries && (i >= max_entries)) {
+        if (truncated)
+          *truncated = true;
+        key_iter = key;
+        return 0;
+      }
+
+      ret = cb(hctx, key, e, param);
+      if (ret < 0)
+        return ret;
+      i++;
+
+    }
+    --iter;
+    start_key = iter->first;
+  } while (true);
+  return 0;
+}
+
+static int bi_log_list_cb(cls_method_context_t hctx, const string& key, rgw_bi_log_entry& info, void *param)
+{
+  list<rgw_bi_log_entry> *l = (list<rgw_bi_log_entry> *)param;
+  l->push_back(info);
+  return 0;
+}
+
+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,
+                              key_iter, max, truncated,
+                              bi_log_list_cb, &entries);
+  return ret;
+}
+
+static int rgw_bi_log_list(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
+{
+  bufferlist::iterator in_iter = in->begin();
+
+  cls_rgw_bi_log_list_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;
+  int ret = bi_log_list_entries(hctx, op.marker, op.max, op_ret.entries, &op_ret.truncated);
+  if (ret < 0)
+    return ret;
+
+  ::encode(op_ret, *out);
+
+  return 0;
+}
+
 static void usage_record_prefix_by_time(uint64_t epoch, string& key)
 {
   char buf[32];
@@ -1307,6 +1439,7 @@ void __cls_init()
   cls_register_cxx_method(h_class, "bucket_rebuild_index", CLS_METHOD_RD | CLS_METHOD_WR, rgw_bucket_rebuild_index, &h_rgw_bucket_rebuild_index);
   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, "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 3113f9ed136..1cd2cc96c7a 100644
--- a/src/cls/rgw/cls_rgw_client.cc
+++ b/src/cls/rgw/cls_rgw_client.cc
@@ -4,6 +4,8 @@
 #include "cls/rgw/cls_rgw_ops.h"
 #include "include/rados/librados.hpp"
 
+#include "common/debug.h"
+
 using namespace librados;
 
 void cls_rgw_bucket_init(ObjectWriteOperation& o)
@@ -153,6 +155,34 @@ int cls_rgw_get_dir_header(IoCtx& io_ctx, string& oid, rgw_bucket_dir_header *he
  return r;
 }
 
+int cls_rgw_bi_log_list(IoCtx& io_ctx, string& oid, string& marker, uint32_t max,
+                    list<rgw_bi_log_entry>& entries, bool *truncated)
+{
+  bufferlist in, out;
+  cls_rgw_bi_log_list_op call;
+  call.marker = marker;
+  call.max = max;
+  ::encode(call, in);
+  int r = io_ctx.exec(oid, "rgw", "bi_log_list", in, out);
+  if (r < 0)
+    return r;
+
+  cls_rgw_bi_log_list_ret ret;
+  try {
+    bufferlist::iterator iter = out.begin();
+    ::decode(ret, iter);
+  } catch (buffer::error& err) {
+    return -EIO;
+  }
+
+  entries = ret.entries;
+
+  if (truncated)
+    *truncated = ret.truncated;
+
+ return r;
+}
+
 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 d955fab6a3a..413ffe23a97 100644
--- a/src/cls/rgw/cls_rgw_client.h
+++ b/src/cls/rgw/cls_rgw_client.h
@@ -32,6 +32,11 @@ void cls_rgw_encode_suggestion(char op, rgw_bucket_dir_entry& dirent, bufferlist
 
 void cls_rgw_suggest_changes(librados::ObjectWriteOperation& o, bufferlist& updates);
 
+/* bucket index log */
+
+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);
+
 /* usage logging */
 int cls_rgw_usage_log_read(librados::IoCtx& io_ctx, string& oid, string& user,
                            uint64_t start_epoch, uint64_t end_epoch, uint32_t max_entries,
diff --git a/src/cls/rgw/cls_rgw_ops.h b/src/cls/rgw/cls_rgw_ops.h
index 572c609f1d0..a13e2dbb885 100644
--- a/src/cls/rgw/cls_rgw_ops.h
+++ b/src/cls/rgw/cls_rgw_ops.h
@@ -384,5 +384,49 @@ struct cls_rgw_gc_remove_op {
 };
 WRITE_CLASS_ENCODER(cls_rgw_gc_remove_op)
 
+struct cls_rgw_bi_log_list_op {
+  string marker;
+  uint32_t max;
+
+  cls_rgw_bi_log_list_op() : max(0) {}
+
+  void encode(bufferlist& bl) const {
+    ENCODE_START(1, 1, bl);
+    ::encode(marker, bl);
+    ::encode(max, bl);
+    ENCODE_FINISH(bl);
+  }
+
+  void decode(bufferlist::iterator& bl) {
+    DECODE_START(1, bl);
+    ::decode(marker, bl);
+    ::decode(max, bl);
+    DECODE_FINISH(bl);
+  }
+};
+WRITE_CLASS_ENCODER(cls_rgw_bi_log_list_op)
+
+struct cls_rgw_bi_log_list_ret {
+  list<rgw_bi_log_entry> entries;
+  bool truncated;
+
+  cls_rgw_bi_log_list_ret() : truncated(false) {}
+
+  void encode(bufferlist& bl) const {
+    ENCODE_START(1, 1, bl);
+    ::encode(entries, bl);
+    ::encode(truncated, bl);
+    ENCODE_FINISH(bl);
+  }
+
+  void decode(bufferlist::iterator& bl) {
+    DECODE_START(1, bl);
+    ::decode(entries, bl);
+    ::decode(truncated, bl);
+    DECODE_FINISH(bl);
+  }
+};
+WRITE_CLASS_ENCODER(cls_rgw_bi_log_list_ret)
+
 
 #endif
diff --git a/src/cls/rgw/cls_rgw_types.cc b/src/cls/rgw/cls_rgw_types.cc
index 828a0f7f78f..286f6771805 100644
--- a/src/cls/rgw/cls_rgw_types.cc
+++ b/src/cls/rgw/cls_rgw_types.cc
@@ -96,6 +96,48 @@ void rgw_bucket_dir_entry::dump(Formatter *f) const
   f->close_section();
 }
 
+void rgw_bi_log_entry::dump(Formatter *f) const
+{
+  f->dump_string("id", id);
+  f->dump_string("object", object);
+
+  f->dump_int("index_ver", index_ver);
+  f->dump_stream("timestamp") << timestamp;
+  f->open_object_section("ver");
+  ver.dump(f);
+  f->close_section();
+
+  switch (op) {
+    case CLS_RGW_OP_ADD:
+      f->dump_string("op", "write");
+      break;
+    case CLS_RGW_OP_DEL:
+      f->dump_string("op", "del");
+      break;
+    case CLS_RGW_OP_CANCEL:
+      f->dump_string("op", "cancel");
+      break;
+    case CLS_RGW_OP_UNKNOWN:
+      f->dump_string("op", "unknown");
+      break;
+    default:
+      f->dump_string("op", "invalid");
+      break;
+  }
+
+  switch (state) {
+    case CLS_RGW_STATE_PENDING_MODIFY:
+      f->dump_string("state", "pending");
+      break;
+    case CLS_RGW_STATE_COMPLETE:
+      f->dump_string("state", "complete");
+      break;
+    default:
+      f->dump_string("state", "invalid");
+      break;
+  }
+}
+
 void rgw_bucket_category_stats::generate_test_instances(list<rgw_bucket_category_stats*>& o)
 {
   rgw_bucket_category_stats *s = new rgw_bucket_category_stats;
diff --git a/src/cls/rgw/cls_rgw_types.h b/src/cls/rgw/cls_rgw_types.h
index 2b15f9c7d2b..1af0a8e29b3 100644
--- a/src/cls/rgw/cls_rgw_types.h
+++ b/src/cls/rgw/cls_rgw_types.h
@@ -99,13 +99,33 @@ WRITE_CLASS_ENCODER(rgw_bucket_dir_entry_meta)
 template<class T>
 void encode_packed_val(T val, bufferlist& bl)
 {
-  unsigned char c = 0x80 | (unsigned char)sizeof(T);
-  ::encode(c, bl);
-  ::encode(val, bl);
+  if ((uint64_t)val < 0x80) {
+    ::encode((uint8_t)val, bl);
+  } else {
+    unsigned char c = 0x80;
+
+    if ((uint64_t)val < 0x100) {
+      c |= 1;
+      ::encode(c, bl);
+      ::encode((uint8_t)val, bl);
+    } else if ((uint64_t)val <= 0x10000) {
+      c |= 2;
+      ::encode(c, bl);
+      ::encode((uint16_t)val, bl);
+    } else if ((uint64_t)val <= 0x1000000) {
+      c |= 4;
+      ::encode(c, bl);
+      ::encode((uint32_t)val, bl);
+    } else {
+      c |= 8;
+      ::encode(c, bl);
+      ::encode((uint64_t)val, bl);
+    }
+  }
 }
 
 template<class T>
-void decode_packed_val(T val, bufferlist::iterator& bl)
+void decode_packed_val(T& val, bufferlist::iterator& bl)
 {
   unsigned char c;
   ::decode(c, bl);
@@ -114,7 +134,7 @@ void decode_packed_val(T val, bufferlist::iterator& bl)
     return;
   }
 
-  c ^= 0x80;
+  c &= ~0x80;
 
   switch (c) {
     case 1:
@@ -239,7 +259,7 @@ struct rgw_bi_log_entry {
     uint8_t c = (uint8_t)op;
     ::encode(c, bl);
     c = (uint8_t)state;
-    ::encode(state, bl);
+    ::encode(c, bl);
     encode_packed_val(index_ver, bl);
     ENCODE_FINISH(bl);
   }
diff --git a/src/rgw/rgw_admin.cc b/src/rgw/rgw_admin.cc
index d9282ed7f14..6d86890dec3 100644
--- a/src/rgw/rgw_admin.cc
+++ b/src/rgw/rgw_admin.cc
@@ -88,7 +88,8 @@ void _usage()
   cerr << "  metadata put               put metadata info\n";
   cerr << "  metadata rm                remove metadata info\n";
   cerr << "  metadata list              list metadata info\n";
-  cerr << "  mdlog show                 show metadata log\n";
+  cerr << "  mdlog list                 list metadata log\n";
+  cerr << "  bilog list                 list bucket index log\n";
   cerr << "options:\n";
   cerr << "   --uid=<id>                user id\n";
   cerr << "   --subuser=<name>          subuser name\n";
@@ -194,7 +195,8 @@ enum {
   OPT_METADATA_PUT,
   OPT_METADATA_RM,
   OPT_METADATA_LIST,
-  OPT_MDLOG_SHOW,
+  OPT_MDLOG_LIST,
+  OPT_BILOG_LIST,
 };
 
 static int get_cmd(const char *cmd, const char *prev_cmd, bool *need_more)
@@ -221,7 +223,8 @@ static int get_cmd(const char *cmd, const char *prev_cmd, bool *need_more)
       strcmp(cmd, "zone") == 0 ||
       strcmp(cmd, "temp") == 0 ||
       strcmp(cmd, "metadata") == 0 ||
-      strcmp(cmd, "mdlog") == 0) {
+      strcmp(cmd, "mdlog") == 0 ||
+      strcmp(cmd, "bilog") == 0) {
     *need_more = true;
     return 0;
   }
@@ -355,8 +358,11 @@ static int get_cmd(const char *cmd, const char *prev_cmd, bool *need_more)
     if (strcmp(cmd, "list") == 0)
       return OPT_METADATA_LIST;
   } else if (strcmp(prev_cmd, "mdlog") == 0) {
-    if (strcmp(cmd, "show") == 0)
-      return OPT_MDLOG_SHOW;
+    if (strcmp(cmd, "list") == 0)
+      return OPT_MDLOG_LIST;
+  } else if (strcmp(prev_cmd, "bilog") == 0) {
+    if (strcmp(cmd, "list") == 0)
+      return OPT_BILOG_LIST;
   }
 
   return -EINVAL;
@@ -1572,7 +1578,7 @@ next:
     store->meta_mgr->list_keys_complete(handle);
   }
 
-  if (opt_cmd == OPT_MDLOG_SHOW) {
+  if (opt_cmd == OPT_MDLOG_LIST) {
     void *handle;
     list<cls_log_entry> entries;
 
@@ -1611,5 +1617,28 @@ next:
     formatter->flush(cout);
   }
 
+  if (opt_cmd == OPT_BILOG_LIST) {
+    string marker;
+    formatter->open_array_section("entries");
+    bool truncated;
+    do {
+      list<rgw_bi_log_entry> entries;
+      int ret = store->list_bi_log_entries(bucket, marker, 1000, entries, &truncated);
+      if (ret < 0) {
+        cerr << "ERROR: list_bi_log_entries(): " << cpp_strerror(-ret) << std::endl;
+        return -ret;
+      }
+
+      for (list<rgw_bi_log_entry>::iterator iter = entries.begin(); iter != entries.end(); ++iter) {
+        rgw_bi_log_entry& entry = *iter;
+        encode_json("entry", entry, formatter);
+      }
+      formatter->flush(cout);
+    } while (truncated);
+
+    formatter->close_section();
+    formatter->flush(cout);
+  }
+  
   return 0;
 }
diff --git a/src/rgw/rgw_rados.cc b/src/rgw/rgw_rados.cc
index 2a431b2610e..60ad5e481ea 100644
--- a/src/rgw/rgw_rados.cc
+++ b/src/rgw/rgw_rados.cc
@@ -3985,6 +3985,30 @@ int RGWRados::list_raw_objects(rgw_bucket& pool, const string& prefix_filter,
   return oids.size();
 }
 
+int RGWRados::list_bi_log_entries(rgw_bucket& bucket, string& marker, uint32_t max,
+                                  std::list<rgw_bi_log_entry>& result, bool *truncated)
+{
+  result.clear();
+
+  librados::IoCtx io_ctx;
+  string oid;
+  int r = open_bucket(bucket, io_ctx, oid);
+  if (r < 0)
+    return r;
+
+  std::list<rgw_bi_log_entry> entries;
+  int ret = cls_rgw_bi_log_list(io_ctx, oid, marker, max - result.size(), entries, truncated);
+  if (ret < 0)
+    return ret;
+
+  std::list<rgw_bi_log_entry>::iterator iter;
+  for (iter = entries.begin(); iter != entries.end(); ++iter) {
+    result.push_back(*iter);
+  }
+
+  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 76e98adfdcb..b81a38acea3 100644
--- a/src/rgw/rgw_rados.h
+++ b/src/rgw/rgw_rados.h
@@ -971,6 +971,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 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,