Merge branch 'wip-rgw-next-2' into next

Reviewed-by: Greg Farnum <greg@inktank.com>

15 files changed, 158 insertions, 95 deletions

diff --git a/src/cls/rgw/cls_rgw.cc b/src/cls/rgw/cls_rgw.cc
index a972d6a1fdd..de2abe5665b 100644
--- a/src/cls/rgw/cls_rgw.cc
+++ b/src/cls/rgw/cls_rgw.cc
@@ -656,7 +656,7 @@ int rgw_bucket_complete_op(cls_method_context_t hctx, bufferlist *in, bufferlist
   }
 
   list<string>::iterator remove_iter;
-  CLS_LOG(0, "rgw_bucket_complete_op(): remove_objs.size()=%d\n", (int)op.remove_objs.size());
+  CLS_LOG(20, "rgw_bucket_complete_op(): remove_objs.size()=%d\n", (int)op.remove_objs.size());
   for (remove_iter = op.remove_objs.begin(); remove_iter != op.remove_objs.end(); ++remove_iter) {
     string& remove_oid_name = *remove_iter;
     CLS_LOG(1, "rgw_bucket_complete_op(): removing entries, read_index_entry name=%s\n", remove_oid_name.c_str());
diff --git a/src/rgw/rgw_admin.cc b/src/rgw/rgw_admin.cc
index abf9e26c8e3..e5d880f7d24 100644
--- a/src/rgw/rgw_admin.cc
+++ b/src/rgw/rgw_admin.cc
@@ -1389,11 +1389,19 @@ int main(int argc, char **argv)
   }
 
   if (opt_cmd == OPT_BUCKET_LINK) {
-   RGWBucketAdminOp::link(store, bucket_op);
+    int r = RGWBucketAdminOp::link(store, bucket_op);
+    if (r < 0) {
+      cerr << "failure: " << cpp_strerror(-r) << std::endl;
+      return -r;
+    }
   }
 
   if (opt_cmd == OPT_BUCKET_UNLINK) {
-    RGWBucketAdminOp::unlink(store, bucket_op);
+    int r = RGWBucketAdminOp::unlink(store, bucket_op);
+    if (r < 0) {
+      cerr << "failure: " << cpp_strerror(-r) << std::endl;
+      return -r;
+    }
   }
 
   if (opt_cmd == OPT_TEMP_REMOVE) {
diff --git a/src/rgw/rgw_bucket.cc b/src/rgw/rgw_bucket.cc
index aae7d31e21c..7b22f44790b 100644
--- a/src/rgw/rgw_bucket.cc
+++ b/src/rgw/rgw_bucket.cc
@@ -718,7 +718,7 @@ int RGWBucket::get_policy(RGWBucketAdminOpState& op_state, ostream& o)
 
   bufferlist bl;
   rgw_obj obj(bucket, object_name);
-  int ret = store->get_attr(NULL, obj, RGW_ATTR_ACL, bl, NULL);
+  int ret = store->get_attr(NULL, obj, RGW_ATTR_ACL, bl);
   if (ret < 0)
     return ret;
 
diff --git a/src/rgw/rgw_cache.h b/src/rgw/rgw_cache.h
index 1a36e1a78d2..b6c4e15eede 100644
--- a/src/rgw/rgw_cache.h
+++ b/src/rgw/rgw_cache.h
@@ -208,7 +208,7 @@ public:
   int obj_stat(void *ctx, rgw_obj& obj, uint64_t *psize, time_t *pmtime, uint64_t *epoch, map<string, bufferlist> *attrs,
                bufferlist *first_chunk, RGWObjVersionTracker *objv_tracker);
 
-  int delete_obj(void *ctx, rgw_obj& obj);
+  int delete_obj(void *ctx, rgw_obj& obj, RGWObjVersionTracker *objv_tracker);
 };
 
 template <class T>
@@ -224,13 +224,13 @@ void RGWCache<T>::normalize_bucket_and_obj(rgw_bucket& src_bucket, string& src_o
 }
 
 template <class T>
-int RGWCache<T>::delete_obj(void *ctx, rgw_obj& obj)
+int RGWCache<T>::delete_obj(void *ctx, rgw_obj& obj, RGWObjVersionTracker *objv_tracker)
 {
   rgw_bucket bucket;
   string oid;
   normalize_bucket_and_obj(obj.bucket, obj.object, bucket, oid);
   if (bucket.name[0] != '.')
-    return T::delete_obj(ctx, obj);
+    return T::delete_obj(ctx, obj, objv_tracker);
 
   string name = normal_name(obj);
   cache.remove(name);
@@ -238,7 +238,7 @@ int RGWCache<T>::delete_obj(void *ctx, rgw_obj& obj)
   ObjectCacheInfo info;
   distribute_cache(name, obj, info, REMOVE_OBJ);
 
-  return T::delete_obj(ctx, obj);
+  return T::delete_obj(ctx, obj, objv_tracker);
 }
 
 template <class T>
