Merge remote-tracking branch 'upstream/next'

52 files changed, 599 insertions, 367 deletions

diff --git a/ceph.spec.in b/ceph.spec.in
index 6ee1c861fee..4365eb55eb0 100644
--- a/ceph.spec.in
+++ b/ceph.spec.in
@@ -146,6 +146,7 @@ managers such as Pacemaker.
 Summary:	RADOS distributed object store client library
 Group:		System Environment/Libraries
 License:	LGPL-2.0
+Obsoletes:	ceph-libs
 %description -n librados2
 RADOS is a reliable, autonomic distributed object storage cluster
 developed as part of the Ceph distributed storage system. This is a
@@ -156,6 +157,7 @@ store using a simple file-like interface.
 Summary:	RADOS block device client library
 Group:		System Environment/Libraries
 License:	LGPL-2.0
+Obsoletes:	ceph-libs
 %description -n librbd1
 RBD is a block device striped across multiple distributed objects in
 RADOS, a reliable, autonomic distributed object storage cluster
@@ -166,6 +168,7 @@ shared library allowing applications to manage these block devices.
 Summary:	Ceph distributed file system client library
 Group:		System Environment/Libraries
 License:	LGPL-2.0
+Obsoletes:	ceph-libs
 %description -n libcephfs1
 Ceph is a distributed network file system designed to provide excellent
 performance, reliability, and scalability. This is a shared library
diff --git a/qa/workunits/mon/caps.sh b/qa/workunits/mon/caps.sh
new file mode 100755
index 00000000000..f5aebbbb9f4
--- /dev/null
+++ b/qa/workunits/mon/caps.sh
@@ -0,0 +1,55 @@
+#!/bin/bash
+
+tmp=/tmp/cephtest-mon-caps-madness
+
+exit_on_error=1
+
+[[ ! -z $TEST_EXIT_ON_ERROR ]] && exit_on_error=$TEST_EXIT_ON_ERROR
+
+expect()
+{
+  cmd=$1
+  expected_ret=$2
+
+  echo $cmd
+  eval $cmd >&/dev/null
+  ret=$?
+
+  if [[ $ret -ne $expected_ret ]]; then
+    echo "Error: Expected return $expected_ret, got $ret"
+    [[ $exit_on_error -eq 1 ]] && exit 1
+    return 1
+  fi
+
+  return 0
+}
+
+expect "ceph auth get-or-create client.bazar > $tmp.bazar.keyring" 0
+expect "ceph -k $tmp.bazar.keyring --user bazar mon_status" 13
+ceph auth del client.bazar
+
+c="'allow command \"auth list\", allow command mon_status'"
+expect "ceph auth get-or-create client.foo mon $c > $tmp.foo.keyring" 0
+expect "ceph -k $tmp.foo.keyring --user foo mon_status" 0
+expect "ceph -k $tmp.foo.keyring --user foo auth list" 0
+expect "ceph -k $tmp.foo.keyring --user foo auth export" 13
+expect "ceph -k $tmp.foo.keyring --user foo auth del client.bazar" 13
+expect "ceph -k $tmp.foo.keyring --user foo osd dump" 13
+expect "ceph -k $tmp.foo.keyring --user foo pg dump" 13
+expect "ceph -k $tmp.foo.keyring --user foo quorum_status" 13
+ceph auth del client.foo
+
+c="'allow command service with prefix=list, allow command mon_status'"
+expect "ceph auth get-or-create client.bar mon $c > $tmp.bar.keyring" 0
+expect "ceph -k $tmp.bar.keyring --user bar mon_status" 0
+expect "ceph -k $tmp.bar.keyring --user bar auth list" 13
+expect "ceph -k $tmp.bar.keyring --user bar auth export" 13
+expect "ceph -k $tmp.bar.keyring --user bar auth del client.foo" 13
+expect "ceph -k $tmp.bar.keyring --user bar osd dump" 13
+expect "ceph -k $tmp.bar.keyring --user bar pg dump" 13
+expect "ceph -k $tmp.bar.keyring --user bar quorum_status" 13
+ceph auth del client.bar
+
+rm $tmp.bazar.keyring $tmp.foo.keyring $tmp.bar.keyring
+
+echo OK
\ No newline at end of file
diff --git a/src/ceph.in b/src/ceph.in
index 8c1f4730667..5a6ddd8abf8 100755
--- a/src/ceph.in
+++ b/src/ceph.in
@@ -290,6 +290,8 @@ def admin_socket(asok_path, cmd):
         sock.sendall(' '.join(cmd) + '\0')
 
         len_str = sock.recv(4)
+        if len(len_str) < 4:
+            raise RuntimeError("no data returned from admin socket")
         l, = struct.unpack(">I", len_str)
         ret = ''
 
diff --git a/src/cls/replica_log/cls_replica_log_types.h b/src/cls/replica_log/cls_replica_log_types.h
index acd55dde533..6056f8e4468 100644
--- a/src/cls/replica_log/cls_replica_log_types.h
+++ b/src/cls/replica_log/cls_replica_log_types.h
@@ -61,7 +61,7 @@ struct cls_replica_log_progress_marker {
                                     position_time(time) {}
   cls_replica_log_progress_marker(const string& entity, const string& marker,
                                   const utime_t& time,
-                                  const std::list<cls_replica_log_item_marker> b) :
+                                  const std::list<cls_replica_log_item_marker>& b) :
                                     entity_id(entity), position_marker(marker),
                                     position_time(time),
                                     items(b) {}
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/cls/rgw/cls_rgw_client.cc b/src/cls/rgw/cls_rgw_client.cc
index 291ca705d7b..165ca437987 100644
--- a/src/cls/rgw/cls_rgw_client.cc
+++ b/src/cls/rgw/cls_rgw_client.cc
@@ -187,8 +187,8 @@ int cls_rgw_bi_log_list(IoCtx& io_ctx, string& oid, string& marker, uint32_t max
 
 int cls_rgw_bi_log_trim(IoCtx& io_ctx, string& oid, string& start_marker, string& end_marker)
 {
-  int r;
   do {
+    int r;
     bufferlist in, out;
     cls_rgw_bi_log_trim_op call;
     call.start_marker = start_marker;
diff --git a/src/common/Cond.h b/src/common/Cond.h
index ee95a65b5b6..e6a13ae48bb 100644
--- a/src/common/Cond.h
+++ b/src/common/Cond.h
@@ -156,4 +156,36 @@ public:
   }
 };
 
+/**
+ * Context providing a simple wait() mechanism to wait for completion
+ *
+ * The context will not be deleted as part of complete and must live
+ * until wait() returns.
+ */
+class C_SaferCond : public Context {
+  Mutex lock;    ///< Mutex to take
+  Cond cond;     ///< Cond to signal
+  bool done;     ///< true after finish() has been called
+  int rval;      ///< return value
+public:
+  C_SaferCond() : lock("C_SaferCond"), done(false), rval(0) {}
+  void finish(int r) { complete(r); }
+
+  /// We overload complete in order to not delete the context
+  void complete(int r) {
+    Mutex::Locker l(lock);
+    done = true;
+    rval = r;
+    cond.Signal();
+  }
+
+  /// Returns rval once the Context is called
+  int wait() {
+    Mutex::Locker l(lock);
+    while (!done)
+      cond.Wait(lock);
+    return rval;
+  }
+};
+
 #endif
diff --git a/src/common/Finisher.cc b/src/common/Finisher.cc
index 72bfb6f9aa7..a1462a184d9 100644
--- a/src/common/Finisher.cc
+++ b/src/common/Finisher.cc
@@ -53,13 +53,11 @@ void *Finisher::finisher_thread_entry()
 	   p != ls.end();
 	   ++p) {
 	if (*p) {
-	  (*p)->finish(0);
-	  delete *p;
+	  (*p)->complete(0);
 	} else {
 	  assert(!ls_rval.empty());
 	  Context *c = ls_rval.front().first;
-	  c->finish(ls_rval.front().second);
-	  delete c;
+	  c->complete(ls_rval.front().second);
 	  ls_rval.pop_front();
 	}
 	if (logger)
diff --git a/src/mon/MonCap.cc b/src/mon/MonCap.cc
index 1114ca3b9da..8e35b775247 100644
--- a/src/mon/MonCap.cc
+++ b/src/mon/MonCap.cc
@@ -261,15 +261,21 @@ bool MonCap::is_capable(CephContext *cct,
     if (cct)
       ldout(cct, 20) << " allow so far " << allow << ", doing grant " << *p << dendl;
 
-    if (p->is_allow_all())
+    if (p->is_allow_all()) {
+      if (cct)
+	ldout(cct, 20) << " allow all" << dendl;
       return true;
+    }
 
     // check enumerated caps
     allow = allow | p->get_allowed(cct, name, service, command, command_args);
-    if (!((op_may_read && !(allow & MON_CAP_R)) ||
-	  (op_may_write && !(allow & MON_CAP_W)) ||
-	  (op_may_exec && !(allow & MON_CAP_X))))
+    if ((!op_may_read || (allow & MON_CAP_R)) &&
+	(!op_may_write || (allow & MON_CAP_W)) &&
+	(!op_may_exec || (allow & MON_CAP_X))) {
+      if (cct)
+	ldout(cct, 20) << " match" << dendl;
       return true;
+    }
   }
   return false;
 }
