Merge branch 'wip-rgw-versionchecks' into next

Conflicts:
	src/rgw/rgw_rest_metadata.h

Reviewed-by: Yehuda Sadeh <yehuda@inktank.com

9 files changed, 170 insertions, 62 deletions

diff --git a/src/rgw/rgw_admin.cc b/src/rgw/rgw_admin.cc
index 67f5f1c68b3..364f60f78f7 100644
--- a/src/rgw/rgw_admin.cc
+++ b/src/rgw/rgw_admin.cc
@@ -1815,7 +1815,7 @@ next:
       cerr << "ERROR: failed to read input: " << cpp_strerror(-ret) << std::endl;
       return ret;
     }
-    ret = store->meta_mgr->put(metadata_key, bl);
+    ret = store->meta_mgr->put(metadata_key, bl, RGWMetadataHandler::APPLY_ALWAYS);
     if (ret < 0) {
       cerr << "ERROR: can't put key: " << cpp_strerror(-ret) << std::endl;
       return -ret;
diff --git a/src/rgw/rgw_bucket.cc b/src/rgw/rgw_bucket.cc
index 8de5a3d101f..bf8da99d616 100644
--- a/src/rgw/rgw_bucket.cc
+++ b/src/rgw/rgw_bucket.cc
@@ -1397,7 +1397,8 @@ public:
     return 0;
   }
 
-  int put(RGWRados *store, string& entry, RGWObjVersionTracker& objv_tracker, time_t mtime, JSONObj *obj) {
+  int put(RGWRados *store, string& entry, RGWObjVersionTracker& objv_tracker,
+          time_t mtime, JSONObj *obj, sync_type_t sync_type) {
     RGWBucketEntryPoint be, old_be;
     decode_json_obj(be, obj);
 
@@ -1410,6 +1411,12 @@ public:
     if (ret < 0 && ret != -ENOENT)
       return ret;
 
+    // are we actually going to perform this put, or is it too old?
+    if (!check_versions(old_ot.read_version, orig_mtime,
+			objv_tracker.write_version, mtime, sync_type)) {
+      return STATUS_NO_APPLY;
+    }
+
     objv_tracker.read_version = old_ot.read_version; /* maintain the obj version we just read */
 
     ret = store->put_bucket_entrypoint_info(entry, be, false, objv_tracker, mtime, &attrs);
@@ -1540,7 +1547,8 @@ public:
     return 0;
   }
 
-  int put(RGWRados *store, string& oid, RGWObjVersionTracker& objv_tracker, time_t mtime, JSONObj *obj) {
+  int put(RGWRados *store, string& oid, RGWObjVersionTracker& objv_tracker,
+          time_t mtime, JSONObj *obj, sync_type_t sync_type) {
     RGWBucketCompleteInfo bci, old_bci;
     decode_json_obj(bci, obj);
 
@@ -1566,6 +1574,13 @@ public:
       bci.info.bucket.index_pool = old_bci.info.bucket.index_pool;
     }
 
+    // are we actually going to perform this put, or is it too old?
+    if (!check_versions(old_bci.info.objv_tracker.read_version, orig_mtime,
+			objv_tracker.write_version, mtime, sync_type)) {
+      objv_tracker.read_version = old_bci.info.objv_tracker.read_version;
+      return STATUS_NO_APPLY;
+    }
+
     /* record the read version (if any), store the new version */
     bci.info.objv_tracker.read_version = old_bci.info.objv_tracker.read_version;
     bci.info.objv_tracker.write_version = objv_tracker.write_version;
@@ -1580,7 +1595,7 @@ public:
     if (ret < 0)
       return ret;
 
-    return 0;
+    return STATUS_APPLIED;
   }
 
   struct list_keys_info {
diff --git a/src/rgw/rgw_common.h b/src/rgw/rgw_common.h
index 7f224a798f5..543bdf21377 100644
--- a/src/rgw/rgw_common.h
+++ b/src/rgw/rgw_common.h
@@ -99,6 +99,8 @@ using ceph::crypto::MD5;
 #define STATUS_NO_CONTENT        1902
 #define STATUS_PARTIAL_CONTENT   1903
 #define STATUS_REDIRECT          1904
+#define STATUS_NO_APPLY          1905
+#define STATUS_APPLIED           1906
 
 #define ERR_INVALID_BUCKET_NAME  2000
 #define ERR_INVALID_OBJECT_NAME  2001
diff --git a/src/rgw/rgw_metadata.cc b/src/rgw/rgw_metadata.cc
index e9094ad35d6..8fc79e69b12 100644
--- a/src/rgw/rgw_metadata.cc
+++ b/src/rgw/rgw_metadata.cc
@@ -194,7 +194,8 @@ public:
   virtual string get_type() { return string(); }
 
   virtual int get(RGWRados *store, string& entry, RGWMetadataObject **obj) { return -ENOTSUP; }
-  virtual int put(RGWRados *store, string& entry, RGWObjVersionTracker& objv_tracker, time_t mtime, JSONObj *obj) { return -ENOTSUP; }
+  virtual int put(RGWRados *store, string& entry, RGWObjVersionTracker& objv_tracker,
+                  time_t mtime, JSONObj *obj, sync_type_t sync_type) { return -ENOTSUP; }
 
   virtual void get_pool_and_oid(RGWRados *store, const string& key, rgw_bucket& bucket, string& oid) {}
 
@@ -328,7 +329,9 @@ int RGWMetadataManager::get(string& metadata_key, Formatter *f)
   return 0;
 }
 
-int RGWMetadataManager::put(string& metadata_key, bufferlist& bl)
+int RGWMetadataManager::put(string& metadata_key, bufferlist& bl,
+                            RGWMetadataHandler::sync_type_t sync_type,
+                            obj_version *existing_version)
 {
   RGWMetadataHandler *handler;
   string entry;
@@ -357,7 +360,11 @@ int RGWMetadataManager::put(string& metadata_key, bufferlist& bl)
     return -EINVAL;
   }
 
-  return handler->put(store, entry, objv_tracker, mtime, jo);
+  ret = handler->put(store, entry, objv_tracker, mtime, jo, sync_type);
+  if (existing_version) {
+    *existing_version = objv_tracker.read_version;
+  }
+  return ret;
 }
 
 int RGWMetadataManager::remove(string& metadata_key)