diff --git a/src/rgw/rgw_common.h b/src/rgw/rgw_common.h
index c885724efbd..1d3596d4418 100644
--- a/src/rgw/rgw_common.h
+++ b/src/rgw/rgw_common.h
@@ -653,6 +653,7 @@ struct RGWBucketInfo
   string placement_rule;
   bool has_instance_obj;
   RGWObjVersionTracker objv_tracker; /* we don't need to serialize this, for runtime tracking */
+  obj_version ep_objv; /* entry point object version, for runtime tracking only */
 
   void encode(bufferlist& bl) const {
      ENCODE_START(8, 4, bl);
@@ -807,8 +808,6 @@ struct req_state {
    map<string, bufferlist> bucket_attrs;
    bool bucket_exists;
 
-   RGWObjVersionTracker objv_tracker;
-
    bool has_bad_meta;
 
    RGWUserInfo user; 
diff --git a/src/rgw/rgw_metadata.cc b/src/rgw/rgw_metadata.cc
index c370addc293..7be73e6ca0c 100644
--- a/src/rgw/rgw_metadata.cc
+++ b/src/rgw/rgw_metadata.cc
@@ -588,7 +588,7 @@ int RGWMetadataManager::remove_entry(RGWMetadataHandler *handler, string& key, R
 
   rgw_obj obj(bucket, oid);
 
-  ret = store->delete_obj(NULL, obj);
+  ret = store->delete_obj(NULL, obj, objv_tracker);
   /* cascading ret into post_modify() */
 
   ret = post_modify(handler, section, key, log_data, objv_tracker, ret);
diff --git a/src/rgw/rgw_op.cc b/src/rgw/rgw_op.cc
index 05c31d61689..17a3aaa8439 100644
--- a/src/rgw/rgw_op.cc
+++ b/src/rgw/rgw_op.cc
@@ -178,8 +178,7 @@ static int decode_policy(CephContext *cct, bufferlist& bl, RGWAccessControlPolic
 
 static int get_bucket_policy_from_attr(CephContext *cct, RGWRados *store, void *ctx,
                                        RGWBucketInfo& bucket_info, map<string, bufferlist>& bucket_attrs,
-                                       RGWAccessControlPolicy *policy, rgw_obj& obj,
-                                       RGWObjVersionTracker *objv_tracker)
+                                       RGWAccessControlPolicy *policy, rgw_obj& obj)
 {
   int ret;
   map<string, bufferlist>::iterator aiter = bucket_attrs.find(RGW_ATTR_ACL);
@@ -203,13 +202,12 @@ static int get_bucket_policy_from_attr(CephContext *cct, RGWRados *store, void *
 
 static int get_obj_policy_from_attr(CephContext *cct, RGWRados *store, void *ctx,
                                     RGWBucketInfo& bucket_info, map<string, bufferlist>& bucket_attrs,
-                                    RGWAccessControlPolicy *policy, rgw_obj& obj,
-                                    RGWObjVersionTracker *objv_tracker)
+                                    RGWAccessControlPolicy *policy, rgw_obj& obj)
 {
   bufferlist bl;
   int ret = 0;
 
-  ret = store->get_attr(ctx, obj, RGW_ATTR_ACL, bl, objv_tracker);
+  ret = store->get_attr(ctx, obj, RGW_ATTR_ACL, bl);
   if (ret >= 0) {
     ret = decode_policy(cct, bl, policy);
     if (ret < 0)
@@ -237,7 +235,7 @@ static int get_obj_policy_from_attr(CephContext *cct, RGWRados *store, void *ctx
  */
 static int get_policy_from_attr(CephContext *cct, RGWRados *store, void *ctx,
                                 RGWBucketInfo& bucket_info, map<string, bufferlist>& bucket_attrs,
-                                RGWAccessControlPolicy *policy, rgw_obj& obj, RGWObjVersionTracker *objv_tracker)
+                                RGWAccessControlPolicy *policy, rgw_obj& obj)
 {
   if (obj.bucket.name.empty()) {
     return 0;
@@ -245,10 +243,10 @@ static int get_policy_from_attr(CephContext *cct, RGWRados *store, void *ctx,
 
   if (obj.object.empty()) {
     return get_bucket_policy_from_attr(cct, store, ctx, bucket_info, bucket_attrs,
-                                       policy, obj, objv_tracker);
+                                       policy, obj);
   }
   return get_obj_policy_from_attr(cct, store, ctx, bucket_info, bucket_attrs,
-                                  policy, obj, objv_tracker);
+                                  policy, obj);
 }
 
 static int get_obj_attrs(RGWRados *store, struct req_state *s, rgw_obj& obj, map<string, bufferlist>& attrs,
@@ -282,14 +280,14 @@ static int read_policy(RGWRados *store, struct req_state *s,
   } else {
     obj.init(bucket, oid);
   }
-  int ret = get_policy_from_attr(s->cct, store, s->obj_ctx, bucket_info, bucket_attrs, policy, obj, &s->objv_tracker);
+  int ret = get_policy_from_attr(s->cct, store, s->obj_ctx, bucket_info, bucket_attrs, policy, obj);
   if (ret == -ENOENT && object.size()) {
     /* object does not exist checking the bucket's ACL to make sure
        that we send a proper error code */
     RGWAccessControlPolicy bucket_policy(s->cct);
     string no_object;
     rgw_obj no_obj(bucket, no_object);
-    ret = get_policy_from_attr(s->cct, store, s->obj_ctx, bucket_info, bucket_attrs, &bucket_policy, no_obj, &s->objv_tracker);
+    ret = get_policy_from_attr(s->cct, store, s->obj_ctx, bucket_info, bucket_attrs, &bucket_policy, no_obj);
     if (ret < 0)
       return ret;
     string& owner = bucket_policy.get_owner().get_id();
@@ -336,8 +334,7 @@ static int rgw_build_policies(RGWRados *store, struct req_state *s, bool only_bu
     ret = store->get_bucket_info(s->obj_ctx, copy_source_str, source_info, NULL);
     if (ret == 0) {
       string& region = source_info.region;
-      s->local_source = (region.empty() && store->region.is_master) ||
-                         (region == store->region.name);
+      s->local_source = store->region.equals(region);
     }
   }
     
@@ -364,8 +361,7 @@ static int rgw_build_policies(RGWRados *store, struct req_state *s, bool only_bu
     s->bucket_owner = s->bucket_acl->get_owner();
 
     string& region = s->bucket_info.region;
-    if (s->bucket_exists && ((region.empty() && !store->region.is_master) ||
-        (region != store->region.name))) {
+    if (s->bucket_exists && !store->region.equals(region)) {
       ldout(s->cct, 0) << "NOTICE: request for data in a different region (" << region << " != " << store->region.name << ")" << dendl;
       /* we now need to make sure that the operation actually requires copy source, that is
        * it's a copy operation
@@ -922,7 +918,7 @@ int RGWCreateBucket::verify_permission()
   return 0;
 }
 
-static int forward_request_to_master(struct req_state *s, RGWRados *store, bufferlist& in_data, JSONParser *jp)
+static int forward_request_to_master(struct req_state *s, obj_version *objv, RGWRados *store, bufferlist& in_data, JSONParser *jp)
 {
   if (!store->rest_master_conn) {
     ldout(s->cct, 0) << "rest connection is invalid" << dendl;
@@ -931,7 +927,7 @@ static int forward_request_to_master(struct req_state *s, RGWRados *store, buffe
   ldout(s->cct, 0) << "sending create_bucket request to master region" << dendl;
   bufferlist response;
 #define MAX_REST_RESPONSE (128 * 1024) // we expect a very small response
-  int ret = store->rest_master_conn->forward(s->user.user_id, s->info, MAX_REST_RESPONSE, &in_data, &response);
+  int ret = store->rest_master_conn->forward(s->user.user_id, s->info, objv, MAX_REST_RESPONSE, &in_data, &response);
   if (ret < 0)
     return ret;
 
@@ -976,7 +972,7 @@ void RGWCreateBucket::execute()
   s->bucket_owner.set_name(s->user.display_name);
   if (s->bucket_exists) {
     r = get_policy_from_attr(s->cct, store, s->obj_ctx, s->bucket_info, s->bucket_attrs,
-                             &old_policy, obj, &s->objv_tracker);
+                             &old_policy, obj);
     if (r >= 0)  {
       if (old_policy.get_owner().get_id().compare(s->user.user_id) != 0) {
         ret = -EEXIST;
@@ -991,10 +987,11 @@ void RGWCreateBucket::execute()
 
   if (!store->region.is_master) {
     JSONParser jp;
-    ret = forward_request_to_master(s, store, in_data, &jp);
+    ret = forward_request_to_master(s, NULL, store, in_data, &jp);
     if (ret < 0)
       return;
 
+    JSONDecoder::decode_json("entry_point_object_ver", ep_objv, &jp);
     JSONDecoder::decode_json("object_ver", objv, &jp);
     JSONDecoder::decode_json("bucket_info", master_info, &jp);
     ldout(s->cct, 20) << "parsed: objv.tag=" << objv.tag << " objv.ver=" << objv.ver << dendl;
@@ -1024,7 +1021,7 @@ void RGWCreateBucket::execute()
 
   s->bucket.name = s->bucket_name_str;
   ret = store->create_bucket(s->user, s->bucket, region_name, placement_rule, attrs, info, pobjv,
-                             creation_time, pmaster_bucket, true);
+                             &ep_objv, creation_time, pmaster_bucket, true);
   /* continue if EEXIST and create_bucket will fail below.  this way we can recover
    * from a partial create by retrying it. */
   ldout(s->cct, 20) << "rgw_create_bucket returned ret=" << ret << " bucket=" << s->bucket << dendl;
@@ -1070,24 +1067,52 @@ void RGWDeleteBucket::execute()
   if (!s->bucket_name)
     return;
 
-  if (!store->region.is_master) {
-    bufferlist in_data;
-    JSONParser jp;
-    ret = forward_request_to_master(s, store, in_data, &jp);
-    if (ret < 0)
-      return;
+  RGWObjVersionTracker ot;
+  ot.read_version = s->bucket_info.ep_objv;
 
-    JSONDecoder::decode_json("object_ver", objv_tracker.read_version, &jp);
+  if (s->system_request) {
+    string tag = s->info.args.get(RGW_SYS_PARAM_PREFIX "tag");
+    string ver_str = s->info.args.get(RGW_SYS_PARAM_PREFIX "ver");
+    if (!tag.empty()) {
+      ot.read_version.tag = tag;
+      uint64_t ver;
+      string err;
+      ver = strict_strtol(ver_str.c_str(), 10, &err);
+      if (!err.empty()) {
+        ldout(s->cct, 0) << "failed to parse ver param" << dendl;
+        ret = -EINVAL;
+        return;
+      }
+      ot.read_version.ver = ver;
+    }
   }
 
-  ret = store->delete_bucket(s->bucket, objv_tracker);
+  ret = store->delete_bucket(s->bucket, ot);
 
   if (ret == 0) {
     ret = rgw_unlink_bucket(store, s->user.user_id, s->bucket.name, false);
     if (ret < 0) {
-      ldout(s->cct, 0) << "WARNING: failed to remove bucket: ret=" << ret << dendl;
+      ldout(s->cct, 0) << "WARNING: failed to unlink bucket: ret=" << ret << dendl;
+    }
+  }
+
+  if (ret < 0) {
+    return;
+  }
+
+  if (!store->region.is_master) {
+    bufferlist in_data;
+    JSONParser jp;
+    ret = forward_request_to_master(s, &ot.read_version, store, in_data, &jp);
+    if (ret < 0) {
+      if (ret == -ENOENT) { /* adjust error,
+                               we want to return with NoSuchBucket and not NoSuchKey */
+        ret = -ERR_NO_SUCH_BUCKET;
+      }
+      return;
     }
   }
+
 }
 
 int RGWPutObj::verify_permission()
@@ -1455,7 +1480,7 @@ void RGWPutMetadata::execute()
   rgw_get_request_metadata(s->cct, s->info, attrs);
 
   /* no need to track object versioning, need it for bucket's data only */
-  RGWObjVersionTracker *ptracker = (s->object ? NULL : &s->objv_tracker);
+  RGWObjVersionTracker *ptracker = (s->object ? NULL : &s->bucket_info.objv_tracker);
 
   /* check if obj exists, read orig attrs */
   ret = get_obj_attrs(store, s, obj, orig_attrs, NULL, ptracker);
@@ -1773,7 +1798,7 @@ void RGWPutACLs::execute()
     *_dout << dendl;
   }
 
-  RGWObjVersionTracker *ptracker = (s->object ? NULL : &s->objv_tracker);
+  RGWObjVersionTracker *ptracker = (s->object ? NULL : &s->bucket_info.objv_tracker);
 
   new_policy.encode(bl);
   obj.init(s->bucket, s->object_str);
@@ -1851,7 +1876,7 @@ void RGWPutCORS::execute()
     *_dout << dendl;
   }
 
-  RGWObjVersionTracker *ptracker = (s->object ? NULL : &s->objv_tracker);
+  RGWObjVersionTracker *ptracker = (s->object ? NULL : &s->bucket_info.objv_tracker);
 
   string no_obj;
   cors_config->encode(bl);
@@ -1883,7 +1908,7 @@ void RGWDeleteCORS::execute()
   map<string, bufferlist> orig_attrs, attrs, rmattrs;
   map<string, bufferlist>::iterator iter;
 
-  RGWObjVersionTracker *ptracker = (s->object ? NULL : &s->objv_tracker);
+  RGWObjVersionTracker *ptracker = (s->object ? NULL : &s->bucket_info.objv_tracker);
 
   /* check if obj exists, read orig attrs */
   ret = get_obj_attrs(store, s, obj, orig_attrs, NULL, ptracker);
@@ -2474,7 +2499,7 @@ int RGWHandler::read_cors_config(void)
   string no_object;
   rgw_obj no_obj(s->bucket, no_object);
   if (no_obj.bucket.name.size()) {
-    ret = store->get_attr(s->obj_ctx, no_obj, RGW_ATTR_CORS, bl, NULL);
+    ret = store->get_attr(s->obj_ctx, no_obj, RGW_ATTR_CORS, bl);
     if (ret >= 0) {
       bufferlist::iterator iter = bl.begin();
       s->bucket_cors = new RGWCORSConfiguration();
diff --git a/src/rgw/rgw_op.h b/src/rgw/rgw_op.h
index 7a2e4920ba8..7bca53b5e43 100644
--- a/src/rgw/rgw_op.h
+++ b/src/rgw/rgw_op.h
@@ -243,6 +243,7 @@ protected:
   string location_constraint;
   string placement_rule;
   RGWBucketInfo info;
+  obj_version ep_objv;
 
   bufferlist in_data;
 
diff --git a/src/rgw/rgw_rados.cc b/src/rgw/rgw_rados.cc
index 067a1adbabd..087fdcf8e09 100644
--- a/src/rgw/rgw_rados.cc
+++ b/src/rgw/rgw_rados.cc
@@ -259,6 +259,13 @@ int RGWRegion::store_info(bool exclusive)
   return ret;
 }
 
+int RGWRegion::equals(const string& other_region)
+{
+  if (is_master && other_region.empty())
+    return true;
+
+  return (name == other_region);
+}
 
 void RGWZoneParams::init_default(RGWRados *store)
 {
@@ -429,8 +436,7 @@ int RGWRegionMap::update(RGWRegion& region)
 {
   Mutex::Locker l(lock);
 
-  if (region.is_master && !master_region.empty() &&
-      master_region.compare(region.name) != 0) {
+  if (region.is_master && !region.equals(master_region)) {
     derr << "cannot update region map, master_region conflict" << dendl;
     return -EINVAL;
   }
@@ -1776,6 +1782,7 @@ int RGWRados::create_bucket(RGWUserInfo& owner, rgw_bucket& bucket,
 			    map<std::string, bufferlist>& attrs,
                             RGWBucketInfo& info,
                             obj_version *pobjv,
+                            obj_version *pep_objv,
                             time_t creation_time,
                             rgw_bucket *pmaster_bucket,
 			    bool exclusive)
@@ -1833,23 +1840,9 @@ int RGWRados::create_bucket(RGWUserInfo& owner, rgw_bucket& bucket,
       time(&info.creation_time);
     else
       info.creation_time = creation_time;
-    ret = put_linked_bucket_info(info, exclusive, 0, &attrs, true);
+    ret = put_linked_bucket_info(info, exclusive, 0, pep_objv, &attrs, true);
     if (ret == -EEXIST) {
-      /* remove bucket meta instance */
-      string entry;
-      get_bucket_instance_entry(bucket, entry);
-      r = rgw_bucket_instance_remove_entry(this, entry, &info.objv_tracker);
-      if (r < 0)
-        return r;
-
-      /* remove bucket index */
-      librados::IoCtx index_ctx; // context for new bucket
-      int r = open_bucket_index_ctx(bucket, index_ctx);
-      if (r < 0)
-        return r;
-
-      /* we need to reread the info and return it, caller will have a use for it */
-      index_ctx.remove(dir_oid);
+       /* we need to reread the info and return it, caller will have a use for it */
       r = get_bucket_info(NULL, bucket.name, info, NULL, NULL);
       if (r < 0) {
         if (r == -ENOENT) {
@@ -1858,6 +1851,24 @@ int RGWRados::create_bucket(RGWUserInfo& owner, rgw_bucket& bucket,
         ldout(cct, 0) << "get_bucket_info returned " << r << dendl;
         return r;
       }
+
+      /* only remove it if it's a different bucket instance */
+      if (info.bucket.bucket_id != bucket.bucket_id) {
+        /* remove bucket meta instance */
+        string entry;
+        get_bucket_instance_entry(bucket, entry);
+        r = rgw_bucket_instance_remove_entry(this, entry, &info.objv_tracker);
+        if (r < 0)
+          return r;
+
+        /* remove bucket index */
+        librados::IoCtx index_ctx; // context for new bucket
+        int r = open_bucket_index_ctx(bucket, index_ctx);
+        if (r < 0)
+          return r;
+
+        index_ctx.remove(dir_oid);
+      }
       /* ret == -ENOENT here */
     }
     return ret;
@@ -1933,8 +1944,7 @@ int RGWRados::set_bucket_location_by_rule(const string& location_rule, const std
   map<string, RGWZonePlacementInfo>::iterator piter = zone.placement_pools.find(location_rule);
   if (piter == zone.placement_pools.end()) {
     /* couldn't find, means we cannot really place data for this bucket in this zone */
-    if ((region_name.empty() && region.is_master) ||
-        region_name == region.name) {
+    if (region.equals(region_name)) {
       /* that's a configuration error, zone should have that rule, as we're within the requested
        * region */
       return -EINVAL;
@@ -2481,11 +2491,8 @@ int RGWRados::copy_obj(void *ctx,
   append_rand_alpha(cct, dest_obj.object, shadow_oid, 32);
   shadow_obj.init_ns(dest_obj.bucket, shadow_oid, shadow_ns);
 
-  remote_dest = ((dest_bucket_info.region.empty() && !region.is_master) ||
-      (dest_bucket_info.region != region.name));
-
-  remote_src = ((src_bucket_info.region.empty() && !region.is_master) ||
-      (src_bucket_info.region != region.name));
+  remote_dest = !region.equals(dest_bucket_info.region);
+  remote_src = !region.equals(src_bucket_info.region);
 
   if (remote_src && remote_dest) {
     ldout(cct, 0) << "ERROR: can't copy object when both src and dest buckets are remote" << dendl;
@@ -3070,7 +3077,7 @@ int RGWRados::defer_gc(void *ctx, rgw_obj& obj)
  * obj: name of the object to delete
  * Returns: 0 on success, -ERR# otherwise.
  */
-int RGWRados::delete_obj_impl(void *ctx, rgw_obj& obj)
+int RGWRados::delete_obj_impl(void *ctx, rgw_obj& obj, RGWObjVersionTracker *objv_tracker)
 {
   rgw_bucket bucket;
   std::string oid, key;
@@ -3096,6 +3103,11 @@ int RGWRados::delete_obj_impl(void *ctx, rgw_obj& obj)
   r = prepare_update_index(state, bucket, CLS_RGW_OP_DEL, obj, tag);
   if (r < 0)
     return r;
+
+  if (objv_tracker) {
+    objv_tracker->prepare_op_for_write(&op);
+  }
+
   cls_refcount_put(op, tag, true);
   r = io_ctx.operate(oid, &op);
   bool removed = (r >= 0);
@@ -3129,11 +3141,11 @@ int RGWRados::delete_obj_impl(void *ctx, rgw_obj& obj)
   return 0;
 }
 
-int RGWRados::delete_obj(void *ctx, rgw_obj& obj)
+int RGWRados::delete_obj(void *ctx, rgw_obj& obj, RGWObjVersionTracker *objv_tracker)
 {
   int r;
 
-  r = delete_obj_impl(ctx, obj);
+  r = delete_obj_impl(ctx, obj, objv_tracker);
   if (r == -ECANCELED)
     r = 0;
 
@@ -3253,8 +3265,7 @@ int RGWRados::get_obj_state(RGWRadosCtx *rctx, rgw_obj& obj, RGWObjState **state
  * dest: bufferlist to store the result in
  * Returns: 0 on success, -ERR# otherwise.
  */
-int RGWRados::get_attr(void *ctx, rgw_obj& obj, const char *name, bufferlist& dest,
-                       RGWObjVersionTracker *objv_tracker)
+int RGWRados::get_attr(void *ctx, rgw_obj& obj, const char *name, bufferlist& dest)
 {
   rgw_bucket bucket;
   std::string oid, key;
@@ -3289,10 +3300,6 @@ int RGWRados::get_attr(void *ctx, rgw_obj& obj, const char *name, bufferlist& de
 
   ObjectReadOperation op;
 
-  if (objv_tracker) {
-    objv_tracker->prepare_op_for_read(&op);
-  }
-
   int rval;
   op.getxattr(name, &dest, &rval);
   
@@ -3592,7 +3599,7 @@ int RGWRados::prepare_get_obj(void *ctx, rgw_obj& obj,
     }
   }
   if (if_match || if_nomatch) {
-    r = get_attr(rctx, obj, RGW_ATTR_ETAG, etag, NULL);
+    r = get_attr(rctx, obj, RGW_ATTR_ETAG, etag);
     if (r < 0)
       goto done_err;
 
@@ -4602,7 +4609,8 @@ int RGWRados::get_bucket_info(void *ctx, string& bucket_name, RGWBucketInfo& inf
 
   RGWBucketEntryPoint entry_point;
   time_t ep_mtime;
-  int ret = get_bucket_entrypoint_info(ctx, bucket_name, entry_point, NULL, &ep_mtime);
+  RGWObjVersionTracker ot;
+  int ret = get_bucket_entrypoint_info(ctx, bucket_name, entry_point, &ot, &ep_mtime);
   if (ret < 0) {
     info.bucket.name = bucket_name; /* only init this field */
     return ret;
@@ -4610,6 +4618,7 @@ int RGWRados::get_bucket_info(void *ctx, string& bucket_name, RGWBucketInfo& inf
 
   if (entry_point.has_bucket_info) {
     info = entry_point.old_bucket_info;
+    info.ep_objv = ot.read_version;
     ldout(cct, 20) << "rgw_get_bucket_info: old bucket info, bucket=" << info.bucket << " owner " << info.owner << dendl;
     return 0;
   }
@@ -4625,6 +4634,7 @@ int RGWRados::get_bucket_info(void *ctx, string& bucket_name, RGWBucketInfo& inf
   get_bucket_meta_oid(entry_point.bucket, oid);
 
   ret = get_bucket_instance_from_oid(ctx, oid, info, pmtime, pattrs);
+  info.ep_objv = ot.read_version;
   if (ret < 0) {
     info.bucket.name = bucket_name;
     return ret;
@@ -4653,7 +4663,7 @@ int RGWRados::put_bucket_instance_info(RGWBucketInfo& info, bool exclusive,
   return rgw_bucket_instance_store_info(this, key, bl, exclusive, pattrs, &info.objv_tracker, mtime);
 }
 
-int RGWRados::put_linked_bucket_info(RGWBucketInfo& info, bool exclusive, time_t mtime,
+int RGWRados::put_linked_bucket_info(RGWBucketInfo& info, bool exclusive, time_t mtime, obj_version *pep_objv,
                                      map<string, bufferlist> *pattrs, bool create_entry_point)
 {
   bufferlist bl;
@@ -4674,7 +4684,14 @@ int RGWRados::put_linked_bucket_info(RGWBucketInfo& info, bool exclusive, time_t
   entry_point.creation_time = info.creation_time;
   entry_point.linked = true;
   RGWObjVersionTracker ot;
-  ot.generate_new_write_ver(cct);
+  if (pep_objv && !pep_objv->tag.empty()) {
+    ot.write_version = *pep_objv;
+  } else {
+    ot.generate_new_write_ver(cct);
+    if (pep_objv) {
+      *pep_objv = ot.write_version;
+    }
+  }
   ret = put_bucket_entrypoint_info(info.bucket.name, entry_point, exclusive, ot, mtime); 
   if (ret < 0)
     return ret;
diff --git a/src/rgw/rgw_rados.h b/src/rgw/rgw_rados.h
index fb1a1756ba8..6422c182adc 100644
--- a/src/rgw/rgw_rados.h
+++ b/src/rgw/rgw_rados.h
@@ -619,6 +619,7 @@ struct RGWRegion {
   int read_info(const string& region_name);
   int read_default(RGWDefaultRegionInfo& default_region);
   int set_as_default();
+  int equals(const string& other_region);
 
   static string get_pool_name(CephContext *cct);
 
@@ -843,7 +844,7 @@ class RGWRados
     v.push_back(info);
     return clone_objs(ctx, dst_obj, v, attrs, category, pmtime, true, false);
   }
-  int delete_obj_impl(void *ctx, rgw_obj& src_obj);
+  int delete_obj_impl(void *ctx, rgw_obj& src_obj, RGWObjVersionTracker *objv_tracker);
   int complete_atomic_overwrite(RGWRadosCtx *rctx, RGWObjState *state, rgw_obj& obj);
 
   int update_placement_map();
@@ -983,6 +984,7 @@ public:
                             map<std::string,bufferlist>& attrs,
                             RGWBucketInfo& bucket_info,
                             obj_version *pobjv,
+                            obj_version *pep_objv,
                             time_t creation_time,
                             rgw_bucket *master_bucket,
                             bool exclusive = true);
@@ -1142,7 +1144,7 @@ public:
   int bucket_suspended(rgw_bucket& bucket, bool *suspended);
 
   /** Delete an object.*/
-  virtual int delete_obj(void *ctx, rgw_obj& src_obj);
+  virtual int delete_obj(void *ctx, rgw_obj& src_obj, RGWObjVersionTracker *objv_tracker = NULL);
 
   /** Remove an object from the bucket index */
   int delete_obj_index(rgw_obj& obj);
@@ -1155,8 +1157,7 @@ public:
    * dest: bufferlist to store the result in
    * Returns: 0 on success, -ERR# otherwise.
    */
-  virtual int get_attr(void *ctx, rgw_obj& obj, const char *name,
-                       bufferlist& dest, RGWObjVersionTracker *objv_tracker);
+  virtual int get_attr(void *ctx, rgw_obj& obj, const char *name, bufferlist& dest);
 
   /**
    * Set an attr on an object.
@@ -1294,7 +1295,7 @@ public:
 
   virtual int get_bucket_info(void *ctx, string& bucket_name, RGWBucketInfo& info,
                               time_t *pmtime, map<string, bufferlist> *pattrs = NULL);
-  virtual int put_linked_bucket_info(RGWBucketInfo& info, bool exclusive, time_t mtime,
+  virtual int put_linked_bucket_info(RGWBucketInfo& info, bool exclusive, time_t mtime, obj_version *pep_objv,
                                      map<string, bufferlist> *pattrs, bool create_entry_point);
 
   int cls_rgw_init_index(librados::IoCtx& io_ctx, librados::ObjectWriteOperation& op, string& oid);
diff --git a/src/rgw/rgw_rest_client.cc b/src/rgw/rgw_rest_client.cc
index 3e712e7e023..2075e535525 100644
--- a/src/rgw/rgw_rest_client.cc
+++ b/src/rgw/rgw_rest_client.cc
@@ -224,6 +224,11 @@ int RGWRESTSimpleRequest::forward_request(RGWAccessKey& key, req_info& info, siz
     headers.push_back(make_pair<string, string>(iter->first, iter->second));
   }
 
+  map<string, string>& meta_map = new_info.x_meta_map;
+  for (iter = meta_map.begin(); iter != meta_map.end(); ++iter) {
+    headers.push_back(make_pair<string, string>(iter->first, iter->second));
+  }
+
   string params_str;
   map<string, string>& args = new_info.args.get_params();
   get_params_str(args, params_str);
diff --git a/src/rgw/rgw_rest_conn.cc b/src/rgw/rgw_rest_conn.cc
index 5caf3ce0bcd..35a8ac258e6 100644
--- a/src/rgw/rgw_rest_conn.cc
+++ b/src/rgw/rgw_rest_conn.cc
@@ -27,7 +27,7 @@ int RGWRESTConn::get_url(string& endpoint)
   return 0;
 }
 
-int RGWRESTConn::forward(const string& uid, req_info& info, size_t max_response, bufferlist *inbl, bufferlist *outbl)
+int RGWRESTConn::forward(const string& uid, req_info& info, obj_version *objv, size_t max_response, bufferlist *inbl, bufferlist *outbl)
 {
   string url;
   int ret = get_url(url);
@@ -36,6 +36,12 @@ int RGWRESTConn::forward(const string& uid, req_info& info, size_t max_response,
   list<pair<string, string> > params;
   params.push_back(make_pair<string, string>(RGW_SYS_PARAM_PREFIX "uid", uid));
   params.push_back(make_pair<string, string>(RGW_SYS_PARAM_PREFIX "region", region));
+  if (objv) {
+    params.push_back(make_pair<string, string>(RGW_SYS_PARAM_PREFIX "tag", objv->tag));
+    char buf[16];
+    snprintf(buf, sizeof(buf), "%lld", (long long)objv->ver);
+    params.push_back(make_pair<string, string>(RGW_SYS_PARAM_PREFIX "ver", buf));
+  }
   RGWRESTSimpleRequest req(cct, url, NULL, &params);
   return req.forward_request(key, info, max_response, inbl, outbl);
 }
diff --git a/src/rgw/rgw_rest_conn.h b/src/rgw/rgw_rest_conn.h
index 6fe572d2cf7..4a0b6087d26 100644
--- a/src/rgw/rgw_rest_conn.h
+++ b/src/rgw/rgw_rest_conn.h
@@ -20,7 +20,7 @@ public:
   int get_url(string& endpoint);
 
   /* sync request */
-  int forward(const string& uid, req_info& info, size_t max_response, bufferlist *inbl, bufferlist *outbl);
+  int forward(const string& uid, req_info& info, obj_version *objv, size_t max_response, bufferlist *inbl, bufferlist *outbl);
 
   /* async request */
   int put_obj_init(const string& uid, rgw_obj& obj, uint64_t obj_size,
diff --git a/src/rgw/rgw_rest_replica_log.cc b/src/rgw/rgw_rest_replica_log.cc
index 863a979a22e..600a8edb78c 100644
--- a/src/rgw/rgw_rest_replica_log.cc
+++ b/src/rgw/rgw_rest_replica_log.cc
@@ -27,13 +27,13 @@
 #define REPLICA_INPUT_MAX_LEN (512*1024)
 
 static int parse_to_utime(string& in, utime_t& out) {
-  struct tm tm;
-  
-  if (!parse_iso8601(in.c_str(), &tm)) 
-    return -EINVAL;
+  uint64_t sec = 0;
+  uint64_t nsec = 0;
+  int ret = utime_t::parse_date(in.c_str(), &sec, &nsec);
+  if (ret < 0)
+    return ret;
 
-  time_t tt = mktime(&tm);
-  out = utime_t(tt, 0);
+  out = utime_t(sec, nsec);
   return 0;
 }
 
diff --git a/src/rgw/rgw_rest_s3.cc b/src/rgw/rgw_rest_s3.cc
index 6e482e8a251..9e8ec3f88a5 100644
--- a/src/rgw/rgw_rest_s3.cc
+++ b/src/rgw/rgw_rest_s3.cc
@@ -449,6 +449,7 @@ void RGWCreateBucket_ObjStore_S3::send_response()
     JSONFormatter f; /* use json formatter for system requests output */
 
     f.open_object_section("info");
+    encode_json("entry_point_object_ver", ep_objv, &f);
     encode_json("object_ver", info.objv_tracker.read_version, &f);
     encode_json("bucket_info", info, &f);
     f.close_section();