diff --git a/src/mon/MonClient.cc b/src/mon/MonClient.cc
index 8139b0259bf..299379ac249 100644
--- a/src/mon/MonClient.cc
+++ b/src/mon/MonClient.cc
@@ -318,9 +318,12 @@ int MonClient::init()
 
 void MonClient::shutdown()
 {
+  ldout(cct, 10) << __func__ << "shutdown" << dendl;
   monc_lock.Lock();
   while (!version_requests.empty()) {
     version_requests.begin()->second->context->complete(-ECANCELED);
+    ldout(cct, 20) << __func__ << " canceling and discarding version request "
+		   << version_requests.begin()->second << dendl;
     delete version_requests.begin()->second;
     version_requests.erase(version_requests.begin());
   }
@@ -872,12 +875,13 @@ int MonClient::start_mon_command(int rank,
 
 void MonClient::get_version(string map, version_t *newest, version_t *oldest, Context *onfinish)
 {
-  ldout(cct, 10) << "get_version " << map << dendl;
+  version_req_d *req = new version_req_d(onfinish, newest, oldest);
+  ldout(cct, 10) << "get_version " << map << " req " << req << dendl;
   Mutex::Locker l(monc_lock);
   MMonGetVersion *m = new MMonGetVersion();
   m->what = map;
   m->handle = ++version_req_id;
-  version_requests[m->handle] = new version_req_d(onfinish, newest, oldest);
+  version_requests[m->handle] = req;
   _send_mon_message(m);
 }
 
@@ -886,10 +890,11 @@ void MonClient::handle_get_version_reply(MMonGetVersionReply* m)
   assert(monc_lock.is_locked());
   map<tid_t, version_req_d*>::iterator iter = version_requests.find(m->handle);
   if (iter == version_requests.end()) {
-    ldout(cct, 0) << "version request with handle " << m->handle
+    ldout(cct, 0) << __func__ << " version request with handle " << m->handle
 		  << " not found" << dendl;
   } else {
     version_req_d *req = iter->second;
+    ldout(cct, 10) << __func__ << " finishing " << req << " version " << m->version << dendl;
     version_requests.erase(iter);
     if (req->newest)
       *req->newest = m->version;
diff --git a/src/mon/Monitor.cc b/src/mon/Monitor.cc
index 90750dd7b11..7e484e8db6b 100644
--- a/src/mon/Monitor.cc
+++ b/src/mon/Monitor.cc
@@ -1523,8 +1523,10 @@ bool Monitor::_allowed_command(MonSession *s, map<string, cmd_vartype>& cmd)
 {
   bool retval = false;
 
-  if (s->caps.is_allow_all())
+  if (s->caps.is_allow_all()) {
+    dout(10) << __func__ << " allow_all" << dendl;
     return true;
+  }
 
   string prefix;
   cmd_getval(g_ceph_context, cmd, "prefix", prefix);
@@ -1538,10 +1540,11 @@ bool Monitor::_allowed_command(MonSession *s, map<string, cmd_vartype>& cmd)
   }
 
   if (s->caps.is_capable(g_ceph_context, s->inst.name,
-			 "", prefix, strmap, false, false, false)) {
+			 "", prefix, strmap, false, false, true)) {
     retval = true; 
   }
 
+  dout(10) << __func__ << " = " << retval << dendl;
   return retval;
 }
 
diff --git a/src/mon/OSDMonitor.cc b/src/mon/OSDMonitor.cc
index c8baac58c83..20e4eac88cb 100644
--- a/src/mon/OSDMonitor.cc
+++ b/src/mon/OSDMonitor.cc
@@ -185,8 +185,11 @@ void OSDMonitor::update_from_paxos(bool *need_bootstrap)
     mon->pgmon()->check_osd_map(osdmap.epoch);
   }
 
-  update_logger();
+  check_subs();
+
   share_map_with_random_osd();
+  update_logger();
+
   process_failures();
 
   // make sure our feature bits reflect the latest map