diff --git a/src/rgw/rgw_metadata.h b/src/rgw/rgw_metadata.h
index 3ff3b3317b8..7880d1fda22 100644
--- a/src/rgw/rgw_metadata.h
+++ b/src/rgw/rgw_metadata.h
@@ -44,14 +44,30 @@ class RGWMetadataManager;
 class RGWMetadataHandler {
   friend class RGWMetadataManager;
 
-protected:
-  virtual void get_pool_and_oid(RGWRados *store, const string& key, rgw_bucket& bucket, string& oid) = 0;
 public:
+  enum sync_type_t {
+    APPLY_ALWAYS,
+    APPLY_UPDATES,
+    APPLY_NEWER
+  };
+  static bool string_to_sync_type(const string& sync_string,
+                                  sync_type_t& type) {
+    if (sync_string.compare("update-by-version") == 0)
+      type = APPLY_UPDATES;
+    else if (sync_string.compare("update-by-timestamp") == 0)
+      type = APPLY_NEWER;
+    else if (sync_string.compare("always") == 0)
+      type = APPLY_ALWAYS;
+    else
+      return false;
+    return true;
+  }
   virtual ~RGWMetadataHandler() {}
   virtual string get_type() = 0;
 
   virtual int get(RGWRados *store, string& entry, RGWMetadataObject **obj) = 0;
-  virtual int put(RGWRados *store, string& entry, RGWObjVersionTracker& objv_tracker, time_t mtime, JSONObj *obj) = 0;
+  virtual int put(RGWRados *store, string& entry, RGWObjVersionTracker& objv_tracker,
+                  time_t mtime, JSONObj *obj, sync_type_t type) = 0;
   virtual int remove(RGWRados *store, string& entry, RGWObjVersionTracker& objv_tracker) = 0;
 
   virtual int list_keys_init(RGWRados *store, void **phandle) = 0;
@@ -62,6 +78,33 @@ public:
   virtual void get_hash_key(const string& section, const string& key, string& hash_key) {
     hash_key = section + ":" + key;
   }
+
+protected:
+  virtual void get_pool_and_oid(RGWRados *store, const string& key, rgw_bucket& bucket, string& oid) = 0;
+  /**
+   * Compare an incoming versus on-disk tag/version+mtime combo against
+   * the sync mode to see if the new one should replace the on-disk one.
+   *
+   * @return true if the update should proceed, false otherwise.
+   */
+  bool check_versions(const obj_version& ondisk, const time_t& ondisk_time,
+                      const obj_version& incoming, const time_t& incoming_time,
+                      sync_type_t sync_mode) {
+    switch (sync_mode) {
+    case APPLY_UPDATES:
+      if ((ondisk.tag != incoming.tag) ||
+	  (ondisk.ver >= incoming.ver))
+	return false;
+      break;
+    case APPLY_NEWER:
+      if (ondisk_time >= incoming_time)
+	return false;
+      break;
+    case APPLY_ALWAYS: //deliberate fall-thru -- we always apply!
+    default: break;
+    }
+    return true;
+  }
 };
 
 #define META_LOG_OBJ_PREFIX "meta.log."
