rgw: tie metadata put to bucket link/unlink

and lot's of constifying

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

12 files changed, 113 insertions, 122 deletions

diff --git a/src/rgw/rgw_bucket.cc b/src/rgw/rgw_bucket.cc
index ac638f80dac..dd8d06694ee 100644
--- a/src/rgw/rgw_bucket.cc
+++ b/src/rgw/rgw_bucket.cc
@@ -73,7 +73,7 @@ int rgw_read_user_buckets(RGWRados *store, string user_id, RGWUserBuckets& bucke
   return 0;
 }
 
-int rgw_add_bucket(RGWRados *store, string user_id, rgw_bucket& bucket, time_t creation_time)
+int rgw_link_bucket(RGWRados *store, string user_id, rgw_bucket& bucket, time_t creation_time)
 {
   int ret;
   string& bucket_name = bucket.name;
@@ -81,6 +81,10 @@ int rgw_add_bucket(RGWRados *store, string user_id, rgw_bucket& bucket, time_t c
   bufferlist bl;
 
   RGWBucketEnt new_bucket;
+
+  RGWBucketEntryPoint ep;
+  RGWObjVersionTracker ot;
+
   new_bucket.bucket = bucket;
   new_bucket.size = 0;
   if (!creation_time)
@@ -89,6 +93,14 @@ int rgw_add_bucket(RGWRados *store, string user_id, rgw_bucket& bucket, time_t c
     new_bucket.creation_time = creation_time;
   ::encode(new_bucket, bl);
 
+  ret = store->get_bucket_entrypoint_info(NULL, bucket_name, ep, &ot, NULL);
+  if (ret < 0 && ret != -ENOENT) {
+    ldout(store->ctx(), 0) << "ERROR: store->get_bucket_entrypoint_info() returned " << ret << dendl;
+  } else if (ret >= 0 && ep.linked && ep.owner != user_id) {
+    ldout(store->ctx(), 0) << "can't link bucket, already linked to a different user: " << ep.owner << dendl;
+    return -EINVAL;
+  }
+
   string buckets_obj_id;
   rgw_get_buckets_obj(user_id, buckets_obj_id);
 
@@ -97,12 +109,25 @@ int rgw_add_bucket(RGWRados *store, string user_id, rgw_bucket& bucket, time_t c
   if (ret < 0) {
     ldout(store->ctx(), 0) << "ERROR: error adding bucket to directory: "
         << cpp_strerror(-ret)<< dendl;
+    goto done_err;
   }
 
+  ep.linked = true;
+  ep.owner = user_id;
+  ret = store->put_bucket_entrypoint_info(bucket_name, ep, false, ot, 0);
+  if (ret < 0)
+    goto done_err;
+
+  return 0;
+done_err:
+  int r = rgw_unlink_bucket(store, user_id, bucket.name);
+  if (r < 0) {
+    ldout(store->ctx(), 0) << "ERROR: failed unlinking bucket on error cleanup: " << cpp_strerror(-r) << dendl;
+  }
   return ret;
 }
 
-int rgw_remove_user_bucket_info(RGWRados *store, string user_id, const string& bucket_name)
+int rgw_unlink_bucket(RGWRados *store, string user_id, const string& bucket_name)
 {
   int ret;
 
@@ -118,10 +143,29 @@ int rgw_remove_user_bucket_info(RGWRados *store, string user_id, const string& b
         << cpp_strerror(-ret)<< dendl;
   }
 
+  RGWBucketEntryPoint ep;
+  RGWObjVersionTracker ot;
+  ret = store->get_bucket_entrypoint_info(NULL, bucket_name, ep, &ot, NULL);
+  if (ret < 0)
+    return ret;
+
+  if (!ep.linked)
+    return 0;
+
+  if (ep.owner != user_id) {
+    ldout(store->ctx(), 0) << "bucket entry point user mismatch, can't unlink bucket: " << ep.owner << " != " << user_id << dendl;
+    return -EINVAL;
+  }
+
+  ep.linked = false;
+  ret = store->put_bucket_entrypoint_info(bucket_name, ep, false, ot, 0);
+  if (ret < 0)
+    return ret;
+
   return ret;
 }
 
-int rgw_bucket_store_info(RGWRados *store, string& bucket_name, bufferlist& bl, bool exclusive,
+int rgw_bucket_store_info(RGWRados *store, const string& bucket_name, bufferlist& bl, bool exclusive,
                           map<string, bufferlist> *pattrs, RGWObjVersionTracker *objv_tracker,
                           time_t mtime) {
   return store->meta_mgr->put_entry(bucket_meta_handler, bucket_name, bl, exclusive, objv_tracker, mtime, pattrs);
@@ -137,57 +181,6 @@ int rgw_bucket_instance_remove_entry(RGWRados *store, string& entry, RGWObjVersi
   return store->meta_mgr->remove_entry(bucket_instance_meta_handler, entry, objv_tracker);
 }
 
-#warning removed RGWBucket::create_bucket(), clean this up when ready
-#if 0
-int RGWBucket::create_bucket(string bucket_str, string& user_id, string& region_name, string& display_name)
-{
-  RGWAccessControlPolicy policy, old_policy;
-  map<string, bufferlist> attrs;
-  bufferlist aclbl;
-  string no_oid;
-  rgw_obj obj;
-  RGWBucketInfo bucket_info;
-
-  int ret;
-
-  // defaule policy (private)
-  policy.create_default(user_id, display_name);
-  policy.encode(aclbl);
-
-  RGWObjVersionTracker objv_tracker;
-  time_t mtime;
-
-  ret = store->get_bucket_info(NULL, bucket_str, bucket_info, &objv_tracker, &mtime);
-  if (ret < 0)
-    return ret;
-
-  rgw_bucket& bucket = bucket_info.bucket;
-
-  RGWBucketInfo new_info;
-  string placement_rule;
-
-  ret = store->create_bucket(user_info, bucket, region_name, placement_rule, attrs, objv_tracker,
-                             NULL, bucket_info.creation_time, NULL, &new_info);
-  if (ret && ret != -EEXIST)
-    goto done;
-
-  obj.init(bucket, no_oid);
-
-  ret = store->set_attr(NULL, obj, RGW_ATTR_ACL, aclbl, &objv_tracker);
-  if (ret < 0) {
-    lderr(store->ctx()) << "ERROR: failed to set acl on bucket" << dendl;
-    goto done;
-  }
-
-  ret = rgw_add_bucket(store, user_id, bucket, new_info.creation_time);
-
-  if (ret == -EEXIST)
-    ret = 0;
-done:
-  return ret;
-}
-#endif
-
 int rgw_bucket_set_attrs(RGWRados *store, rgw_bucket& bucket,
                          map<string, bufferlist>& attrs,
                          map<string, bufferlist>* rmattrs,
@@ -259,7 +252,7 @@ void check_bad_user_bucket_mapping(RGWRados *store, const string& user_id, bool
         cout << "bucket info mismatch: expected " << actual_bucket << " got " << bucket << std::endl;
         if (fix) {
           cout << "fixing" << std::endl;
-          r = rgw_add_bucket(store, user_id, actual_bucket, bucket_info.creation_time);
+          r = rgw_link_bucket(store, user_id, actual_bucket, bucket_info.creation_time);
           if (r < 0) {
             cerr << "failed to fix bucket: " << cpp_strerror(-r) << std::endl;
           }
@@ -350,7 +343,7 @@ int rgw_remove_bucket(RGWRados *store, rgw_bucket& bucket, bool delete_children)
     return ret;
   }
 
-  ret = rgw_remove_user_bucket_info(store, info.owner, bucket.name);
+  ret = rgw_unlink_bucket(store, info.owner, bucket.name);
   if (ret < 0) {
     lderr(store->ctx()) << "ERROR: unable to remove user bucket information" << dendl;
   }
@@ -437,7 +430,7 @@ int RGWBucket::link(RGWBucketAdminOpState& op_state, std::string *err_msg)
       return -EIO;
     }
 
-    r = rgw_remove_user_bucket_info(store, owner.get_id(), bucket.name);
+    r = rgw_unlink_bucket(store, owner.get_id(), bucket.name);
     if (r < 0) {
       set_err_msg(err_msg, "could not unlink policy from user " + owner.get_id());
       return r;
@@ -464,22 +457,10 @@ int RGWBucket::link(RGWBucketAdminOpState& op_state, std::string *err_msg)
     if (r < 0)
       return r;
 
-    r = rgw_add_bucket(store, user_info.user_id, bucket, 0);
+    r = rgw_link_bucket(store, user_info.user_id, bucket, 0);
     if (r < 0)
       return r;
   }
-#warning not creating bucket on bucket link, clean this up later
-#if 0
-  else {
-    // the bucket seems not to exist, so we should probably create it...
-    r = create_bucket(bucket_name.c_str(), uid_str, store->region.name, display_name);
-    if (r < 0) {
-      set_err_msg(err_msg, "error linking bucket to user r=" + cpp_strerror(-r));
-    }
-
-    return r;
-  }
-#endif
 
   return 0;
 }
@@ -493,7 +474,7 @@ int RGWBucket::unlink(RGWBucketAdminOpState& op_state, std::string *err_msg)
     return -EINVAL;
   }
 
-  int r = rgw_remove_user_bucket_info(store, user_info.user_id, bucket.name);
+  int r = rgw_unlink_bucket(store, user_info.user_id, bucket.name);
   if (r < 0) {
     set_err_msg(err_msg, "error unlinking bucket" + cpp_strerror(-r));
   }
@@ -1378,7 +1359,11 @@ public:
       return ret;
 
     /* link bucket */
-    ret = rgw_add_bucket(store, be.owner, be.bucket, be.creation_time);
+    if (be.linked) {
+      ret = rgw_link_bucket(store, be.owner, be.bucket, be.creation_time);
+    } else {
+      ret = rgw_unlink_bucket(store, be.owner, be.bucket.name);
+    }
 
     return 0;
   }
@@ -1395,7 +1380,7 @@ public:
     if (ret < 0)
       return ret;
 
-    ret = rgw_remove_user_bucket_info(store, be.owner, entry);
+    ret = rgw_unlink_bucket(store, be.owner, entry);
     if (ret < 0) {
       lderr(store->ctx()) << "could not unlink bucket=" << entry << " owner=" << be.owner << dendl;
     }
@@ -1408,7 +1393,7 @@ public:
     return 0;
   }
 
-  void get_pool_and_oid(RGWRados *store, string& key, rgw_bucket& bucket, string& oid) {
+  void get_pool_and_oid(RGWRados *store, const string& key, rgw_bucket& bucket, string& oid) {
     oid = key;
     bucket = store->zone.domain_root;
   }
@@ -1550,7 +1535,7 @@ public:
     return rgw_bucket_instance_remove_entry(store, entry, &info.objv_tracker);
   }
 
-  void get_pool_and_oid(RGWRados *store, string& key, rgw_bucket& bucket, string& oid) {
+  void get_pool_and_oid(RGWRados *store, const string& key, rgw_bucket& bucket, string& oid) {
     oid = RGW_BUCKET_INSTANCE_MD_PREFIX + key;
     bucket = store->zone.domain_root;
   }
diff --git a/src/rgw/rgw_bucket.h b/src/rgw/rgw_bucket.h
index 2e6ed69d7ca..92076cd05f6 100644
--- a/src/rgw/rgw_bucket.h
+++ b/src/rgw/rgw_bucket.h
@@ -22,7 +22,7 @@ using namespace std;
 // define as static when RGWBucket implementation compete
 extern void rgw_get_buckets_obj(string& user_id, string& buckets_obj_id);
 
-extern int rgw_bucket_store_info(RGWRados *store, string& bucket_name, bufferlist& bl, bool exclusive,
+extern int rgw_bucket_store_info(RGWRados *store, const string& bucket_name, bufferlist& bl, bool exclusive,
                                  map<string, bufferlist> *pattrs, RGWObjVersionTracker *objv_tracker,
                                  time_t mtime);
 extern int rgw_bucket_instance_store_info(RGWRados *store, string& oid, bufferlist& bl, bool exclusive,
@@ -99,8 +99,8 @@ extern void rgw_bucket_init(RGWMetadataManager *mm);
 extern int rgw_read_user_buckets(RGWRados *store, string user_id, RGWUserBuckets& buckets,
                                  const string& marker, uint64_t max, bool need_stats);
 
-extern int rgw_add_bucket(RGWRados *store, string user_id, rgw_bucket& bucket, time_t creation_time);
-extern int rgw_remove_user_bucket_info(RGWRados *store, string user_id, const string& bucket_name);
+extern int rgw_link_bucket(RGWRados *store, string user_id, rgw_bucket& bucket, time_t creation_time);
+extern int rgw_unlink_bucket(RGWRados *store, string user_id, const string& bucket_name);
 
 extern int rgw_remove_object(RGWRados *store, rgw_bucket& bucket, std::string& object);
 extern int rgw_remove_bucket(RGWRados *store, rgw_bucket& bucket, bool delete_children);
diff --git a/src/rgw/rgw_common.h b/src/rgw/rgw_common.h
index d5e5ae61e4c..e9c12a3d35c 100644
--- a/src/rgw/rgw_common.h
+++ b/src/rgw/rgw_common.h
@@ -685,16 +685,18 @@ struct RGWBucketEntryPoint
   rgw_bucket bucket;
   string owner;
   time_t creation_time;
+  bool linked;
 
   bool has_bucket_info;
   RGWBucketInfo old_bucket_info;
 
-  RGWBucketEntryPoint() : creation_time(0), has_bucket_info(false) {}
+  RGWBucketEntryPoint() : creation_time(0), linked(false), has_bucket_info(false) {}
 
   void encode(bufferlist& bl) const {
     ENCODE_START(8, 8, bl);
     ::encode(bucket, bl);
     ::encode(owner, bl);
+    ::encode(linked, bl);
     uint64_t ctime = (uint64_t)creation_time;
     ::encode(ctime, bl);
     ENCODE_FINISH(bl);
@@ -711,6 +713,7 @@ struct RGWBucketEntryPoint
     has_bucket_info = false;
     ::decode(bucket, bl);
     ::decode(owner, bl);
+    ::decode(linked, bl);
     uint64_t ctime;
     ::decode(ctime, bl);
     creation_time = (uint64_t)ctime;
@@ -910,32 +913,32 @@ public:
     std::string _o(o);
     init(b, _o);
   }
-  rgw_obj(rgw_bucket& b, std::string& o) {
+  rgw_obj(rgw_bucket& b, const std::string& o) {
     init(b, o);
   }
-  rgw_obj(rgw_bucket& b, std::string& o, std::string& k) {
+  rgw_obj(rgw_bucket& b, const std::string& o, const std::string& k) {
     init(b, o, k);
   }
-  rgw_obj(rgw_bucket& b, std::string& o, std::string& k, std::string& n) {
+  rgw_obj(rgw_bucket& b, const std::string& o, const std::string& k, const std::string& n) {
     init(b, o, k, n);
   }
-  void init(rgw_bucket& b, std::string& o, std::string& k, std::string& n) {
+  void init(rgw_bucket& b, const std::string& o, const std::string& k, const std::string& n) {
     bucket = b;
     set_ns(n);
     set_obj(o);
     set_key(k);
   }
-  void init(rgw_bucket& b, std::string& o, std::string& k) {
+  void init(rgw_bucket& b, const std::string& o, const std::string& k) {
     bucket = b;
     set_obj(o);
     set_key(k);
   }
-  void init(rgw_bucket& b, std::string& o) {
+  void init(rgw_bucket& b, const std::string& o) {
     bucket = b;
     set_obj(o);
     orig_key = key = o;
   }
-  void init_ns(rgw_bucket& b, std::string& o, std::string& n) {
+  void init_ns(rgw_bucket& b, const std::string& o, const std::string& n) {
     bucket = b;
     set_ns(n);
     set_obj(o);
@@ -947,7 +950,7 @@ public:
     string ns_str(n);
     return set_ns(ns_str);
   }
-  int set_ns(string& n) {
+  int set_ns(const string& n) {
     if (n[0] == '_')
       return -EINVAL;
     ns = n;
@@ -955,7 +958,7 @@ public:
     return 0;
   }
 
-  void set_key(string& k) {
+  void set_key(const string& k) {
     orig_key = k;
     key = k;
   }
@@ -965,7 +968,7 @@ public:
     key.clear();
   }
 
-  void set_obj(string& o) {
+  void set_obj(const string& o) {
     orig_obj = o;
     if (ns.empty()) {
       if (o.empty())
diff --git a/src/rgw/rgw_json_enc.cc b/src/rgw/rgw_json_enc.cc
index 1be85d43607..78dd6f10c87 100644
--- a/src/rgw/rgw_json_enc.cc
+++ b/src/rgw/rgw_json_enc.cc
@@ -401,7 +401,7 @@ void RGWUserInfo::decode_json(JSONObj *obj)
   JSONDecoder::decode_json("user_id", user_id, obj, true);
   JSONDecoder::decode_json("display_name", display_name, obj);
   JSONDecoder::decode_json("email", user_email, obj);
-  bool susp;
+  bool susp = false;
   JSONDecoder::decode_json("suspended", susp, obj);
   suspended = (__u8)susp;
   JSONDecoder::decode_json("max_buckets", max_buckets, obj);
@@ -412,7 +412,7 @@ void RGWUserInfo::decode_json(JSONObj *obj)
   JSONDecoder::decode_json("subusers", subusers, decode_subusers, obj);
 
   JSONDecoder::decode_json("caps", caps, obj);
-  bool sys;
+  bool sys = false;
   JSONDecoder::decode_json("system", sys, obj);
   system = (__u8)sys;
   JSONDecoder::decode_json("default_placement", default_placement, obj);
@@ -441,6 +441,7 @@ void RGWBucketEntryPoint::dump(Formatter *f) const
   encode_json("bucket", bucket, f);
   encode_json("owner", owner, f);
   encode_json("creation_time", creation_time, f);
+  encode_json("linked", linked, f);
   encode_json("has_bucket_info", has_bucket_info, f);
   if (has_bucket_info) {
     encode_json("old_bucket_info", old_bucket_info, f);
@@ -451,6 +452,7 @@ void RGWBucketEntryPoint::decode_json(JSONObj *obj) {
   JSONDecoder::decode_json("bucket", bucket, obj);
   JSONDecoder::decode_json("owner", owner, obj);
   JSONDecoder::decode_json("creation_time", creation_time, obj);
+  JSONDecoder::decode_json("linked", linked, obj);
   JSONDecoder::decode_json("has_bucket_info", has_bucket_info, obj);
   if (has_bucket_info) {
     JSONDecoder::decode_json("old_bucket_info", old_bucket_info, obj);
diff --git a/src/rgw/rgw_metadata.cc b/src/rgw/rgw_metadata.cc
index 99d06fda5c4..d27ba14d009 100644
--- a/src/rgw/rgw_metadata.cc
+++ b/src/rgw/rgw_metadata.cc
@@ -74,7 +74,7 @@ struct RGWMetadataLogData {
 WRITE_CLASS_ENCODER(RGWMetadataLogData);
 
 
-int RGWMetadataLog::add_entry(RGWRados *store, string& section, string& key, bufferlist& bl) {
+int RGWMetadataLog::add_entry(RGWRados *store, const string& section, const string& key, bufferlist& bl) {
   string oid;
 
   store->shard_name(prefix, cct->_conf->rgw_md_log_max_shards, section, key, oid);
@@ -172,7 +172,7 @@ public:
   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 void get_pool_and_oid(RGWRados *store, string& key, rgw_bucket& bucket, string& oid) {}
+  virtual void get_pool_and_oid(RGWRados *store, const string& key, rgw_bucket& bucket, string& oid) {}
 
   virtual int remove(RGWRados *store, string& entry, RGWObjVersionTracker& objv_tracker) { return -ENOTSUP; }
 
@@ -471,7 +471,7 @@ void RGWMetadataManager::get_sections(list<string>& sections)
   }
 }
 
-int RGWMetadataManager::pre_modify(RGWMetadataHandler *handler, string& section, string& key,
+int RGWMetadataManager::pre_modify(RGWMetadataHandler *handler, string& section, const string& key,
                                    RGWMetadataLogData& log_data, RGWObjVersionTracker *objv_tracker,
                                    RGWMDLogStatus op_type)
 {
@@ -501,7 +501,7 @@ int RGWMetadataManager::pre_modify(RGWMetadataHandler *handler, string& section,
   return 0;
 }
 
-int RGWMetadataManager::post_modify(string& section, string& key, RGWMetadataLogData& log_data,
+int RGWMetadataManager::post_modify(const string& section, const string& key, RGWMetadataLogData& log_data,
                                     RGWObjVersionTracker *objv_tracker, int ret)
 {
   if (ret >= 0)
@@ -522,7 +522,7 @@ int RGWMetadataManager::post_modify(string& section, string& key, RGWMetadataLog
   return 0;
 }
 
-int RGWMetadataManager::put_entry(RGWMetadataHandler *handler, string& key, bufferlist& bl, bool exclusive,
+int RGWMetadataManager::put_entry(RGWMetadataHandler *handler, const string& key, bufferlist& bl, bool exclusive,
                                   RGWObjVersionTracker *objv_tracker, time_t mtime, map<string, bufferlist> *pattrs)
 {
   string section;
diff --git a/src/rgw/rgw_metadata.h b/src/rgw/rgw_metadata.h
index 6dd21d10a21..7c991e852ba 100644
--- a/src/rgw/rgw_metadata.h
+++ b/src/rgw/rgw_metadata.h
@@ -45,7 +45,7 @@ class RGWMetadataHandler {
   friend class RGWMetadataManager;
 
 protected:
-  virtual void get_pool_and_oid(RGWRados *store, string& key, rgw_bucket& bucket, string& oid) = 0;
+  virtual void get_pool_and_oid(RGWRados *store, const string& key, rgw_bucket& bucket, string& oid) = 0;
 public:
   virtual ~RGWMetadataHandler() {}
   virtual string get_type() = 0;
@@ -77,7 +77,7 @@ public:
     prefix = META_LOG_OBJ_PREFIX;
   }
 
-  int add_entry(RGWRados *store, string& section, string& key, bufferlist& bl);
+  int add_entry(RGWRados *store, const string& section, const string& key, bufferlist& bl);
 
   struct LogListCtx {
     int cur_shard;
@@ -114,10 +114,10 @@ class RGWMetadataManager {
   void parse_metadata_key(const string& metadata_key, string& type, string& entry);
 
   int find_handler(const string& metadata_key, RGWMetadataHandler **handler, string& entry);
-  int pre_modify(RGWMetadataHandler *handler, string& section, string& key,
+  int pre_modify(RGWMetadataHandler *handler, string& section, const string& key,
                  RGWMetadataLogData& log_data, RGWObjVersionTracker *objv_tracker,
                  RGWMDLogStatus op_type);
-  int post_modify(string& section, string& key, RGWMetadataLogData& log_data,
+  int post_modify(const string& section, const string& key, RGWMetadataLogData& log_data,
                  RGWObjVersionTracker *objv_tracker, int ret);
 
 public:
@@ -128,7 +128,7 @@ public:
 
   RGWMetadataHandler *get_handler(const char *type);
 
-  int put_entry(RGWMetadataHandler *handler, string& key, bufferlist& bl, bool exclusive,
+  int put_entry(RGWMetadataHandler *handler, const string& key, bufferlist& bl, bool exclusive,
                 RGWObjVersionTracker *objv_tracker, time_t mtime, map<string, bufferlist> *pattrs = NULL);
   int remove_entry(RGWMetadataHandler *handler, string& key, RGWObjVersionTracker *objv_tracker);
   int set_attr(RGWMetadataHandler *handler, string& key, rgw_obj& obj, string& attr, bufferlist& bl,
diff --git a/src/rgw/rgw_op.cc b/src/rgw/rgw_op.cc
index 43d2206eace..fce62494566 100644
--- a/src/rgw/rgw_op.cc
+++ b/src/rgw/rgw_op.cc
@@ -1031,9 +1031,9 @@ void RGWCreateBucket::execute()
     s->bucket = info.bucket;
   }
 
-  ret = rgw_add_bucket(store, s->user.user_id, s->bucket, info.creation_time);
+  ret = rgw_link_bucket(store, s->user.user_id, s->bucket, info.creation_time);
   if (ret && !existed && ret != -EEXIST)   /* if it exists (or previously existed), don't remove it! */
-    rgw_remove_user_bucket_info(store, s->user.user_id, s->bucket.name);
+    rgw_unlink_bucket(store, s->user.user_id, s->bucket.name);
 
   if (ret == -EEXIST)
     ret = -ERR_BUCKET_EXISTS;
@@ -1067,7 +1067,7 @@ void RGWDeleteBucket::execute()
   ret = store->delete_bucket(s->bucket, objv_tracker);
 
   if (ret == 0) {
-    ret = rgw_remove_user_bucket_info(store, s->user.user_id, s->bucket.name);
+    ret = rgw_unlink_bucket(store, s->user.user_id, s->bucket.name);
     if (ret < 0) {
       ldout(s->cct, 0) << "WARNING: failed to remove bucket: ret=" << ret << dendl;
     }
diff --git a/src/rgw/rgw_rados.cc b/src/rgw/rgw_rados.cc
index 6b952573d3c..c13a7741b21 100644
--- a/src/rgw/rgw_rados.cc
+++ b/src/rgw/rgw_rados.cc
@@ -1426,7 +1426,7 @@ next:
   return 0;
 }
 
-void RGWRados::shard_name(const string& prefix, unsigned max_shards, string& key, string& name)
+void RGWRados::shard_name(const string& prefix, unsigned max_shards, const string& key, string& name)
 {
   uint32_t val = ceph_str_hash_linux(key.c_str(), key.size());
   char buf[16];
@@ -1434,7 +1434,7 @@ void RGWRados::shard_name(const string& prefix, unsigned max_shards, string& key
   name = prefix + buf;
 }
 
-void RGWRados::shard_name(const string& prefix, unsigned max_shards, string& section, string& key, string& name)
+void RGWRados::shard_name(const string& prefix, unsigned max_shards, const string& section, const string& key, string& name)
 {
   uint32_t val = ceph_str_hash_linux(key.c_str(), key.size());
   val ^= ceph_str_hash_linux(section.c_str(), section.size());
@@ -1448,7 +1448,7 @@ void RGWRados::time_log_prepare_entry(cls_log_entry& entry, const utime_t& ut, s
   cls_log_add_prepare_entry(entry, ut, section, key, bl);
 }
 
-int RGWRados::time_log_add(const string& oid, const utime_t& ut, string& section, string& key, bufferlist& bl)
+int RGWRados::time_log_add(const string& oid, const utime_t& ut, const string& section, const string& key, bufferlist& bl)
 {
   librados::IoCtx io_ctx;
 
@@ -1805,7 +1805,7 @@ int RGWRados::create_bucket(RGWUserInfo& owner, rgw_bucket& bucket,
       time(&info.creation_time);
     else
       info.creation_time = creation_time;
-    ret = put_bucket_info(info, exclusive, 0, &attrs, true);
+    ret = put_linked_bucket_info(info, exclusive, 0, &attrs, true);
     if (ret == -EEXIST) {
       /* remove bucket meta instance */
       string entry;
@@ -4533,7 +4533,7 @@ int RGWRados::get_bucket_instance_from_oid(void *ctx, string& oid, RGWBucketInfo
   return 0;
 }
 
-int RGWRados::get_bucket_entrypoint_info(void *ctx, string& bucket_name,
+int RGWRados::get_bucket_entrypoint_info(void *ctx, const string& bucket_name,
                                          RGWBucketEntryPoint& entry_point,
                                          RGWObjVersionTracker *objv_tracker,
                                          time_t *pmtime)
@@ -4592,7 +4592,7 @@ int RGWRados::get_bucket_info(void *ctx, string& bucket_name, RGWBucketInfo& inf
   return 0;
 }
 
-int RGWRados::put_bucket_entrypoint_info(string& bucket_name, RGWBucketEntryPoint& entry_point,
+int RGWRados::put_bucket_entrypoint_info(const string& bucket_name, RGWBucketEntryPoint& entry_point,
                                          bool exclusive, RGWObjVersionTracker& objv_tracker, time_t mtime)
 {
   bufferlist epbl;
@@ -4613,8 +4613,8 @@ 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_bucket_info(RGWBucketInfo& info, bool exclusive, time_t mtime,
-                              map<string, bufferlist> *pattrs, bool create_entry_point)
+int RGWRados::put_linked_bucket_info(RGWBucketInfo& info, bool exclusive, time_t mtime,
+                                     map<string, bufferlist> *pattrs, bool create_entry_point)
 {
   bufferlist bl;
 
@@ -4632,6 +4632,7 @@ int RGWRados::put_bucket_info(RGWBucketInfo& info, bool exclusive, time_t mtime,
   entry_point.bucket = info.bucket;
   entry_point.owner = info.owner;
   entry_point.creation_time = info.creation_time;
+  entry_point.linked = true;
   RGWObjVersionTracker ot;
   ot.generate_new_write_ver(cct);
   ret = put_bucket_entrypoint_info(info.bucket.name, entry_point, exclusive, ot, mtime); 
diff --git a/src/rgw/rgw_rados.h b/src/rgw/rgw_rados.h
index d777a543b7b..72fa4cbd6d1 100644
--- a/src/rgw/rgw_rados.h
+++ b/src/rgw/rgw_rados.h
@@ -1272,17 +1272,17 @@ public:
   void get_bucket_instance_entry(rgw_bucket& bucket, string& entry);
   void get_bucket_meta_oid(rgw_bucket& bucket, string& oid);
 
-  int put_bucket_entrypoint_info(string& bucket_name, RGWBucketEntryPoint& entry_point, bool exclusive, RGWObjVersionTracker& objv_tracker, time_t mtime);
+  int put_bucket_entrypoint_info(const string& bucket_name, RGWBucketEntryPoint& entry_point, bool exclusive, RGWObjVersionTracker& objv_tracker, time_t mtime);
   int put_bucket_instance_info(RGWBucketInfo& info, bool exclusive, time_t mtime, map<string, bufferlist> *pattrs);
-  int get_bucket_entrypoint_info(void *ctx, string& bucket_name, RGWBucketEntryPoint& entry_point, RGWObjVersionTracker *objv_tracker, time_t *pmtime);
+  int get_bucket_entrypoint_info(void *ctx, const string& bucket_name, RGWBucketEntryPoint& entry_point, RGWObjVersionTracker *objv_tracker, time_t *pmtime);
   int get_bucket_instance_info(void *ctx, const string& meta_key, RGWBucketInfo& info, time_t *pmtime, map<string, bufferlist> *pattrs);
   int get_bucket_instance_info(void *ctx, rgw_bucket& bucket, RGWBucketInfo& info, time_t *pmtime, map<string, bufferlist> *pattrs);
   int get_bucket_instance_from_oid(void *ctx, string& oid, RGWBucketInfo& info, time_t *pmtime, map<string, bufferlist> *pattrs);
 
   virtual int get_bucket_info(void *ctx, string& bucket_name, RGWBucketInfo& info,
                               time_t *pmtime, map<string, bufferlist> *pattrs = NULL);
-  virtual int put_bucket_info(RGWBucketInfo& info, bool exclusive, time_t mtime,
-                              map<string, bufferlist> *pattrs, bool create_entry_point);
+  virtual int put_linked_bucket_info(RGWBucketInfo& info, bool exclusive, time_t mtime,
+                                     map<string, bufferlist> *pattrs, bool create_entry_point);
 
   int cls_rgw_init_index(librados::IoCtx& io_ctx, librados::ObjectWriteOperation& op, string& oid);
   int cls_obj_prepare_op(rgw_bucket& bucket, RGWModifyOp op, string& tag,
@@ -1322,11 +1322,11 @@ public:
                              string& read_iter, map<rgw_user_bucket, rgw_usage_log_entry>& usage, bool *is_truncated);
   int cls_obj_usage_log_trim(string& oid, string& user, uint64_t start_epoch, uint64_t end_epoch);
 
-  void shard_name(const string& prefix, unsigned max_shards, string& key, string& name);
-  void shard_name(const string& prefix, unsigned max_shards, string& section, string& key, string& name);
+  void shard_name(const string& prefix, unsigned max_shards, const string& key, string& name);
+  void shard_name(const string& prefix, unsigned max_shards, const string& section, const string& key, string& name);
   void time_log_prepare_entry(cls_log_entry& entry, const utime_t& ut, string& section, string& key, bufferlist& bl);
   int time_log_add(const string& oid, list<cls_log_entry>& entries);
-  int time_log_add(const string& oid, const utime_t& ut, string& section, string& key, bufferlist& bl);
+  int time_log_add(const string& oid, const utime_t& ut, const string& section, const string& key, bufferlist& bl);
   int time_log_list(const string& oid, utime_t& start_time, utime_t& end_time,
                     int max_entries, list<cls_log_entry>& entries, string& marker, bool *truncated);
   int time_log_trim(const string& oid, utime_t& start_time, utime_t& end_time);
diff --git a/src/rgw/rgw_tools.cc b/src/rgw/rgw_tools.cc
index 95ae52b7941..10f29df4b26 100644
--- a/src/rgw/rgw_tools.cc
+++ b/src/rgw/rgw_tools.cc
@@ -34,7 +34,7 @@ int rgw_put_system_obj(RGWRados *rgwstore, rgw_bucket& bucket, string& oid, cons
   return ret;
 }
 
-int rgw_get_system_obj(RGWRados *rgwstore, void *ctx, rgw_bucket& bucket, string& key, bufferlist& bl,
+int rgw_get_system_obj(RGWRados *rgwstore, void *ctx, rgw_bucket& bucket, const string& key, bufferlist& bl,
                        RGWObjVersionTracker *objv_tracker, time_t *pmtime, map<string, bufferlist> *pattrs)
 {
   int ret;
diff --git a/src/rgw/rgw_tools.h b/src/rgw/rgw_tools.h
index 0c3ba0559c7..3e673c1656b 100644
--- a/src/rgw/rgw_tools.h
+++ b/src/rgw/rgw_tools.h
@@ -13,7 +13,7 @@ struct obj_version;
 
 int rgw_put_system_obj(RGWRados *rgwstore, rgw_bucket& bucket, string& oid, const char *data, size_t size, bool exclusive,
                        RGWObjVersionTracker *objv_tracker, time_t set_mtime, map<string, bufferlist> *pattrs = NULL);
-int rgw_get_system_obj(RGWRados *rgwstore, void *ctx, rgw_bucket& bucket, string& key, bufferlist& bl,
+int rgw_get_system_obj(RGWRados *rgwstore, void *ctx, rgw_bucket& bucket, const string& key, bufferlist& bl,
                        RGWObjVersionTracker *objv_tracker, time_t *pmtime, map<string, bufferlist> *pattrs = NULL);
 
 int rgw_tools_init(CephContext *cct);
diff --git a/src/rgw/rgw_user.cc b/src/rgw/rgw_user.cc
index 23cfc418674..8cb28a2b314 100644
--- a/src/rgw/rgw_user.cc
+++ b/src/rgw/rgw_user.cc
@@ -2319,7 +2319,7 @@ public:
     return rgw_delete_user(store, info, objv_tracker);
   }
 
-  void get_pool_and_oid(RGWRados *store, string& key, rgw_bucket& bucket, string& oid) {
+  void get_pool_and_oid(RGWRados *store, const string& key, rgw_bucket& bucket, string& oid) {
     oid = key;
     bucket = store->zone.user_uid_pool;
   }