@@ -294,9 +297,6 @@ void OSDMonitor::on_active()
 {
   update_logger();
 
-  send_to_waiting();
-  check_subs();
-
   if (thrash_map) {
     if (mon->is_leader()) {
       if (thrash())
@@ -310,22 +310,25 @@ void OSDMonitor::on_active()
     mon->clog.info() << "osdmap " << osdmap << "\n"; 
 
   if (!mon->is_leader()) {
-    kick_all_failures();
+    list<MOSDFailure*> ls;
+    take_all_failures(ls);
+    while (!ls.empty()) {
+      dispatch(ls.front());
+      ls.pop_front();
+    }
   }
 }
 
 void OSDMonitor::on_shutdown()
 {
   dout(10) << __func__ << dendl;
-  map<epoch_t, list<PaxosServiceMessage*> >::iterator p = waiting_for_map.begin();
-  while (p != waiting_for_map.end()) {
-    while (!p->second.empty()) {
-      Message *m = p->second.front();
-      dout(20) << " discarding " << m << " " << *m << dendl;
-      m->put();
-      p->second.pop_front();
-    }
-    waiting_for_map.erase(p++);
+
+  // discard failure info, waiters
+  list<MOSDFailure*> ls;
+  take_all_failures(ls);
+  while (!ls.empty()) {
+    ls.front()->put();
+    ls.pop_front();
   }
 }
 
@@ -1049,23 +1052,16 @@ void OSDMonitor::process_failures()
   }
 }
 
-void OSDMonitor::kick_all_failures()
+void OSDMonitor::take_all_failures(list<MOSDFailure*>& ls)
 {
-  dout(10) << "kick_all_failures on " << failure_info.size() << " osds" << dendl;
-  assert(!mon->is_leader());
+  dout(10) << __func__ << " on " << failure_info.size() << " osds" << dendl;
 
-  list<MOSDFailure*> ls;
   for (map<int,failure_info_t>::iterator p = failure_info.begin();
        p != failure_info.end();
        ++p) {
     p->second.take_report_messages(ls);
   }
   failure_info.clear();
-
-  while (!ls.empty()) {
-    dispatch(ls.front());
-    ls.pop_front();
-  }
 }
 
 
@@ -1311,7 +1307,6 @@ void OSDMonitor::_reply_map(PaxosServiceMessage *m, epoch_t e)
 {
   dout(7) << "_reply_map " << e
 	  << " from " << m->get_orig_source_inst()
-	  << " for " << m
 	  << dendl;
   send_latest(m, e);
 }
@@ -1450,53 +1445,15 @@ bool OSDMonitor::prepare_remove_snaps(MRemoveSnaps *m)
 // ---------------
 // map helpers
 
-void OSDMonitor::send_to_waiting()
-{
-  dout(10) << "send_to_waiting " << osdmap.get_epoch() << dendl;
-
-  map<epoch_t, list<PaxosServiceMessage*> >::iterator p = waiting_for_map.begin();
-  while (p != waiting_for_map.end()) {
-    epoch_t from = p->first;
-    
-    if (from) {
-      if (from <= osdmap.get_epoch()) {
-	while (!p->second.empty()) {
-	  send_incremental(p->second.front(), from);
-	  p->second.front()->put();
-	  p->second.pop_front();
-	}
-      } else {
-	dout(10) << "send_to_waiting from " << from << dendl;
-	++p;
-	continue;
-      }
-    } else {
-      while (!p->second.empty()) {
-	send_full(p->second.front());
-	p->second.front()->put();
-	p->second.pop_front();
-      }
-    }
-
-    waiting_for_map.erase(p++);
-  }
-}
-
 void OSDMonitor::send_latest(PaxosServiceMessage *m, epoch_t start)
 {
-  if (is_readable()) {
-    dout(5) << "send_latest to " << m->get_orig_source_inst()
-	    << " start " << start << dendl;
-    if (start == 0)
-      send_full(m);
-    else
-      send_incremental(m, start);
-    m->put();
-  } else {
-    dout(5) << "send_latest to " << m->get_orig_source_inst()
-	    << " start " << start << " later" << dendl;
-    waiting_for_map[start].push_back(m);
-  }
+  dout(5) << "send_latest to " << m->get_orig_source_inst()
+	  << " start " << start << dendl;
+  if (start == 0)
+    send_full(m);
+  else
+    send_incremental(m, start);
+  m->put();
 }
 
 
@@ -1651,6 +1608,7 @@ epoch_t OSDMonitor::blacklist(const entity_addr_t& a, utime_t until)
 
 void OSDMonitor::check_subs()
 {
+  dout(10) << __func__ << dendl;
   string type = "osdmap";
   if (mon->session_map.subs.count(type) == 0)
     return;
@@ -1664,6 +1622,8 @@ void OSDMonitor::check_subs()
 
 void OSDMonitor::check_sub(Subscription *sub)
 {
+  dout(10) << __func__ << " " << sub << " next " << sub->next
+	   << (sub->onetime ? " (onetime)":" (ongoing)") << dendl;
   if (sub->next <= osdmap.get_epoch()) {
     if (sub->next >= 1)
       send_incremental(sub->next, sub->session->inst, sub->incremental_onetime);
diff --git a/src/mon/OSDMonitor.h b/src/mon/OSDMonitor.h
index d6553228321..d7cb8fdf369 100644
--- a/src/mon/OSDMonitor.h
+++ b/src/mon/OSDMonitor.h
@@ -118,8 +118,6 @@ public:
   OSDMap osdmap;
 
 private:
-  map<epoch_t, list<PaxosServiceMessage*> > waiting_for_map;
-
   // [leader]
   OSDMap::Incremental pending_inc;
   map<int, failure_info_t> failure_info;
@@ -192,7 +190,6 @@ private:
   bool can_mark_in(int o);
 
   // ...
-  void send_to_waiting();     // send current map to waiters.
   MOSDMap *build_latest_full();
   MOSDMap *build_incremental(epoch_t first, epoch_t last);
   void send_full(PaxosServiceMessage *m);
@@ -212,7 +209,7 @@ private:
   bool prepare_failure(class MOSDFailure *m);
   bool prepare_mark_me_down(class MOSDMarkMeDown *m);
   void process_failures();
-  void kick_all_failures();
+  void take_all_failures(list<MOSDFailure*>& ls);
 
   bool preprocess_boot(class MOSDBoot *m);
   bool prepare_boot(class MOSDBoot *m);
diff --git a/src/mon/PGMap.cc b/src/mon/PGMap.cc
index dfb6b1acfd3..8b37376ba7a 100644
--- a/src/mon/PGMap.cc
+++ b/src/mon/PGMap.cc
@@ -728,23 +728,29 @@ void PGMap::recovery_summary(Formatter *f, ostream *out) const
     }
     first = false;
   }
-  if (pg_sum_delta.stats.sum.num_objects_recovered ||
-      pg_sum_delta.stats.sum.num_bytes_recovered ||
-      pg_sum_delta.stats.sum.num_keys_recovered) {
-    uint64_t objps = pg_sum_delta.stats.sum.num_objects_recovered / (double)stamp_delta;
-    uint64_t bps = pg_sum_delta.stats.sum.num_bytes_recovered / (double)stamp_delta;
-    uint64_t kps = pg_sum_delta.stats.sum.num_keys_recovered / (double)stamp_delta;
+
+  // make non-negative; we can get negative values if osds send
+  // uncommitted stats and then "go backward" or if they are just
+  // buggy/wrong.
+  pool_stat_t pos_delta = pg_sum_delta;
+  pos_delta.floor(0);
+  if (pos_delta.stats.sum.num_objects_recovered ||
+      pos_delta.stats.sum.num_bytes_recovered ||
+      pos_delta.stats.sum.num_keys_recovered) {
+    int64_t objps = pos_delta.stats.sum.num_objects_recovered / (double)stamp_delta;
+    int64_t bps = pos_delta.stats.sum.num_bytes_recovered / (double)stamp_delta;
+    int64_t kps = pos_delta.stats.sum.num_keys_recovered / (double)stamp_delta;
     if (f) {
-      f->dump_unsigned("recovering_objects_per_sec", objps);
-      f->dump_unsigned("recovering_bytes_per_sec", bps);
-      f->dump_unsigned("recovering_keys_per_sec", kps);
+      f->dump_int("recovering_objects_per_sec", objps);
+      f->dump_int("recovering_bytes_per_sec", bps);
+      f->dump_int("recovering_keys_per_sec", kps);
     } else {
       if (!first)
 	*out << "; ";
       *out << " recovering "
 	   << si_t(objps) << " o/s, "
 	   << si_t(bps) << "B/s";
-      if (pg_sum_delta.stats.sum.num_keys_recovered)
+      if (pos_delta.stats.sum.num_keys_recovered)
 	*out << ", " << si_t(kps) << " key/s";
     }
   }
@@ -775,7 +781,7 @@ void PGMap::clear_delta()
 {
   pg_sum_delta = pool_stat_t();
   pg_sum_deltas.clear();
-  stamp_delta = ceph_clock_now(g_ceph_context);
+  stamp_delta = utime_t();
 }
 
 void PGMap::print_summary(Formatter *f, ostream *out) const
@@ -815,29 +821,34 @@ void PGMap::print_summary(Formatter *f, ostream *out) const
 	 << kb_t(osd_sum.kb) << " avail";
   }
 
-  if (pg_sum_delta.stats.sum.num_rd ||
-      pg_sum_delta.stats.sum.num_wr) {
+  // make non-negative; we can get negative values if osds send
+  // uncommitted stats and then "go backward" or if they are just
+  // buggy/wrong.
+  pool_stat_t pos_delta = pg_sum_delta;
+  pos_delta.floor(0);
+  if (pos_delta.stats.sum.num_rd ||
+      pos_delta.stats.sum.num_wr) {
     if (!f)
       *out << "; ";
-    if (pg_sum_delta.stats.sum.num_rd) {
-      uint64_t rd = (pg_sum_delta.stats.sum.num_rd_kb << 10) / (double)stamp_delta;
+    if (pos_delta.stats.sum.num_rd) {
+      int64_t rd = (pos_delta.stats.sum.num_rd_kb << 10) / (double)stamp_delta;
       if (f) {
-	f->dump_unsigned("read_bytes_sec", rd);
+	f->dump_int("read_bytes_sec", rd);
       } else {
 	*out << si_t(rd) << "B/s rd, ";
       }
     }
-    if (pg_sum_delta.stats.sum.num_wr) {
-      uint64_t wr = (pg_sum_delta.stats.sum.num_wr_kb << 10) / (double)stamp_delta;
+    if (pos_delta.stats.sum.num_wr) {
+      int64_t wr = (pos_delta.stats.sum.num_wr_kb << 10) / (double)stamp_delta;
       if (f) {
-	f->dump_unsigned("write_bytes_sec", wr);
+	f->dump_int("write_bytes_sec", wr);
       } else {
 	*out << si_t(wr) << "B/s wr, ";
       }
     }
-    uint64_t iops = (pg_sum_delta.stats.sum.num_rd + pg_sum_delta.stats.sum.num_wr) / (double)stamp_delta;
+    int64_t iops = (pos_delta.stats.sum.num_rd + pos_delta.stats.sum.num_wr) / (double)stamp_delta;
     if (f) {
-      f->dump_unsigned("op_per_sec", iops);
+      f->dump_int("op_per_sec", iops);
     } else {
       *out << si_t(iops) << "op/s";
     }
diff --git a/src/mon/PGMonitor.cc b/src/mon/PGMonitor.cc
index 1f11d5486cf..648a8fe2384 100644
--- a/src/mon/PGMonitor.cc
+++ b/src/mon/PGMonitor.cc
@@ -1794,8 +1794,8 @@ void PGMonitor::get_health(list<pair<health_status_t,string> >& summary,
       ss << sum << " requests are blocked > " << g_conf->mon_osd_max_op_age << " sec";
       summary.push_back(make_pair(HEALTH_WARN, ss.str()));
 
-      unsigned num_slow_osds = 0;
       if (detail) {
+        unsigned num_slow_osds = 0;
 	// do per-osd warnings
 	for (hash_map<int32_t,osd_stat_t>::const_iterator p = pg_map.osd_stat.begin();
 	     p != pg_map.osd_stat.end();
diff --git a/src/mon/PaxosService.h b/src/mon/PaxosService.h
index a5761d19ad8..74d5a90494c 100644
--- a/src/mon/PaxosService.h
+++ b/src/mon/PaxosService.h
@@ -438,8 +438,9 @@ public:
   /**
    * This is called when the Paxos state goes to active.
    *
-   * @remarks It's a courtesy method, in case the class implementing this 
-   *	      service has anything it wants/needs to do at that time.
+   * On the peon, this is after each election.
+   * On the leader, this is after each election, *and* after each completed
+   * proposal.
    *
    * @note This function may get called twice in certain recovery cases.
    */
diff --git a/src/os/FileStore.cc b/src/os/FileStore.cc
index 1363eff27d1..10f2b1f2aad 100644
--- a/src/os/FileStore.cc
+++ b/src/os/FileStore.cc
@@ -2028,8 +2028,7 @@ void FileStore::_finish_op(OpSequencer *osr)
   logger->tinc(l_os_apply_lat, lat);
 
   if (o->onreadable_sync) {
-    o->onreadable_sync->finish(0);
-    delete o->onreadable_sync;
+    o->onreadable_sync->complete(0);
   }
   op_finisher.queue(o->onreadable);
   delete o;
@@ -2126,8 +2125,7 @@ int FileStore::queue_transactions(Sequencer *posr, list<Transaction*> &tls,
   // start on_readable finisher after we queue journal item, as on_readable callback
   // is allowed to delete the Transaction
   if (onreadable_sync) {
-    onreadable_sync->finish(r);
-    delete onreadable_sync;
+    onreadable_sync->complete(r);
   }
   op_finisher.queue(onreadable, r);
 
diff --git a/src/osd/OSD.cc b/src/osd/OSD.cc
index 1ee4c09a63e..8cc9e31459d 100644
--- a/src/osd/OSD.cc
+++ b/src/osd/OSD.cc
@@ -3267,21 +3267,30 @@ bool remove_dir(
       }
       t->remove(coll, *i);
       if (num >= g_conf->osd_target_transaction_size) {
-	store->apply_transaction(osr, *t);
+	C_SaferCond waiter;
+	store->queue_transaction(osr, t, &waiter);
+	bool cont = dstate->pause_clearing();
+	waiter.wait();
+	if (cont)
+	  cont = dstate->resume_clearing();
 	delete t;
-	if (!dstate->check_canceled()) {
-	  // canceled!
+	if (!cont)
 	  return false;
-	}
 	t = new ObjectStore::Transaction;
 	num = 0;
       }
     }
     olist.clear();
   }
-  store->apply_transaction(osr, *t);
+
+  C_SaferCond waiter;
+  store->queue_transaction(osr, t, &waiter);
+  bool cont = dstate->pause_clearing();
+  waiter.wait();
+  if (cont)
+    cont = dstate->resume_clearing();
   delete t;
-  return true;
+  return cont;
 }
 
 void OSD::RemoveWQ::_process(pair<PGRef, DeletingStateRef> item)
diff --git a/src/osd/OSD.h b/src/osd/OSD.h
index 238c5b43594..04ad4dcd7d7 100644
--- a/src/osd/OSD.h
+++ b/src/osd/OSD.h
@@ -148,6 +148,7 @@ class DeletingState {
   enum {
     QUEUED,
     CLEARING_DIR,
+    CLEARING_WAITING,
     DELETING_DIR,
     DELETED_DIR,
     CANCELED,
@@ -160,8 +161,23 @@ public:
     lock("DeletingState::lock"), status(QUEUED), stop_deleting(false),
     pgid(in.first), old_pg_state(in.second) {}
 
-  /// check whether removal was canceled
-  bool check_canceled() {
+  /// transition status to clearing
+  bool start_clearing() {
+    Mutex::Locker l(lock);
+    assert(
+      status == QUEUED ||
+      status == DELETED_DIR);
+    if (stop_deleting) {
+      status = CANCELED;
+      cond.Signal();
+      return false;
+    }
+    status = CLEARING_DIR;
+    return true;
+  } ///< @return false if we should cancel deletion
+
+  /// transition status to CLEARING_WAITING
+  bool pause_clearing() {
     Mutex::Locker l(lock);
     assert(status == CLEARING_DIR);
     if (stop_deleting) {
@@ -169,15 +185,14 @@ public:
       cond.Signal();
       return false;
     }
+    status = CLEARING_WAITING;
     return true;
-  } ///< @return false if canceled, true if we should continue
+  } ///< @return false if we should cancel deletion
 
-  /// transition status to clearing
-  bool start_clearing() {
+  /// transition status to CLEARING_DIR
+  bool resume_clearing() {
     Mutex::Locker l(lock);
-    assert(
-      status == QUEUED ||
-      status == DELETED_DIR);
+    assert(status == CLEARING_WAITING);
     if (stop_deleting) {
       status = CANCELED;
       cond.Signal();
@@ -215,11 +230,10 @@ public:
     /**
      * If we are in DELETING_DIR or CLEARING_DIR, there are in progress
      * operations we have to wait for before continuing on.  States
-     * DELETED_DIR, QUEUED, and CANCELED either check for stop_deleting
-     * prior to performing any operations or signify the end of the
-     * deleting process.  We don't want to wait to leave the QUEUED
-     * state, because this might block the caller behind an entire pg
-     * removal.
+     * CLEARING_WAITING and QUEUED indicate that the remover will check
+     * stop_deleting before queueing any further operations.  CANCELED
+     * indicates that the remover has already halted.  DELETED_DIR
+     * indicates that the deletion has been fully queueud.
      */
     while (status == DELETING_DIR || status == CLEARING_DIR)
       cond.Wait(lock);
diff --git a/src/osd/PG.cc b/src/osd/PG.cc
index 2a07887a2a3..7373357db11 100644
--- a/src/osd/PG.cc
+++ b/src/osd/PG.cc
@@ -4511,7 +4511,8 @@ void PG::start_flush(ObjectStore::Transaction *t,
 /* Called before initializing peering during advance_map */
 void PG::start_peering_interval(const OSDMapRef lastmap,
 				const vector<int>& newup,
-				const vector<int>& newacting)
+				const vector<int>& newacting,
+				ObjectStore::Transaction *t)
 {
   const OSDMapRef osdmap = get_osdmap();
 
@@ -4600,7 +4601,7 @@ void PG::start_peering_interval(const OSDMapRef lastmap,
 
     
   // pg->on_*
-  on_change();
+  on_change(t);
 
   assert(!deleting);
 
@@ -5237,7 +5238,8 @@ boost::statechart::result PG::RecoveryState::Reset::react(const AdvMap& advmap)
     pg->is_split(advmap.lastmap, advmap.osdmap)) {
     dout(10) << "up or acting affected, calling start_peering_interval again"
 	     << dendl;
-    pg->start_peering_interval(advmap.lastmap, advmap.newup, advmap.newacting);
+    pg->start_peering_interval(advmap.lastmap, advmap.newup, advmap.newacting,
+			       context< RecoveryMachine >().get_cur_transaction());
   }
   return discard_event();
 }
@@ -6894,14 +6896,14 @@ void PG::RecoveryState::WaitUpThru::exit()
 
 void PG::RecoveryState::RecoveryMachine::log_enter(const char *state_name)
 {
-  dout(20) << "enter " << state_name << dendl;
+  dout(5) << "enter " << state_name << dendl;
   pg->osd->pg_recovery_stats.log_enter(state_name);
 }
 
 void PG::RecoveryState::RecoveryMachine::log_exit(const char *state_name, utime_t enter_time)
 {
   utime_t dur = ceph_clock_now(g_ceph_context) - enter_time;
-  dout(20) << "exit " << state_name << " " << dur << " " << event_count << " " << event_time << dendl;
+  dout(5) << "exit " << state_name << " " << dur << " " << event_count << " " << event_time << dendl;
   pg->osd->pg_recovery_stats.log_exit(state_name, ceph_clock_now(g_ceph_context) - enter_time,
 				      event_count, event_time);
   event_count = 0;
diff --git a/src/osd/PG.h b/src/osd/PG.h
index e1f64dbc9fb..819c9c62f62 100644
--- a/src/osd/PG.h
+++ b/src/osd/PG.h
@@ -1703,7 +1703,8 @@ public:
 
   void start_peering_interval(const OSDMapRef lastmap,
 			      const vector<int>& newup,
-			      const vector<int>& newacting);
+			      const vector<int>& newacting,
+			      ObjectStore::Transaction *t);
   void start_flush(ObjectStore::Transaction *t,
 		   list<Context *> *on_applied,
 		   list<Context *> *on_safe);
@@ -1793,7 +1794,7 @@ public:
   virtual bool same_for_rep_modify_since(epoch_t e) = 0;
 
   virtual void on_role_change() = 0;
-  virtual void on_change() = 0;
+  virtual void on_change(ObjectStore::Transaction *t) = 0;
   virtual void on_activate() = 0;
   virtual void on_flushed() = 0;
   virtual void on_shutdown() = 0;
diff --git a/src/osd/ReplicatedPG.cc b/src/osd/ReplicatedPG.cc
index 453fdacfb76..9c8d42dbf3c 100644
--- a/src/osd/ReplicatedPG.cc
+++ b/src/osd/ReplicatedPG.cc
@@ -5517,10 +5517,14 @@ void ReplicatedPG::submit_push_data(
   ObjectStore::Transaction *t)
 {
   coll_t target_coll;
-  if (first && complete)
+  if (first && complete) {
     target_coll = coll;
-  else
+  } else {
+    dout(10) << __func__ << ": Creating oid "
+	     << recovery_info.soid << " in the temp collection" << dendl;
+    temp_contents.insert(recovery_info.soid);
     target_coll = get_temp_coll(t);
+  }
 
   if (first) {
     pg_log.revise_have(recovery_info.soid, eversion_t());
@@ -5547,8 +5551,13 @@ void ReplicatedPG::submit_push_data(
 	      attrs);
 
   if (complete) {
-    if (!first)
+    if (!first) {
+      assert(temp_contents.count(recovery_info.soid));
+      dout(10) << __func__ << ": Removing oid "
+	       << recovery_info.soid << " from the temp collection" << dendl;
+      temp_contents.erase(recovery_info.soid);
       t->collection_move(coll, target_coll, recovery_info.soid);
+    }
 
     submit_push_complete(recovery_info, t);
   }
@@ -6702,6 +6711,15 @@ void ReplicatedPG::on_shutdown()
 void ReplicatedPG::on_flushed()
 {
   assert(object_contexts.empty());
+  if (have_temp_coll() &&
+      !osd->store->collection_empty(get_temp_coll())) {
+    vector<hobject_t> objects;
+    osd->store->collection_list(get_temp_coll(), objects);
+    derr << __func__ << ": found objects in the temp collection: "
+	 << objects << ", crashing now"
+	 << dendl;
+    assert(0 == "found garbage in the temp collection");
+  }
 }
 
 void ReplicatedPG::on_activate()
@@ -6717,7 +6735,7 @@ void ReplicatedPG::on_activate()
   }
 }
 
-void ReplicatedPG::on_change()
+void ReplicatedPG::on_change(ObjectStore::Transaction *t)
 {
   dout(10) << "on_change" << dendl;
 
@@ -6751,6 +6769,16 @@ void ReplicatedPG::on_change()
   pulling.clear();
   pull_from_peer.clear();
 
+  // clear temp
+  for (set<hobject_t>::iterator i = temp_contents.begin();
+       i != temp_contents.end();
+       ++i) {
+    dout(10) << __func__ << ": Removing oid "
+	     << *i << " from the temp collection" << dendl;
+    t->remove(get_temp_coll(t), *i);
+  }
+  temp_contents.clear();
+
   // clear snap_trimmer state
   snap_trimmer_machine.process_event(Reset());
 
diff --git a/src/osd/ReplicatedPG.h b/src/osd/ReplicatedPG.h
index 8f0cf6ec31c..0d4867f6e6d 100644
--- a/src/osd/ReplicatedPG.h
+++ b/src/osd/ReplicatedPG.h
@@ -551,6 +551,9 @@ protected:
   };
   map<hobject_t, PullInfo> pulling;
 
+  // Track contents of temp collection, clear on reset
+  set<hobject_t> temp_contents;
+
   ObjectRecoveryInfo recalc_subsets(const ObjectRecoveryInfo& recovery_info);
   static void trim_pushed_data(const interval_set<uint64_t> &copy_subset,
 			       const interval_set<uint64_t> &intervals_received,
@@ -1042,7 +1045,7 @@ public:
   void _finish_mark_all_unfound_lost(list<ObjectContext*>& obcs);
 
   void on_role_change();
-  void on_change();
+  void on_change(ObjectStore::Transaction *t);
   void on_activate();
   void on_flushed();
   void on_removal(ObjectStore::Transaction *t);
diff --git a/src/osd/osd_types.cc b/src/osd/osd_types.cc
index fa3b7ecc45d..02c1ef7b69d 100644
--- a/src/osd/osd_types.cc
+++ b/src/osd/osd_types.cc
@@ -1351,8 +1351,8 @@ void pg_stat_t::generate_test_instances(list<pg_stat_t*>& o)
 void pool_stat_t::dump(Formatter *f) const
 {
   stats.dump(f);
-  f->dump_unsigned("log_size", log_size);
-  f->dump_unsigned("ondisk_log_size", ondisk_log_size);
+  f->dump_int("log_size", log_size);
+  f->dump_int("ondisk_log_size", ondisk_log_size);
 }
 
 void pool_stat_t::encode(bufferlist &bl, uint64_t features) const
diff --git a/src/osd/osd_types.h b/src/osd/osd_types.h
index ff6d70fa69a..3a6db4d8315 100644
--- a/src/osd/osd_types.h
+++ b/src/osd/osd_types.h
@@ -888,6 +888,28 @@ struct object_stat_sum_t {
       num_keys_recovered(0)
   {}
 
+  void floor(int64_t f) {
+#define FLOOR(x) if (x < f) x = f
+    FLOOR(num_bytes);
+    FLOOR(num_objects);
+    FLOOR(num_object_clones);
+    FLOOR(num_object_copies);
+    FLOOR(num_objects_missing_on_primary);
+    FLOOR(num_objects_degraded);
+    FLOOR(num_objects_unfound);
+    FLOOR(num_rd);
+    FLOOR(num_rd_kb);
+    FLOOR(num_wr);
+    FLOOR(num_wr_kb);
+    FLOOR(num_scrub_errors);
+    FLOOR(num_shallow_scrub_errors);
+    FLOOR(num_deep_scrub_errors);
+    FLOOR(num_objects_recovered);
+    FLOOR(num_bytes_recovered);
+    FLOOR(num_keys_recovered);
+#undef FLOOR
+  }
+
   void clear() {
     memset(this, 0, sizeof(*this));
   }
@@ -940,6 +962,12 @@ struct object_stat_collection_t {
     cat_sum.clear();
   }
 
+  void floor(int64_t f) {
+    sum.floor(f);
+    for (map<string,object_stat_sum_t>::iterator p = cat_sum.begin(); p != cat_sum.end(); ++p)
+      p->second.floor(f);
+  }
+
   void add(const object_stat_sum_t& o, const string& cat) {
     sum.add(o);
     if (cat.length())
@@ -1031,6 +1059,14 @@ struct pg_stat_t {
     return make_pair(reported_epoch, reported_seq);
   }
 
+  void floor(int64_t f) {
+    stats.floor(f);
+    if (log_size < f)
+      log_size = f;
+    if (ondisk_log_size < f)
+      ondisk_log_size = f;
+  }
+
   void add(const pg_stat_t& o) {
     stats.add(o.stats);
     log_size += o.log_size;
@@ -1054,12 +1090,20 @@ WRITE_CLASS_ENCODER(pg_stat_t)
  */
 struct pool_stat_t {
   object_stat_collection_t stats;
-  uint64_t log_size;
-  uint64_t ondisk_log_size;    // >= active_log_size
+  int64_t log_size;
+  int64_t ondisk_log_size;    // >= active_log_size
 
   pool_stat_t() : log_size(0), ondisk_log_size(0)
   { }
 
+  void floor(int64_t f) {
+    stats.floor(f);
+    if (log_size < f)
+      log_size = f;
+    if (ondisk_log_size < f)
+      ondisk_log_size = f;
+  }
+
   void add(const pg_stat_t& o) {
     stats.add(o.stats);
     log_size += o.log_size;
diff --git a/src/pybind/ceph_rest_api.py b/src/pybind/ceph_rest_api.py
index 4841022e1d6..fdfe84ee3cb 100755
--- a/src/pybind/ceph_rest_api.py
+++ b/src/pybind/ceph_rest_api.py
@@ -2,11 +2,8 @@
 # vim: ts=4 sw=4 smarttab expandtab
 
 import os
-import sys
-import argparse
 import collections
 import ConfigParser
-import errno
 import json
 import logging
 import logging.handlers
diff --git a/src/rgw/rgw_admin.cc b/src/rgw/rgw_admin.cc
index abf9e26c8e3..67f5f1c68b3 100644
--- a/src/rgw/rgw_admin.cc
+++ b/src/rgw/rgw_admin.cc
@@ -236,32 +236,31 @@ enum {
 static int get_cmd(const char *cmd, const char *prev_cmd, bool *need_more)
 {
   *need_more = false;
-  if (strcmp(cmd, "bucket") == 0 ||
+  // NOTE: please keep the checks in alphabetical order !!!
+  if (strcmp(cmd, "bilog") == 0 ||
+      strcmp(cmd, "bucket") == 0 ||
       strcmp(cmd, "buckets") == 0 ||
-      strcmp(cmd, "user") == 0 ||
       strcmp(cmd, "caps") == 0 ||
+      strcmp(cmd, "datalog") == 0 ||
       strcmp(cmd, "gc") == 0 || 
       strcmp(cmd, "key") == 0 ||
       strcmp(cmd, "log") == 0 ||
+      strcmp(cmd, "mdlog") == 0 ||
+      strcmp(cmd, "metadata") == 0 ||
       strcmp(cmd, "object") == 0 ||
+      strcmp(cmd, "opstate") == 0 ||
       strcmp(cmd, "pool") == 0 ||
       strcmp(cmd, "pools") == 0 ||
-      strcmp(cmd, "subuser") == 0 ||
-      strcmp(cmd, "temp") == 0 ||
-      strcmp(cmd, "usage") == 0 ||
-      strcmp(cmd, "user") == 0 ||
       strcmp(cmd, "region") == 0 ||
       strcmp(cmd, "regions") == 0 ||
       strcmp(cmd, "region-map") == 0 ||
       strcmp(cmd, "regionmap") == 0 ||
-      strcmp(cmd, "zone") == 0 ||
+      strcmp(cmd, "replicalog") == 0 ||
+      strcmp(cmd, "subuser") == 0 ||
       strcmp(cmd, "temp") == 0 ||
-      strcmp(cmd, "metadata") == 0 ||
-      strcmp(cmd, "mdlog") == 0 ||
-      strcmp(cmd, "bilog") == 0 ||
-      strcmp(cmd, "datalog") == 0 ||
-      strcmp(cmd, "opstate") == 0 ||
-      strcmp(cmd, "replicalog") == 0) {
+      strcmp(cmd, "usage") == 0 ||
+      strcmp(cmd, "user") == 0 ||
+      strcmp(cmd, "zone") == 0) {
     *need_more = true;
     return 0;
   }
@@ -1389,11 +1388,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..d4cfdc88e64 100644
--- a/src/rgw/rgw_bucket.cc
+++ b/src/rgw/rgw_bucket.cc
@@ -115,7 +115,7 @@ int rgw_link_bucket(RGWRados *store, string user_id, rgw_bucket& bucket, time_t
   }
 
   if (!update_entrypoint)
-    return false;
+    return 0;
 
   ep.linked = true;
   ep.owner = user_id;
@@ -149,7 +149,7 @@ int rgw_unlink_bucket(RGWRados *store, string user_id, const string& bucket_name
   }
 
   if (!update_entrypoint)
-    return false;
+    return 0;
 
   RGWBucketEntryPoint ep;
   RGWObjVersionTracker ot;
@@ -416,7 +416,6 @@ int RGWBucket::link(RGWBucketAdminOpState& op_state, std::string *err_msg)
   std::string display_name = op_state.get_user_display_name();
   rgw_bucket bucket = op_state.get_bucket();
 
-  string uid_str(user_info.user_id);
   rgw_obj obj(bucket, no_oid);
   RGWObjVersionTracker objv_tracker;
 
@@ -718,7 +717,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;
 
@@ -1419,7 +1418,7 @@ public:
       ret = rgw_unlink_bucket(store, be.owner, be.bucket.name, false);
     }
 
-    return 0;
+    return ret;
   }
 
   struct list_keys_info {
@@ -1464,7 +1463,7 @@ public:
   }
 
   int list_keys_next(void *handle, int max, list<string>& keys, bool *truncated) {
-    list_keys_info *info = (list_keys_info *)handle;
+    list_keys_info *info = static_cast<list_keys_info *>(handle);
 
     string no_filter;
 
@@ -1498,7 +1497,7 @@ public:
   }
 
   void list_keys_complete(void *handle) {
-    list_keys_info *info = (list_keys_info *)handle;
+    list_keys_info *info = static_cast<list_keys_info *>(handle);
     delete info;
   }
 };
@@ -1611,7 +1610,7 @@ public:
   }
 
   int list_keys_next(void *handle, int max, list<string>& keys, bool *truncated) {
-    list_keys_info *info = (list_keys_info *)handle;
+    list_keys_info *info = static_cast<list_keys_info *>(handle);
 
     string no_filter;
 
@@ -1646,7 +1645,7 @@ public:
   }
 
   void list_keys_complete(void *handle) {
-    list_keys_info *info = (list_keys_info *)handle;
+    list_keys_info *info = static_cast<list_keys_info *>(handle);
     delete info;
   }
 
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..ac8c703f5e0 100644
--- a/src/rgw/rgw_metadata.cc
+++ b/src/rgw/rgw_metadata.cc
@@ -104,7 +104,7 @@ void RGWMetadataLog::init_list_entries(int shard_id, utime_t& from_time, utime_t
 }
 
 void RGWMetadataLog::complete_list_entries(void *handle) {
-  LogListCtx *ctx = (LogListCtx *)handle;
+  LogListCtx *ctx = static_cast<LogListCtx *>(handle);
   delete ctx;
 }
 
@@ -112,7 +112,7 @@ int RGWMetadataLog::list_entries(void *handle,
                  int max_entries,
                  list<cls_log_entry>& entries, 
                  bool *truncated) {
-  LogListCtx *ctx = (LogListCtx *)handle;
+  LogListCtx *ctx = static_cast<LogListCtx *>(handle);
 
   if (!max_entries) {
     *truncated = false;
@@ -210,7 +210,7 @@ public:
     return 0;
   }
   virtual int list_keys_next(void *handle, int max, list<string>& keys, bool *truncated)  {
-    iter_data *data = (iter_data *)handle;
+    iter_data *data = static_cast<iter_data *>(handle);
     for (int i = 0; i < max && data->iter != data->sections.end(); ++i, ++(data->iter)) {
       keys.push_back(*data->iter);
     }
@@ -220,7 +220,7 @@ public:
     return 0;
   }
   virtual void list_keys_complete(void *handle) {
-    iter_data *data = (iter_data *)handle;
+    iter_data *data = static_cast<iter_data *>(handle);
 
     delete data;
   }
@@ -451,7 +451,7 @@ int RGWMetadataManager::list_keys_init(string& section, void **handle)
 
 int RGWMetadataManager::list_keys_next(void *handle, int max, list<string>& keys, bool *truncated)
 {
-  list_keys_handle *h = (list_keys_handle *)handle;
+  list_keys_handle *h = static_cast<list_keys_handle *>(handle);
 
   RGWMetadataHandler *handler = h->handler;
 
@@ -461,7 +461,7 @@ int RGWMetadataManager::list_keys_next(void *handle, int max, list<string>& keys
 
 void RGWMetadataManager::list_keys_complete(void *handle)
 {
-  list_keys_handle *h = (list_keys_handle *)handle;
+  list_keys_handle *h = static_cast<list_keys_handle *>(handle);
 
   RGWMetadataHandler *handler = h->handler;
 
@@ -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..45477486ccc 100644
--- a/src/rgw/rgw_op.cc
+++ b/src/rgw/rgw_op.cc
@@ -178,14 +178,12 @@ 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);
 
   if (aiter != bucket_attrs.end()) {
-    ret = decode_policy(cct, aiter->second, policy);
+    int ret = decode_policy(cct, aiter->second, policy);
     if (ret < 0)
       return ret;
   } else {
@@ -203,13 +201,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 +234,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 +242,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 +279,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 +333,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 +360,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 +917,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 +926,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 +971,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 +986,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 +1020,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 +1066,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 +1479,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 +1797,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 +1875,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 +1907,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 +2498,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..e107b90a155 100644
--- a/src/rgw/rgw_op.h
+++ b/src/rgw/rgw_op.h
@@ -50,7 +50,7 @@ public:
   virtual void execute() = 0;
   virtual void send_response() {}
   virtual void complete() { send_response(); }
-  virtual const char *name() = 0;
+  virtual const string name() = 0;
 
   virtual uint32_t op_mask() { return 0; }
 };
@@ -117,7 +117,7 @@ public:
   virtual int get_params() = 0;
   virtual int send_response_data(bufferlist& bl, off_t ofs, off_t len) = 0;
 
-  virtual const char *name() { return "get_obj"; }
+  virtual const string name() { return "get_obj"; }
   virtual uint32_t op_mask() { return RGW_OP_TYPE_READ; }
 };
 
@@ -147,7 +147,7 @@ public:
 
   virtual bool should_get_stats() { return false; }
 
-  virtual const char *name() { return "list_buckets"; }
+  virtual const string name() { return "list_buckets"; }
   virtual uint32_t op_mask() { return RGW_OP_TYPE_READ; }
 };
 
@@ -172,7 +172,7 @@ public:
   void execute();
 
   virtual void send_response() = 0;
-  virtual const char *name() { return "stat_account"; }
+  virtual const string name() { return "stat_account"; }
   virtual uint32_t op_mask() { return RGW_OP_TYPE_READ; }
 };
 
@@ -204,7 +204,7 @@ public:
 
   virtual int get_params() = 0;
   virtual void send_response() = 0;
-  virtual const char *name() { return "list_bucket"; }
+  virtual const string name() { return "list_bucket"; }
   virtual uint32_t op_mask() { return RGW_OP_TYPE_READ; }
 };
 
@@ -215,7 +215,7 @@ public:
   void execute() {}
 
   virtual void send_response() = 0;
-  virtual const char *name() { return "get_bucket_logging"; }
+  virtual const string name() { return "get_bucket_logging"; }
   virtual uint32_t op_mask() { return RGW_OP_TYPE_READ; }
 };
 
@@ -232,7 +232,7 @@ public:
   void execute();
 
   virtual void send_response() = 0;
-  virtual const char *name() { return "stat_bucket"; }
+  virtual const string name() { return "stat_bucket"; }
   virtual uint32_t op_mask() { return RGW_OP_TYPE_READ; }
 };
 
@@ -243,6 +243,7 @@ protected:
   string location_constraint;
   string placement_rule;
   RGWBucketInfo info;
+  obj_version ep_objv;
 
   bufferlist in_data;
 
@@ -257,7 +258,7 @@ public:
   }
   virtual int get_params() { return 0; }
   virtual void send_response() = 0;
-  virtual const char *name() { return "create_bucket"; }
+  virtual const string name() { return "create_bucket"; }
   virtual uint32_t op_mask() { return RGW_OP_TYPE_WRITE; }
 };
 
@@ -274,7 +275,7 @@ public:
   void execute();
 
   virtual void send_response() = 0;
-  virtual const char *name() { return "delete_bucket"; }
+  virtual const string name() { return "delete_bucket"; }
   virtual uint32_t op_mask() { return RGW_OP_TYPE_DELETE; }
 };
 
@@ -318,7 +319,7 @@ public:
   virtual int get_params() = 0;
   virtual int get_data(bufferlist& bl) = 0;
   virtual void send_response() = 0;
-  virtual const char *name() { return "put_obj"; }
+  virtual const string name() { return "put_obj"; }
   virtual uint32_t op_mask() { return RGW_OP_TYPE_WRITE; }
 };
 
@@ -360,7 +361,7 @@ public:
   virtual int get_params() = 0;
   virtual int get_data(bufferlist& bl) = 0;
   virtual void send_response() = 0;
-  virtual const char *name() { return "post_obj"; }
+  virtual const string name() { return "post_obj"; }
   virtual uint32_t op_mask() { return RGW_OP_TYPE_WRITE; }
 };
 
@@ -388,7 +389,7 @@ public:
 
   virtual int get_params() = 0;
   virtual void send_response() = 0;
-  virtual const char *name() { return "put_obj_metadata"; }
+  virtual const string name() { return "put_obj_metadata"; }
   virtual uint32_t op_mask() { return RGW_OP_TYPE_WRITE; }
 };
 
@@ -403,7 +404,7 @@ public:
   void execute();
 
   virtual void send_response() = 0;
-  virtual const char *name() { return "delete_obj"; }
+  virtual const string name() { return "delete_obj"; }
   virtual uint32_t op_mask() { return RGW_OP_TYPE_DELETE; }
 };
 
@@ -471,7 +472,7 @@ public:
   virtual int init_dest_policy() { return 0; }
   virtual int get_params() = 0;
   virtual void send_response() = 0;
-  virtual const char *name() { return "copy_obj"; }
+  virtual const string name() { return "copy_obj"; }
   virtual uint32_t op_mask() { return RGW_OP_TYPE_WRITE; }
 };
 
@@ -487,7 +488,7 @@ public:
   void execute();
 
   virtual void send_response() = 0;
-  virtual const char *name() { return "get_acls"; }
+  virtual const string name() { return "get_acls"; }
   virtual uint32_t op_mask() { return RGW_OP_TYPE_READ; }
 };
 
@@ -513,7 +514,7 @@ public:
   virtual int get_policy_from_state(RGWRados *store, struct req_state *s, stringstream& ss) { return 0; }
   virtual int get_params() = 0;
   virtual void send_response() = 0;
-  virtual const char *name() { return "put_acls"; }
+  virtual const string name() { return "put_acls"; }
   virtual uint32_t op_mask() { return RGW_OP_TYPE_WRITE; }
 };
 
@@ -529,7 +530,7 @@ public:
   void execute();
 
   virtual void send_response() = 0;
-  virtual const char *name() { return "get_cors"; }
+  virtual const string name() { return "get_cors"; }
   virtual uint32_t op_mask() { return RGW_OP_TYPE_READ; }
 };
 
@@ -554,7 +555,7 @@ public:
 
   virtual int get_params() = 0;
   virtual void send_response() = 0;
-  virtual const char *name() { return "put_cors"; }
+  virtual const string name() { return "put_cors"; }
   virtual uint32_t op_mask() { return RGW_OP_TYPE_WRITE; }
 };
 
@@ -569,7 +570,7 @@ public:
   void execute();
 
   virtual void send_response() = 0;
-  virtual const char *name() { return "delete_cors"; }
+  virtual const string name() { return "delete_cors"; }
   virtual uint32_t op_mask() { return RGW_OP_TYPE_WRITE; }
 };
 
@@ -589,7 +590,7 @@ public:
   void execute();
   void get_response_params(string& allowed_hdrs, string& exp_hdrs, unsigned *max_age);
   virtual void send_response() = 0;
-  virtual const char *name() { return "options_cors"; }
+  virtual const string name() { return "options_cors"; }
   virtual uint32_t op_mask() { return RGW_OP_TYPE_READ; }
 };
 