@@ -152,7 +195,9 @@ public:
                 map<string, bufferlist>* rmattrs,
                 RGWObjVersionTracker *objv_tracker);
   int get(string& metadata_key, Formatter *f);
-  int put(string& metadata_key, bufferlist& bl);
+  int put(string& metadata_key, bufferlist& bl,
+          RGWMetadataHandler::sync_type_t sync_mode,
+          obj_version *existing_version = NULL);
   int remove(string& metadata_key);
 
   int list_keys_init(string& section, void **phandle);
diff --git a/src/rgw/rgw_rest_metadata.cc b/src/rgw/rgw_rest_metadata.cc
index 35ec0ab9b04..0705a46ed6c 100644
--- a/src/rgw/rgw_rest_metadata.cc
+++ b/src/rgw/rgw_rest_metadata.cc
@@ -161,12 +161,43 @@ void RGWOp_Metadata_Put::execute() {
   
   frame_metadata_key(s, metadata_key);
   
-  http_ret = store->meta_mgr->put(metadata_key, bl);
+  RGWMetadataHandler::sync_type_t sync_type = RGWMetadataHandler::APPLY_ALWAYS;
+
+  bool mode_exists = false;
+  string mode_string = s->info.args.get("sync-type", &mode_exists);
+  if (mode_exists) {
+    bool parsed = RGWMetadataHandler::string_to_sync_type(mode_string,
+                                                          sync_type);
+    if (!parsed) {
+      http_ret = -EINVAL;
+      return;
+    }
+  }
+
+  http_ret = store->meta_mgr->put(metadata_key, bl, sync_type, &ondisk_version);
   if (http_ret < 0) {
     dout(5) << "ERROR: can't put key: " << cpp_strerror(http_ret) << dendl;
     return;
   }
-  http_ret = 0;
+  // translate internal codes into return header
+  if (http_ret == STATUS_NO_APPLY)
+    update_status = "skipped";
+  else if (http_ret == STATUS_APPLIED)
+    update_status = "applied";
+}
+
+void RGWOp_Metadata_Put::send_response() {
+  int http_return_code = http_ret;
+  if ((http_ret == STATUS_NO_APPLY) || (http_ret == STATUS_APPLIED))
+    http_return_code = STATUS_NO_CONTENT;
+  set_req_state_err(s, http_return_code);
+  dump_errno(s);
+  stringstream ver_stream;
+  ver_stream << "ver:" << ondisk_version.ver
+	     <<",tag:" << ondisk_version.tag;
+  dump_pair(s, "RGWX_UPDATE_STATUS", update_status.c_str());
+  dump_pair(s, "RGWX_UPDATE_VERSION", ver_stream.str().c_str());
+  end_header(s);
 }
 
 void RGWOp_Metadata_Delete::execute() {
diff --git a/src/rgw/rgw_rest_metadata.h b/src/rgw/rgw_rest_metadata.h
index 59d7c5f7045..7f3cf1f2207 100644
--- a/src/rgw/rgw_rest_metadata.h
+++ b/src/rgw/rgw_rest_metadata.h
@@ -40,6 +40,8 @@ public:
 
 class RGWOp_Metadata_Put : public RGWRESTOp {
   int get_data(bufferlist& bl);
+  string update_status;
+  obj_version ondisk_version;
 public:
   RGWOp_Metadata_Put() {}
   ~RGWOp_Metadata_Put() {}
@@ -48,6 +50,7 @@ public:
     return caps.check_cap("metadata", RGW_CAP_WRITE);
   }
   void execute();
+  void send_response();
   virtual const string name() { return "set_metadata"; }
 };
 
