cls_rgw: bucket index versioning

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

7 files changed, 104 insertions, 40 deletions

diff --git a/src/cls/rgw/cls_rgw.cc b/src/cls/rgw/cls_rgw.cc
index 0b3646224ba..64b7e6912c3 100644
--- a/src/cls/rgw/cls_rgw.cc
+++ b/src/cls/rgw/cls_rgw.cc
@@ -56,7 +56,7 @@ static bool bi_is_objs_index(const string& s) {
   return ((unsigned char)s[0] != BI_PREFIX_CHAR);
 }
 
-static int bi_entry_type(const string& s)
+int bi_entry_type(const string& s)
 {
   if (bi_is_objs_index(s)) {
     return BI_BUCKET_OBJS_INDEX;
@@ -82,17 +82,26 @@ static void get_time_key(utime_t& ut, string *key)
   *key = buf;
 }
 
-static void bi_log_index_key(string& key, utime_t& t, string& obj)
+static void get_index_ver_key(cls_method_context_t hctx, uint64_t index_ver, string *key)
+{
+  char buf[48];
+  snprintf(buf, sizeof(buf), "%011llu.%llu.%d", (unsigned long long)index_ver,
+           (unsigned long long)cls_current_version(hctx),
+           cls_current_subop_num(hctx));
+  *key = buf;
+}
+
+static void bi_log_index_key(cls_method_context_t hctx, string& key, uint64_t index_ver)
 {
   key = BI_PREFIX_CHAR;
   key.append(bucket_index_prefixes[BI_BUCKET_LOG_INDEX]);
 
-  string tk;
-  get_time_key(t, &tk);
-  key.append(tk);
+  string k;
+  get_index_ver_key(hctx, index_ver, &k);
+  key.append(k);
 }
 
-static int log_index_operation(cls_method_context_t hctx, string& obj, RGWModifyOp op, rgw_bucket_entry_ver& ver, RGWPendingState state)
+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)
 {
   bufferlist bl;
 
@@ -103,11 +112,12 @@ static int log_index_operation(cls_method_context_t hctx, string& obj, RGWModify
   entry.op = op;
   entry.ver = ver;
   entry.state = state;
+  entry.index_ver = index_ver;
   ::encode(entry, bl);
 
   string key;
 
-  bi_log_index_key(key, entry.timestamp, obj);
+  bi_log_index_key(hctx, key, index_ver);
 
   return cls_cxx_map_set_val(hctx, key, &bl);
 }
@@ -192,6 +202,7 @@ static int check_index(cls_method_context_t hctx, struct rgw_bucket_dir_header *
   }
 
   calc_header->tag_timeout = existing_header->tag_timeout;
+  calc_header->ver = existing_header->ver;
 
   bufferlist bl;
 
@@ -247,6 +258,16 @@ int rgw_bucket_check_index(cls_method_context_t hctx, bufferlist *in, bufferlist
   return 0;
 }
 
+static int write_bucket_header(cls_method_context_t hctx, struct rgw_bucket_dir_header *header)
+{
+  header->ver++;
+
+  bufferlist header_bl;
+  ::encode(*header, header_bl);
+  return cls_cxx_map_write_header(hctx, &header_bl);
+}
+
+
 int rgw_bucket_rebuild_index(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
 {
   struct rgw_bucket_dir_header existing_header;
@@ -255,10 +276,7 @@ int rgw_bucket_rebuild_index(cls_method_context_t hctx, bufferlist *in, bufferli
   if (rc < 0)
     return rc;
 
-  bufferlist header_bl;
-  ::encode(calc_header, header_bl);
-  rc = cls_cxx_map_write_header(hctx, &header_bl);
-  return rc;
+  return write_bucket_header(hctx, &calc_header);
 }
 
 
@@ -285,9 +303,8 @@ int rgw_bucket_init_index(cls_method_context_t hctx, bufferlist *in, bufferlist
   }
 
   rgw_bucket_dir dir;
-  ::encode(dir.header, header_bl);
-  rc = cls_cxx_map_write_header(hctx, &header_bl);
-  return rc;
+
+  return write_bucket_header(hctx, &dir.header);
 }
 
 int rgw_bucket_set_tag_timeout(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
@@ -317,10 +334,7 @@ int rgw_bucket_set_tag_timeout(cls_method_context_t hctx, bufferlist *in, buffer
 
   header.tag_timeout = op.tag_timeout;
 
-  header_bl.clear();
-  ::encode(header, header_bl);
-  rc = cls_cxx_map_write_header(hctx, &header_bl);
-  return rc;
+  return write_bucket_header(hctx, &header);
 }
 
 int rgw_bucket_prepare_op(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
@@ -379,7 +393,22 @@ int rgw_bucket_prepare_op(cls_method_context_t hctx, bufferlist *in, bufferlist
   info.state = CLS_RGW_STATE_PENDING_MODIFY;
   info.op = op.op;
 
-  rc = log_index_operation(hctx, op.name, op.op, entry.ver, info.state);
+
+  bufferlist header_bl;
+  struct rgw_bucket_dir_header header;
+  rc = cls_cxx_map_read_header(hctx, &header_bl);
+  if (rc < 0)
+    return rc;
+
+  bufferlist::iterator header_iter = header_bl.begin();
+  try {
+    ::decode(header, header_iter);
+  } catch (buffer::error& err) {
+    CLS_LOG(1, "ERROR: rgw_bucket_complete_op(): failed to decode header\n");
+    return -EINVAL;
+  }
+
+  rc = log_index_operation(hctx, op.name, op.op, entry.ver, info.state, header.ver);
   if (rc < 0)
     return rc;
 
@@ -387,7 +416,10 @@ int rgw_bucket_prepare_op(cls_method_context_t hctx, bufferlist *in, bufferlist
   bufferlist info_bl;
   ::encode(entry, info_bl);
   rc = cls_cxx_map_set_val(hctx, op.name, &info_bl);
-  return rc;
+  if (rc < 0)
+    return rc;
+
+  return write_bucket_header(hctx, &header);
 }
 
 static void unaccount_entry(struct rgw_bucket_dir_header& header, struct rgw_bucket_dir_entry& entry)
@@ -463,6 +495,8 @@ int rgw_bucket_complete_op(cls_method_context_t hctx, bufferlist *in, bufferlist
     return rc;
   }
 
+  entry.index_ver = header.ver;
+
   if (op.tag.size()) {
     map<string, struct rgw_bucket_pending_info>::iterator pinter = entry.pending_map.find(op.tag);
     if (pinter == entry.pending_map.end()) {
@@ -486,7 +520,7 @@ int rgw_bucket_complete_op(cls_method_context_t hctx, bufferlist *in, bufferlist
 
   bufferlist op_bl;
   if (cancel) {
-    rc = log_index_operation(hctx, op.name, op.op, entry.ver, CLS_RGW_STATE_COMPLETE);
+    rc = log_index_operation(hctx, op.name, op.op, entry.ver, CLS_RGW_STATE_COMPLETE, header.ver);
     if (rc < 0)
       return rc;
 
@@ -542,7 +576,7 @@ int rgw_bucket_complete_op(cls_method_context_t hctx, bufferlist *in, bufferlist
     break;
   }
 
-  rc = log_index_operation(hctx, op.name, op.op, entry.ver, CLS_RGW_STATE_COMPLETE);
+  rc = log_index_operation(hctx, op.name, op.op, entry.ver, CLS_RGW_STATE_COMPLETE, header.ver);
   if (rc < 0)
     return rc;
 
@@ -560,7 +594,7 @@ int rgw_bucket_complete_op(cls_method_context_t hctx, bufferlist *in, bufferlist
     CLS_LOG(0, "rgw_bucket_complete_op(): entry.name=%s entry.meta.category=%d\n", remove_entry.name.c_str(), remove_entry.meta.category);
     unaccount_entry(header, remove_entry);
 
-    rc = log_index_operation(hctx, op.name, CLS_RGW_OP_DEL, remove_entry.ver, CLS_RGW_STATE_COMPLETE);
+    rc = log_index_operation(hctx, op.name, CLS_RGW_OP_DEL, remove_entry.ver, CLS_RGW_STATE_COMPLETE, header.ver);
     if (rc < 0)
       continue;
 
@@ -571,9 +605,7 @@ int rgw_bucket_complete_op(cls_method_context_t hctx, bufferlist *in, bufferlist
     }
   }
 
-  bufferlist new_header_bl;
-  ::encode(header, new_header_bl);
-  return cls_cxx_map_write_header(hctx, &new_header_bl);
+  return write_bucket_header(hctx, &header);
 }
 
 int rgw_dir_suggest_changes(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
@@ -666,6 +698,7 @@ int rgw_dir_suggest_changes(cls_method_context_t hctx, bufferlist *in, bufferlis
         stats.total_size += cur_change.meta.size;
         stats.total_size_rounded += get_rounded_size(cur_change.meta.size);
         header_changed = true;
+        cur_change.index_ver = header.ver;
         bufferlist cur_state_bl;
         ::encode(cur_change, cur_state_bl);
         ret = cls_cxx_map_set_val(hctx, cur_change.name, &cur_state_bl);
@@ -676,11 +709,8 @@ int rgw_dir_suggest_changes(cls_method_context_t hctx, bufferlist *in, bufferlis
     }
   }
 
-  bufferlist update_bl;
   if (header_changed) {
-    bufferlist new_header_bl;
-    ::encode(header, new_header_bl);
-    return cls_cxx_map_write_header(hctx, &new_header_bl);
+    return write_bucket_header(hctx, &header);
   }
   return 0;
 }
diff --git a/src/cls/rgw/cls_rgw_types.cc b/src/cls/rgw/cls_rgw_types.cc
index 5a9dd9f548f..828a0f7f78f 100644
--- a/src/cls/rgw/cls_rgw_types.cc
+++ b/src/cls/rgw/cls_rgw_types.cc
@@ -135,6 +135,8 @@ void rgw_bucket_dir_header::generate_test_instances(list<rgw_bucket_dir_header*>
 
 void rgw_bucket_dir_header::dump(Formatter *f) const
 {
+  f->dump_int("ver", ver);
+  f->dump_int("master_ver", master_ver);
   map<uint8_t, struct rgw_bucket_category_stats>::const_iterator iter = stats.begin();
   f->open_array_section("stats");
   for (; iter != stats.end(); ++iter) {
diff --git a/src/cls/rgw/cls_rgw_types.h b/src/cls/rgw/cls_rgw_types.h
index 7f7c9ccc88c..2b15f9c7d2b 100644
--- a/src/cls/rgw/cls_rgw_types.h
+++ b/src/cls/rgw/cls_rgw_types.h
@@ -178,22 +178,24 @@ struct rgw_bucket_dir_entry {
   bool exists;
   struct rgw_bucket_dir_entry_meta meta;
   map<string, struct rgw_bucket_pending_info> pending_map;
+  uint64_t index_ver;
 
   rgw_bucket_dir_entry() :
     exists(false) {}
 
   void encode(bufferlist &bl) const {
-    ENCODE_START(4, 3, bl);
+    ENCODE_START(5, 3, bl);
     ::encode(name, bl);
     ::encode(ver, bl);
     ::encode(exists, bl);
     ::encode(meta, bl);
     ::encode(pending_map, bl);
     ::encode(locator, bl);
+    ::encode_packed_val(index_ver, bl);
     ENCODE_FINISH(bl);
   }
   void decode(bufferlist::iterator &bl) {
-    DECODE_START_LEGACY_COMPAT_LEN(4, 3, 3, bl);
+    DECODE_START_LEGACY_COMPAT_LEN(5, 3, 3, bl);
     ::decode(name, bl);
     if (struct_v >= 4) {
       ::decode(ver, bl);
@@ -207,6 +209,9 @@ struct rgw_bucket_dir_entry {
     if (struct_v >= 2) {
       ::decode(locator, bl);
     }
+    if (struct_v >= 5) {
+      ::decode_packed_val(index_ver, bl);
+    }
     DECODE_FINISH(bl);
   }
   void dump(Formatter *f) const;
@@ -215,16 +220,19 @@ struct rgw_bucket_dir_entry {
 WRITE_CLASS_ENCODER(rgw_bucket_dir_entry)
 
 struct rgw_bi_log_entry {
+  string id;
   string object;
   utime_t timestamp;
   rgw_bucket_entry_ver ver;
   RGWModifyOp op;
   RGWPendingState state;
+  uint64_t index_ver;
 
-  rgw_bi_log_entry() : op(CLS_RGW_OP_UNKNOWN) {}
+  rgw_bi_log_entry() : op(CLS_RGW_OP_UNKNOWN), index_ver(0) {}
 
   void encode(bufferlist &bl) const {
     ENCODE_START(1, 1, bl);
+    ::encode(id, bl);
     ::encode(object, bl);
     ::encode(timestamp, bl);
     ::encode(ver, bl);
@@ -232,10 +240,12 @@ struct rgw_bi_log_entry {
     ::encode(c, bl);
     c = (uint8_t)state;
     ::encode(state, bl);
+    encode_packed_val(index_ver, bl);
     ENCODE_FINISH(bl);
   }
   void decode(bufferlist::iterator &bl) {
     DECODE_START(1, bl);
+    ::decode(id, bl);
     ::decode(object, bl);
     ::decode(timestamp, bl);
     ::decode(ver, bl);
@@ -244,6 +254,7 @@ struct rgw_bi_log_entry {
     op = (RGWModifyOp)c;
     ::decode(c, bl);
     state = (RGWPendingState)c;
+    decode_packed_val(index_ver, bl);
     DECODE_FINISH(bl);
   }
   void dump(Formatter *f) const;
@@ -280,13 +291,17 @@ WRITE_CLASS_ENCODER(rgw_bucket_category_stats)
 struct rgw_bucket_dir_header {
   map<uint8_t, rgw_bucket_category_stats> stats;
   uint64_t tag_timeout;
+  uint64_t ver;
+  uint64_t master_ver;
 
-  rgw_bucket_dir_header() : tag_timeout(0) {}
+  rgw_bucket_dir_header() : tag_timeout(0), ver(0), master_ver(0) {}
 
   void encode(bufferlist &bl) const {
-    ENCODE_START(3, 2, bl);
+    ENCODE_START(4, 2, bl);
     ::encode(stats, bl);
     ::encode(tag_timeout, bl);
+    ::encode(ver, bl);
+    ::encode(master_ver, bl);
     ENCODE_FINISH(bl);
   }
   void decode(bufferlist::iterator &bl) {
@@ -297,6 +312,12 @@ struct rgw_bucket_dir_header {
     } else {
       tag_timeout = 0;
     }
+    if (struct_v >= 4) {
+      ::decode(ver, bl);
+      ::decode(master_ver, bl);
+    } else {
+      ver = 0;
+    }
     DECODE_FINISH(bl);
   }
   void dump(Formatter *f) const;
diff --git a/src/rgw/rgw_admin.cc b/src/rgw/rgw_admin.cc
index 29c9f690943..d9282ed7f14 100644
--- a/src/rgw/rgw_admin.cc
+++ b/src/rgw/rgw_admin.cc
@@ -416,7 +416,8 @@ int bucket_stats(rgw_bucket& bucket, Formatter *formatter)
     return r;
 
   map<RGWObjCategory, RGWBucketStats> stats;
-  int ret = store->get_bucket_stats(bucket, stats);
+  uint64_t bucket_ver, master_ver;
+  int ret = store->get_bucket_stats(bucket, &bucket_ver, &master_ver, stats);
   if (ret < 0) {
     cerr << "error getting bucket stats ret=" << ret << std::endl;
     return ret;
@@ -428,6 +429,8 @@ int bucket_stats(rgw_bucket& bucket, Formatter *formatter)
   formatter->dump_string("id", bucket.bucket_id);
   formatter->dump_string("marker", bucket.marker);
   formatter->dump_string("owner", bucket_info.owner);
+  formatter->dump_int("ver", bucket_ver);
+  formatter->dump_int("master_ver", master_ver);
   dump_bucket_usage(stats, formatter);
   formatter->close_section();
 
diff --git a/src/rgw/rgw_bucket.cc b/src/rgw/rgw_bucket.cc
index b2032264f32..aa4ead07426 100644
--- a/src/rgw/rgw_bucket.cc
+++ b/src/rgw/rgw_bucket.cc
@@ -321,7 +321,9 @@ int rgw_remove_bucket(RGWRados *store, rgw_bucket& bucket, bool delete_children)
   RGWBucketInfo info;
   bufferlist bl;
 
-  ret = store->get_bucket_stats(bucket, stats);
+  uint64_t bucket_ver, master_ver;
+
+  ret = store->get_bucket_stats(bucket, &bucket_ver, &master_ver, stats);
   if (ret < 0)
     return ret;
 
@@ -895,7 +897,8 @@ static int bucket_stats(RGWRados *store, std::string&  bucket_name, Formatter *f
 
   bucket = bucket_info.bucket;
 
-  int ret = store->get_bucket_stats(bucket, stats);
+  uint64_t bucket_ver, master_ver;
+  int ret = store->get_bucket_stats(bucket, &bucket_ver, &master_ver, stats);
   if (ret < 0) {
     cerr << "error getting bucket stats ret=" << ret << std::endl;
     return ret;
@@ -908,6 +911,8 @@ static int bucket_stats(RGWRados *store, std::string&  bucket_name, Formatter *f
   formatter->dump_string("id", bucket.bucket_id);
   formatter->dump_string("marker", bucket.marker);
   formatter->dump_string("owner", bucket_info.owner);
+  formatter->dump_int("ver", bucket_ver);
+  formatter->dump_int("master_ver", master_ver);
   dump_bucket_usage(stats, formatter);
   formatter->close_section();
 
diff --git a/src/rgw/rgw_rados.cc b/src/rgw/rgw_rados.cc
index 7bc500fea8e..79e558a681d 100644
--- a/src/rgw/rgw_rados.cc
+++ b/src/rgw/rgw_rados.cc
@@ -3699,7 +3699,7 @@ int RGWRados::obj_stat(void *ctx, rgw_obj& obj, uint64_t *psize, time_t *pmtime,
   return 0;
 }
 
-int RGWRados::get_bucket_stats(rgw_bucket& bucket, map<RGWObjCategory, RGWBucketStats>& stats)
+int RGWRados::get_bucket_stats(rgw_bucket& bucket, uint64_t *bucket_ver, uint64_t *master_ver, map<RGWObjCategory, RGWBucketStats>& stats)
 {
   rgw_bucket_dir_header header;
   int r = cls_bucket_head(bucket, header);
@@ -3710,6 +3710,9 @@ int RGWRados::get_bucket_stats(rgw_bucket& bucket, map<RGWObjCategory, RGWBucket
 
   translate_raw_stats(header, stats);
 
+  *bucket_ver = header.ver;
+  *master_ver = header.master_ver;
+
   return 0;
 }
 
diff --git a/src/rgw/rgw_rados.h b/src/rgw/rgw_rados.h
index 05a32577782..a181304fc2b 100644
--- a/src/rgw/rgw_rados.h
+++ b/src/rgw/rgw_rados.h
@@ -935,7 +935,7 @@ public:
   }
 
   int decode_policy(bufferlist& bl, ACLOwner *owner);
-  int get_bucket_stats(rgw_bucket& bucket, map<RGWObjCategory, RGWBucketStats>& stats);
+  int get_bucket_stats(rgw_bucket& bucket, uint64_t *bucket_ver, uint64_t *master_ver, map<RGWObjCategory, RGWBucketStats>& stats);
   virtual int get_bucket_info(void *ctx, string& bucket_name, RGWBucketInfo& info, map<string, bufferlist> *pattrs = NULL);
   virtual int put_bucket_info(string& bucket_name, RGWBucketInfo& info, bool exclusive, map<string, bufferlist> *pattrs);