@@ -613,7 +614,7 @@ public:
 
   virtual int get_params() = 0;
   virtual void send_response() = 0;
-  virtual const char *name() { return "init_multipart"; }
+  virtual const string name() { return "init_multipart"; }
   virtual uint32_t op_mask() { return RGW_OP_TYPE_WRITE; }
 };
 
@@ -642,7 +643,7 @@ public:
 
   virtual int get_params() = 0;
   virtual void send_response() = 0;
-  virtual const char *name() { return "complete_multipart"; }
+  virtual const string name() { return "complete_multipart"; }
   virtual uint32_t op_mask() { return RGW_OP_TYPE_WRITE; }
 };
 
@@ -657,7 +658,7 @@ public:
   void execute();
 
   virtual void send_response() = 0;
-  virtual const char *name() { return "abort_multipart"; }
+  virtual const string name() { return "abort_multipart"; }
   virtual uint32_t op_mask() { return RGW_OP_TYPE_DELETE; }
 };
 
@@ -686,7 +687,7 @@ public:
 
   virtual int get_params() = 0;
   virtual void send_response() = 0;
-  virtual const char *name() { return "list_multipart"; }
+  virtual const string name() { return "list_multipart"; }
   virtual uint32_t op_mask() { return RGW_OP_TYPE_READ; }
 };
 