diff --git a/src/rgw/rgw_user.cc b/src/rgw/rgw_user.cc
index 6cdcaa62935..6fcecd4a98d 100644
--- a/src/rgw/rgw_user.cc
+++ b/src/rgw/rgw_user.cc
@@ -2295,22 +2295,29 @@ public:
     return 0;
   }
 
-  int put(RGWRados *store, string& entry, RGWObjVersionTracker& objv_tracker, time_t mtime, JSONObj *obj) {
+  int put(RGWRados *store, string& entry, RGWObjVersionTracker& objv_tracker,
+          time_t mtime, JSONObj *obj, sync_type_t sync_mode) {
     RGWUserInfo info;
 
     decode_json_obj(info, obj);
 
     RGWUserInfo old_info;
-    int ret = rgw_get_user_info_by_uid(store, entry, old_info, &objv_tracker);
+    time_t orig_mtime;
+    int ret = rgw_get_user_info_by_uid(store, entry, old_info, &objv_tracker, &orig_mtime);
     if (ret < 0 && ret != -ENOENT)
       return ret;
 
+    // are we actually going to perform this put, or is it too old?
+    if (!check_versions(objv_tracker.read_version, orig_mtime,
+			objv_tracker.write_version, mtime, sync_mode)) {
+      return STATUS_NO_APPLY;
+    }
 
     ret = rgw_store_user_info(store, info, &old_info, &objv_tracker, mtime, false);
     if (ret < 0)
       return ret;
 
-    return 0;
+    return STATUS_APPLIED;
   }
 
   struct list_keys_info {
diff --git a/src/test/cls_replica_log/test_cls_replica_log.cc b/src/test/cls_replica_log/test_cls_replica_log.cc
index eabe0b3860d..8c204caef04 100644
--- a/src/test/cls_replica_log/test_cls_replica_log.cc
+++ b/src/test/cls_replica_log/test_cls_replica_log.cc
@@ -14,34 +14,42 @@
 #include "cls/replica_log/cls_replica_log_client.h"
 #include "cls/replica_log/cls_replica_log_types.h"
 
-#define SETUP_DATA \
-  librados::Rados rados; \
-  librados::IoCtx ioctx; \
-  string pool_name = get_temp_pool_name(); \
-  ASSERT_EQ("", create_one_pool_pp(pool_name, rados)); \
-  ASSERT_EQ(0, rados.ioctx_create(pool_name.c_str(), ioctx)); \
-  string oid = "obj"; \
-  ASSERT_EQ(0, ioctx.create(oid, true));
-
-#define ADD_MARKER \
-  string entity = "tester_entity"; \
-  string marker = "tester_marker1"; \
-  utime_t time; \
-  time.set_from_double(10); \
-  list<pair<string, utime_t> > entries; \
-  entries.push_back(make_pair("tester_obj1", time)); \
-  time.set_from_double(20); \
-  cls_replica_log_progress_marker progress; \
-  cls_replica_log_prepare_marker(progress, entity, marker, time, &entries); \
-  librados::ObjectWriteOperation opw; \
-  cls_replica_log_update_bound(opw, progress); \
-  ASSERT_EQ(0, ioctx.operate(oid, &opw));
-
-TEST(cls_replica_log, test_set_get_marker)
+class cls_replica_log_Test : public ::testing::Test {
+public:
+  librados::Rados rados;
+  librados::IoCtx ioctx;
+  string pool_name;
+  string oid;
+  string entity;
+  string marker;
+  utime_t time;
+  list<pair<string, utime_t> > entries;
+  cls_replica_log_progress_marker progress;
+
+  void SetUp() {
+    pool_name = get_temp_pool_name();
+    ASSERT_EQ("", create_one_pool_pp(pool_name, rados));
+    ASSERT_EQ(0, rados.ioctx_create(pool_name.c_str(), ioctx));
+    oid = "obj";
+    ASSERT_EQ(0, ioctx.create(oid, true));
+  }
+
+  void add_marker() {
+    entity = "tester_entity";
+    marker = "tester_marker1";
+    time.set_from_double(10);
+    entries.push_back(make_pair("tester_obj1", time));
+    time.set_from_double(20);
+    cls_replica_log_prepare_marker(progress, entity, marker, time, &entries);
+    librados::ObjectWriteOperation opw;
+    cls_replica_log_update_bound(opw, progress);
+    ASSERT_EQ(0, ioctx.operate(oid, &opw));
+  }
+};
+
+TEST_F(cls_replica_log_Test, test_set_get_marker)
 {
-  SETUP_DATA
-
-  ADD_MARKER
+  add_marker();
 
   string reply_position_marker;
   utime_t reply_time;
@@ -66,11 +74,9 @@ TEST(cls_replica_log, test_set_get_marker)
   ASSERT_EQ("tester_obj1", response_item_list.front().first);
 }
 
-TEST(cls_replica_log, test_bad_update)
+TEST_F(cls_replica_log_Test, test_bad_update)
 {
-  SETUP_DATA
-
-  ADD_MARKER
+  add_marker();
 
   time.set_from_double(15);
   cls_replica_log_progress_marker bad_marker;
@@ -80,22 +86,18 @@ TEST(cls_replica_log, test_bad_update)
   ASSERT_EQ(-EINVAL, ioctx.operate(oid, &badw));
 }
 
-TEST(cls_replica_log, test_bad_delete)
+TEST_F(cls_replica_log_Test, test_bad_delete)
 {
-  SETUP_DATA
-
-  ADD_MARKER
+  add_marker();
 
   librados::ObjectWriteOperation badd;
   cls_replica_log_delete_bound(badd, entity);
   ASSERT_EQ(-ENOTEMPTY, ioctx.operate(oid, &badd));
 }
 
-TEST(cls_replica_log, test_good_delete)
+TEST_F(cls_replica_log_Test, test_good_delete)
 {
-  SETUP_DATA
-
-  ADD_MARKER
+  add_marker();
 
   librados::ObjectWriteOperation opc;
   progress.items.clear();
@@ -113,10 +115,8 @@ TEST(cls_replica_log, test_good_delete)
   ASSERT_EQ((unsigned)0, return_progress_list.size());
 }
 
-TEST(cls_replica_log, test_bad_get)
+TEST_F(cls_replica_log_Test, test_bad_get)
 {
-  SETUP_DATA
-
   string reply_position_marker;
   utime_t reply_time;
   list<cls_replica_log_progress_marker> return_progress_list;
@@ -125,11 +125,9 @@ TEST(cls_replica_log, test_bad_get)
                                        reply_time, return_progress_list));
 }
 
-TEST(cls_replica_log, test_double_delete)
+TEST_F(cls_replica_log_Test, test_double_delete)
 {
-  SETUP_DATA
-
-  ADD_MARKER
+  add_marker();
 
   librados::ObjectWriteOperation opc;
   progress.items.clear();