@@ -789,7 +790,7 @@ public:
 
   virtual int get_params() = 0;
   virtual void send_response() = 0;
-  virtual const char *name() { return "list_bucket_multiparts"; }
+  virtual const string name() { return "list_bucket_multiparts"; }
   virtual uint32_t op_mask() { return RGW_OP_TYPE_READ; }
 };
 
@@ -822,7 +823,7 @@ public:
   virtual void begin_response() = 0;
   virtual void send_partial_response(pair<string,int>& result) = 0;
   virtual void end_response() = 0;
-  virtual const char *name() { return "multi_object_delete"; }
+  virtual const string name() { return "multi_object_delete"; }
   virtual uint32_t op_mask() { return RGW_OP_TYPE_DELETE; }
 };
 
diff --git a/src/rgw/rgw_rados.cc b/src/rgw/rgw_rados.cc
index 067a1adbabd..31400dc8402 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;
@@ -5647,7 +5664,7 @@ int RGWStateLog::list_entries(void *handle, int max_entries,
                               list<cls_statelog_entry>& entries,
                               bool *done)
 {
-  list_state *state = (list_state *)handle;
+  list_state *state = static_cast<list_state *>(handle);
 
   librados::IoCtx ioctx;
   int r = open_ioctx(ioctx);
@@ -5695,7 +5712,7 @@ int RGWStateLog::list_entries(void *handle, int max_entries,
 
 void RGWStateLog::finish_list_entries(void *handle)
 {
-  list_state *state = (list_state *)handle;
+  list_state *state = static_cast<list_state *>(handle);
   delete state;
 }
 
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.cc b/src/rgw/rgw_rest.cc
index 623bb0b5e16..0f9e61d1740 100644
--- a/src/rgw/rgw_rest.cc
+++ b/src/rgw/rgw_rest.cc
@@ -835,8 +835,10 @@ int rgw_rest_read_all_input(struct req_state *s, char **pdata, int *plen, int ma
        return -ENOMEM;
     }
     int ret = s->cio->read(data, cl, &len);
-    if (ret < 0)
+    if (ret < 0) {
+      free(data);
       return ret;
+    }
     data[len] = '\0';
   } else if (!s->length) {
     const char *encoding = s->info.env->get("HTTP_TRANSFER_ENCODING");
diff --git a/src/rgw/rgw_rest_bucket.cc b/src/rgw/rgw_rest_bucket.cc
index 80b5b876916..e7068b43c49 100644
--- a/src/rgw/rgw_rest_bucket.cc
+++ b/src/rgw/rgw_rest_bucket.cc
@@ -17,7 +17,7 @@ public:
 
   void execute();
 
-  virtual const char *name() { return "get_bucket_info"; }
+  virtual const string name() { return "get_bucket_info"; }
 };
 
 void RGWOp_Bucket_Info::execute()
@@ -52,7 +52,7 @@ public:
 
   void execute();
 
-  virtual const char *name() { return "get_policy"; }
+  virtual const string name() { return "get_policy"; }
 };
 
 void RGWOp_Get_Policy::execute()
@@ -82,7 +82,7 @@ public:
 
   void execute();
 
-  virtual const char *name() { return "check_bucket_index"; }
+  virtual const string name() { return "check_bucket_index"; }
 };
 
 void RGWOp_Check_Bucket_Index::execute()
@@ -116,7 +116,7 @@ public:
 
   void execute();
 
-  virtual const char *name() { return "link_bucket"; }
+  virtual const string name() { return "link_bucket"; }
 };
 
 void RGWOp_Bucket_Link::execute()
@@ -146,7 +146,7 @@ public:
 
   void execute();
 
-  virtual const char *name() { return "unlink_bucket"; }
+  virtual const string name() { return "unlink_bucket"; }
 };
 
 void RGWOp_Bucket_Unlink::execute()
@@ -176,7 +176,7 @@ public:
 
   void execute();
 
-  virtual const char *name() { return "remove_bucket"; }
+  virtual const string name() { return "remove_bucket"; }
 };
 
 void RGWOp_Bucket_Remove::execute()
@@ -206,7 +206,7 @@ public:
 
   void execute();
 
-  virtual const char *name() { return "remove_object"; }
+  virtual const string name() { return "remove_object"; }
 };
 
 void RGWOp_Object_Remove::execute()
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_config.h b/src/rgw/rgw_rest_config.h
index cb1712ac3d7..2e0408afb3d 100644
--- a/src/rgw/rgw_rest_config.h
+++ b/src/rgw/rgw_rest_config.h
@@ -25,7 +25,7 @@ public:
   }
   void execute();
   virtual void send_response();
-  virtual const char *name() {
+  virtual const string name() {
     return "get_region_map";
   }
 };
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_log.h b/src/rgw/rgw_rest_log.h
index 38c6b5fb4ab..2d60e289b84 100644
--- a/src/rgw/rgw_rest_log.h
+++ b/src/rgw/rgw_rest_log.h
@@ -32,7 +32,7 @@ public:
   virtual void send_response(list<rgw_bi_log_entry>& entries, string& marker);
   virtual void send_response_end();
   void execute();
-  virtual const char *name() {
+  virtual const string name() {
     return "list_bucket_index_log";
   }
 };
@@ -53,7 +53,7 @@ public:
   }
   virtual void send_response();
   void execute();
-  virtual const char *name() {
+  virtual const string name() {
     return "bucket_index_log_info";
   }
 };
@@ -67,7 +67,7 @@ public:
     return caps.check_cap("bilog", RGW_CAP_WRITE);
   }
   void execute();
-  virtual const char *name() {
+  virtual const string name() {
     return "trim_bucket_index_log";
   }
 };
@@ -87,7 +87,7 @@ public:
   }
   void execute();
   virtual void send_response();
-  virtual const char *name() {
+  virtual const string name() {
     return "list_metadata_log";
   }
 };
@@ -107,7 +107,7 @@ public:
   }
   void execute();
   virtual void send_response();
-  virtual const char *name() {
+  virtual const string name() {
     return "get_metadata_log_info";
   }
 };
@@ -126,7 +126,7 @@ public:
   }
   void execute();
   virtual void send_response();
-  virtual const char *name() {
+  virtual const string name() {
     return "get_metadata_log_shard_info";
   }
 };
@@ -140,7 +140,7 @@ public:
     return caps.check_cap("mdlog", RGW_CAP_WRITE);
   }
   void execute();
-  virtual const char *name() {
+  virtual const string name() {
     return "lock_mdlog_object";
   }
 };
@@ -154,7 +154,7 @@ public:
     return caps.check_cap("mdlog", RGW_CAP_WRITE);
   }
   void execute();
-  virtual const char *name() {
+  virtual const string name() {
     return "unlock_mdlog_object";
   }
 };
@@ -168,7 +168,7 @@ public:
     return caps.check_cap("mdlog", RGW_CAP_WRITE);
   }
   void execute();
-  virtual const char *name() {
+  virtual const string name() {
     return "trim_metadata_log";
   }
 };
@@ -188,7 +188,7 @@ public:
   }
   void execute();
   virtual void send_response();
-  virtual const char *name() {
+  virtual const string name() {
     return "list_data_changes_log";
   }
 };
@@ -208,7 +208,7 @@ public:
   }
   void execute();
   virtual void send_response();
-  virtual const char *name() {
+  virtual const string name() {
     return "get_data_changes_log_info";
   }
 };
@@ -227,7 +227,7 @@ public:
   }
   void execute();
   virtual void send_response();
-  virtual const char *name() {
+  virtual const string name() {
     return "get_data_changes_log_shard_info";
   }
 };
@@ -241,7 +241,7 @@ public:
     return caps.check_cap("datalog", RGW_CAP_WRITE);
   }
   void execute();
-  virtual const char *name() {
+  virtual const string name() {
     return "lock_datalog_object";
   }
 };
@@ -255,7 +255,7 @@ public:
     return caps.check_cap("datalog", RGW_CAP_WRITE);
   }
   void execute();
-  virtual const char *name() {
+  virtual const string name() {
     return "unlock_datalog_object";
   }
 };
@@ -269,7 +269,7 @@ public:
     return caps.check_cap("datalog", RGW_CAP_WRITE);
   }
   void execute();
-  virtual const char *name() {
+  virtual const string name() {
     return "trim_data_changes_log";
   }
 };
diff --git a/src/rgw/rgw_rest_metadata.cc b/src/rgw/rgw_rest_metadata.cc
index 69f8a5ccbc4..35ec0ab9b04 100644
--- a/src/rgw/rgw_rest_metadata.cc
+++ b/src/rgw/rgw_rest_metadata.cc
@@ -23,7 +23,7 @@
 
 #define dout_subsys ceph_subsys_rgw
 
-const char *RGWOp_Metadata_Get::name() {
+const string RGWOp_Metadata_Get::name() {
   return "get_metadata";
 }
 
@@ -62,7 +62,7 @@ void RGWOp_Metadata_Get::execute() {
   http_ret = 0;
 }
 
-const char *RGWOp_Metadata_List::name() {
+const string RGWOp_Metadata_List::name() {
   return "list_metadata";
 }
 
diff --git a/src/rgw/rgw_rest_metadata.h b/src/rgw/rgw_rest_metadata.h
index 85993d08d58..59d7c5f7045 100644
--- a/src/rgw/rgw_rest_metadata.h
+++ b/src/rgw/rgw_rest_metadata.h
@@ -23,7 +23,7 @@ public:
     return caps.check_cap("metadata", RGW_CAP_READ);
   }
   void execute();
-  virtual const char *name();
+  virtual const string name();
 };
 
 class RGWOp_Metadata_Get : public RGWRESTOp {
@@ -35,7 +35,7 @@ public:
     return caps.check_cap("metadata", RGW_CAP_READ);
   }
   void execute();
-  virtual const char *name();
+  virtual const string name();
 };
 
 class RGWOp_Metadata_Put : public RGWRESTOp {
@@ -48,7 +48,7 @@ public:
     return caps.check_cap("metadata", RGW_CAP_WRITE);
   }
   void execute();
-  virtual const char *name() { return "set_metadata"; }
+  virtual const string name() { return "set_metadata"; }
 };
 
 class RGWOp_Metadata_Delete : public RGWRESTOp {
@@ -60,7 +60,7 @@ public:
     return caps.check_cap("metadata", RGW_CAP_WRITE);
   }
   void execute();
-  virtual const char *name() { return "remove_metadata"; }
+  virtual const string name() { return "remove_metadata"; }
 };
 
 class RGWOp_Metadata_Lock : public RGWRESTOp {
@@ -72,7 +72,7 @@ public:
     return caps.check_cap("metadata", RGW_CAP_WRITE);
   }
   void execute();
-  virtual const char *name() {
+  virtual const string name() {
     return "lock_metadata_object";
   }
 };
@@ -86,7 +86,7 @@ public:
     return caps.check_cap("metadata", RGW_CAP_WRITE);
   }
   void execute();
-  virtual const char *name() {
+  virtual const string name() {
     return "unlock_metadata_object";
   }
 };
diff --git a/src/rgw/rgw_rest_opstate.h b/src/rgw/rgw_rest_opstate.h
index 8f6a9675a68..de13dde6966 100644
--- a/src/rgw/rgw_rest_opstate.h
+++ b/src/rgw/rgw_rest_opstate.h
@@ -30,7 +30,7 @@ public:
   virtual void send_response();
   virtual void send_response(list<cls_statelog_entry> entries);
   virtual void send_response_end();
-  virtual const char *name() {
+  virtual const string name() {
     return "opstate_list";
   }
 };
@@ -44,7 +44,7 @@ public:
     return caps.check_cap("opstate", RGW_CAP_WRITE);
   }
   void execute();
-  virtual const char *name() {
+  virtual const string name() {
     return "set_opstate";
   }
 };
@@ -58,7 +58,7 @@ public:
     return caps.check_cap("opstate", RGW_CAP_WRITE);
   }
   void execute();
-  virtual const char *name() {
+  virtual const string name() {
     return "renew_opstate";
   }
 };
@@ -72,7 +72,7 @@ public:
     return caps.check_cap("opstate", RGW_CAP_WRITE);
   }
   void execute();
-  virtual const char *name() {
+  virtual const string name() {
     return "delete_opstate";
   }
 };
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_replica_log.h b/src/rgw/rgw_rest_replica_log.h
index 91e3d614062..c879150cc07 100644
--- a/src/rgw/rgw_rest_replica_log.h
+++ b/src/rgw/rgw_rest_replica_log.h
@@ -32,11 +32,11 @@ public:
   }
   void execute();
   virtual void send_response();
-  virtual const char *name() {
+  virtual const string name() {
     string s = "replica";
     s.append(obj_type);
     s.append("_getbounds");
-    return s.c_str();
+    return s;
   }
 };
 
@@ -52,11 +52,11 @@ public:
     return caps.check_cap(obj_type.c_str(), RGW_CAP_WRITE);
   }
   void execute();
-  virtual const char *name() {
+  virtual const string name() {
     string s = "replica";
     s.append(obj_type);
     s.append("_updatebounds");
-    return s.c_str();
+    return s;
   }
 };
 
@@ -72,11 +72,11 @@ public:
     return caps.check_cap(obj_type.c_str(), RGW_CAP_WRITE);
   }
   void execute();
-  virtual const char *name() {
+  virtual const string name() {
     string s = "replica";
     s.append(obj_type);
     s.append("_deletebound");
-    return s.c_str();
+    return s;
   }
 };
 
@@ -94,7 +94,7 @@ public:
   }
   void execute();
   virtual void send_response();
-  virtual const char *name() {
+  virtual const string name() {
     return "replicabilog_getbounds";
   }
 };
@@ -108,7 +108,7 @@ public:
     return caps.check_cap("bilog", RGW_CAP_WRITE);
   }
   void execute();
-  virtual const char *name() {
+  virtual const string name() {
     return "replicabilog_updatebounds";
   }
 };
@@ -122,7 +122,7 @@ public:
     return caps.check_cap("bilog", RGW_CAP_WRITE);
   }
   void execute();
-  virtual const char *name() {
+  virtual const string name() {
     return "replicabilog_deletebound";
   }
 };
diff --git a/src/rgw/rgw_rest_s3.cc b/src/rgw/rgw_rest_s3.cc
index 6e482e8a251..66f6652ec6a 100644
--- a/src/rgw/rgw_rest_s3.cc
+++ b/src/rgw/rgw_rest_s3.cc
@@ -409,12 +409,13 @@ int RGWCreateBucket_ObjStore_S3::get_params()
 
     bool success = parser.parse(data, len, 1);
     ldout(s->cct, 20) << "create bucket input data=" << data << dendl;
-    free(data);
 
     if (!success) {
       ldout(s->cct, 0) << "failed to parse input: " << data << dendl;
+      free(data);
       return -EINVAL;
     }
+    free(data);
 
     if (!parser.get_location_constraint(location_constraint)) {
       ldout(s->cct, 0) << "provided input did not specify location constraint correctly" << dendl;
@@ -449,6 +450,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();
diff --git a/src/rgw/rgw_rest_usage.cc b/src/rgw/rgw_rest_usage.cc
index 769e167019a..1124d2b298b 100644
--- a/src/rgw/rgw_rest_usage.cc
+++ b/src/rgw/rgw_rest_usage.cc
@@ -16,7 +16,7 @@ public:
   }
   void execute();
 
-  virtual const char *name() { return "get_usage"; }
+  virtual const string name() { return "get_usage"; }
 };
 
 void RGWOp_Usage_Get::execute() {
@@ -58,7 +58,7 @@ public:
   }
   void execute();
 
-  virtual const char *name() { return "trim_usage"; }
+  virtual const string name() { return "trim_usage"; }
 };
 
 void RGWOp_Usage_Delete::execute() {
diff --git a/src/rgw/rgw_rest_user.cc b/src/rgw/rgw_rest_user.cc
index ac0d794846c..3d08e403229 100644
--- a/src/rgw/rgw_rest_user.cc
+++ b/src/rgw/rgw_rest_user.cc
@@ -17,7 +17,7 @@ public:
 
   void execute();
 
-  virtual const char *name() { return "get_user_info"; }
+  virtual const string name() { return "get_user_info"; }
 };
 
 void RGWOp_User_Info::execute()
@@ -44,7 +44,7 @@ public:
 
   void execute();
 
-  virtual const char *name() { return "create_user"; }
+  virtual const string name() { return "create_user"; }
 };
 
 void RGWOp_User_Create::execute()
@@ -138,7 +138,7 @@ public:
 
   void execute();
 
-  virtual const char *name() { return "modify_user"; }
+  virtual const string name() { return "modify_user"; }
 };
 
 void RGWOp_User_Modify::execute()
@@ -232,7 +232,7 @@ public:
 
   void execute();
 
-  virtual const char *name() { return "remove_user"; }
+  virtual const string name() { return "remove_user"; }
 };
 
 void RGWOp_User_Remove::execute()
@@ -265,7 +265,7 @@ public:
 
   void execute();
 
-  virtual const char *name() { return "create_subuser"; }
+  virtual const string name() { return "create_subuser"; }
 };
 
 void RGWOp_Subuser_Create::execute()
@@ -334,7 +334,7 @@ public:
 
   void execute();
 
-  virtual const char *name() { return "modify_subuser"; }
+  virtual const string name() { return "modify_subuser"; }
 };
 
 void RGWOp_Subuser_Modify::execute()
@@ -399,7 +399,7 @@ public:
 
   void execute();
 
-  virtual const char *name() { return "remove_subuser"; }
+  virtual const string name() { return "remove_subuser"; }
 };
 
 void RGWOp_Subuser_Remove::execute()
@@ -438,7 +438,7 @@ public:
 
   void execute();
 
-  virtual const char *name() { return "create_access_key"; }
+  virtual const string name() { return "create_access_key"; }
 };
 
 void RGWOp_Key_Create::execute()
@@ -500,7 +500,7 @@ public:
 
   void execute();
 
-  virtual const char *name() { return "remove_access_key"; }
+  virtual const string name() { return "remove_access_key"; }
 };
 
 void RGWOp_Key_Remove::execute()
@@ -552,7 +552,7 @@ public:
 
   void execute();
 
-  virtual const char *name() { return "add_user_caps"; }
+  virtual const string name() { return "add_user_caps"; }
 };
 
 void RGWOp_Caps_Add::execute()
@@ -586,7 +586,7 @@ public:
 
   void execute();
 
-  virtual const char *name() { return "remove_user_caps"; }
+  virtual const string name() { return "remove_user_caps"; }
 };
 
 void RGWOp_Caps_Remove::execute()
diff --git a/src/rgw/rgw_swift_auth.h b/src/rgw/rgw_swift_auth.h
index 8a58b476496..670a339210d 100644
--- a/src/rgw/rgw_swift_auth.h
+++ b/src/rgw/rgw_swift_auth.h
@@ -15,7 +15,7 @@ public:
 
   int verify_permission() { return 0; }
   void execute();
-  virtual const char *name() { return "swift_auth_get"; }
+  virtual const string name() { return "swift_auth_get"; }
 };
 
 class RGWHandler_SWIFT_Auth : public RGWHandler {
diff --git a/src/test/cls_version/test_cls_version.cc b/src/test/cls_version/test_cls_version.cc
index 6392424644a..caa0a36cd74 100644
--- a/src/test/cls_version/test_cls_version.cc
+++ b/src/test/cls_version/test_cls_version.cc
@@ -76,7 +76,7 @@ TEST(cls_rgw, test_version_inc_read)
   ASSERT_EQ(ver2.ver, ver3.ver);
   ASSERT_EQ(1, (long long)ver2.compare(&ver3));
 
-  delete op;
+  delete rop;
 }