rgw: metadata lock/unlock implemented with test cases

metadata lock/unlock implemented with test cases
split Post operations to Lock and Unlock operations, and
checking for WRITE permission only for lock and unlock

Signed-off-by: Babu Shanmugam <anbu@enovance.com>

Conflicts:
	src/rgw/rgw_rest_log.cc
	src/rgw/rgw_rest_log.h

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

13 files changed, 409 insertions, 151 deletions

diff --git a/src/cls/log/cls_log.cc b/src/cls/log/cls_log.cc
index 73169edb924..76125e1a1db 100644
--- a/src/cls/log/cls_log.cc
+++ b/src/cls/log/cls_log.cc
@@ -110,7 +110,7 @@ static int cls_log_list(cls_method_context_t hctx, bufferlist *in, bufferlist *o
   } else {
     from_index = op.marker;
   }
-  bool use_time_boundary = (!op.from_time.is_zero() && !op.to_time.is_zero() && (op.to_time >= op.from_time));
+  bool use_time_boundary = (!op.from_time.is_zero() && (op.to_time >= op.from_time));
 
   if (use_time_boundary)
     get_index_time_prefix(op.to_time, to_index);
diff --git a/src/rgw/rgw_bucket.cc b/src/rgw/rgw_bucket.cc
index b22ceef97e0..a493bf2fc50 100644
--- a/src/rgw/rgw_bucket.cc
+++ b/src/rgw/rgw_bucket.cc
@@ -1197,11 +1197,14 @@ int RGWDataChangesLog::list_entries(utime_t& start_time, utime_t& end_time, int
 
 int RGWDataChangesLog::trim_entries(int shard_id, utime_t& start_time, utime_t& end_time)
 {
-  int ret = store->time_log_trim(oids[shard_id], start_time, end_time);
-  if ((ret < 0) && (ret != -ENOENT))
-    return ret;
+  int ret;
 
-  return 0;
+  ret = store->time_log_trim(oids[shard_id], start_time, end_time);
+
+  if (ret == -ENOENT)
+    ret = 0;
+
+  return ret;
 }
 
 int RGWDataChangesLog::trim_entries(utime_t& start_time, utime_t& end_time)
diff --git a/src/rgw/rgw_bucket.h b/src/rgw/rgw_bucket.h
index f1fbd184df0..cec6192b75e 100644
--- a/src/rgw/rgw_bucket.h
+++ b/src/rgw/rgw_bucket.h
@@ -344,10 +344,10 @@ public:
   int trim_entries(int shard_id, utime_t& start_time, utime_t& end_time);
   int trim_entries(utime_t& start_time, utime_t& end_time);
   int lock_exclusive(int shard_id, utime_t& duration, string& owner_id) {
-    return store->log_lock_exclusive(oids[shard_id], duration, owner_id);
+    return store->lock_exclusive(store->zone.log_pool, oids[shard_id], duration, owner_id);
   }
   int unlock(int shard_id, string& owner_id) {
-    return store->log_unlock(oids[shard_id], owner_id);
+    return store->unlock(store->zone.log_pool, oids[shard_id], owner_id);
   }
   struct LogMarker {
     int shard;
diff --git a/src/rgw/rgw_metadata.cc b/src/rgw/rgw_metadata.cc
index 6c571bd8cfe..2bf0f2e885f 100644
--- a/src/rgw/rgw_metadata.cc
+++ b/src/rgw/rgw_metadata.cc
@@ -133,24 +133,21 @@ int RGWMetadataLog::trim(int shard_id, utime_t& from_time, utime_t& end_time)
   if (ret == -ENOENT)
     ret = 0;
 
-  if (ret < 0)
-    return ret;
-
-  return 0;
+  return ret;
 }
   
 int RGWMetadataLog::lock_exclusive(int shard_id, utime_t& duration, string& owner_id) {
   string oid;
   get_shard_oid(shard_id, oid);
 
-  return store->log_lock_exclusive(oid, duration, owner_id);
+  return store->lock_exclusive(store->zone.log_pool, oid, duration, owner_id);
 }
 
 int RGWMetadataLog::unlock(int shard_id, string& owner_id) {
   string oid;
   get_shard_oid(shard_id, oid);
 
-  return store->log_unlock(oid, owner_id);
+  return store->unlock(store->zone.log_pool, oid, owner_id);
 }
 
 obj_version& RGWMetadataObject::get_version()
@@ -353,6 +350,38 @@ int RGWMetadataManager::remove(string& metadata_key)
   return handler->remove(store, entry, objv_tracker);
 }
 
+int RGWMetadataManager::lock_exclusive(string& metadata_key, utime_t duration, string& owner_id) {
+  RGWMetadataHandler *handler;
+  string entry;
+
+  int ret = find_handler(metadata_key, &handler, entry);
+  if (ret < 0) 
+    return ret;
+
+  rgw_bucket pool;
+  string oid;
+
+  handler->get_pool_and_oid(store, entry, pool, oid);
+
+  return store->lock_exclusive(pool, oid, duration, owner_id);  
+}
+
+int RGWMetadataManager::unlock(string& metadata_key, string& owner_id) {
+  librados::IoCtx io_ctx;
+  RGWMetadataHandler *handler;
+  string entry;
+
+  int ret = find_handler(metadata_key, &handler, entry);
+  if (ret < 0) 
+    return ret;
+
+  rgw_bucket pool;
+  string oid;
+
+  handler->get_pool_and_oid(store, entry, pool, oid);
+
+  return store->unlock(pool, oid, owner_id);  
+}
 
 struct list_keys_handle {
   void *handle;
diff --git a/src/rgw/rgw_metadata.h b/src/rgw/rgw_metadata.h
index 299b3ff35b3..80c9f41ae6a 100644
--- a/src/rgw/rgw_metadata.h
+++ b/src/rgw/rgw_metadata.h
@@ -145,6 +145,8 @@ public:
   void dump_log_entry(cls_log_entry& entry, Formatter *f);
 
   void get_sections(list<string>& sections);
+  int lock_exclusive(string& metadata_key, utime_t duration, string& owner_id);
+  int unlock(string& metadata_key, string& owner_id);
 
   RGWMetadataLog *get_log() { return md_log; }
 };
diff --git a/src/rgw/rgw_rados.cc b/src/rgw/rgw_rados.cc
index 7eff15f3ade..e634b617d80 100644
--- a/src/rgw/rgw_rados.cc
+++ b/src/rgw/rgw_rados.cc
@@ -58,7 +58,6 @@ static string region_info_oid_prefix = "region_info.";
 
 static string default_region_info_oid = "default.region";
 static string region_map_oid = "region_map";
-static string rgw_log_lock_name = "rgw_process";
 
 static RGWObjCategory main_category = RGW_OBJ_CATEGORY_MAIN;
 
@@ -1217,28 +1216,37 @@ int RGWRados::time_log_trim(const string& oid, utime_t& start_time, utime_t& end
   return cls_log_trim(io_ctx, oid, start_time, end_time);
 }
 
-int RGWRados::log_lock_exclusive(const string& oid, utime_t& duration, string& owner_id) {
+
+int RGWRados::lock_exclusive(rgw_bucket& pool, const string& oid, utime_t& duration, string& owner_id) {
   librados::IoCtx io_ctx;
 
-  const char *log_pool = zone.log_pool.name.c_str();
-  int r = rados->ioctx_create(log_pool, io_ctx);
+  const char *pool_name = pool.name.c_str();
+  
+  int r = rados->ioctx_create(pool_name, io_ctx);
   if (r < 0)
     return r;
-  rados::cls::lock::Lock l(rgw_log_lock_name);
+  
+  string lock_name = RGW_INDEX_LOCK_NAME;
+  rados::cls::lock::Lock l(lock_name);
   l.set_duration(duration);
   l.set_cookie(owner_id);
+  
   return l.lock_exclusive(&io_ctx, oid);
 }
 
-int RGWRados::log_unlock(const string& oid, string& owner_id) {
+int RGWRados::unlock(rgw_bucket& pool, const string& oid, string& owner_id) {
   librados::IoCtx io_ctx;
 
-  const char *log_pool = zone.log_pool.name.c_str();
-  int r = rados->ioctx_create(log_pool, io_ctx);
+  const char *pool_name = pool.name.c_str();
+
+  int r = rados->ioctx_create(pool_name, io_ctx);
   if (r < 0)
     return r;
-  rados::cls::lock::Lock l(rgw_log_lock_name);
+  
+  string lock_name = RGW_INDEX_LOCK_NAME;
+  rados::cls::lock::Lock l(lock_name);
   l.set_cookie(owner_id);
+  
   return l.unlock(&io_ctx, oid);
 }
 
diff --git a/src/rgw/rgw_rados.h b/src/rgw/rgw_rados.h
index 9a723b74eec..9417c24d5fd 100644
--- a/src/rgw/rgw_rados.h
+++ b/src/rgw/rgw_rados.h
@@ -24,6 +24,7 @@ class RGWGC;
 
 #define RGW_OBJ_NS_MULTIPART "multipart"
 #define RGW_OBJ_NS_SHADOW    "shadow"
+#define RGW_INDEX_LOCK_NAME  "rgw_process"
 
 static inline void prepend_bucket_marker(rgw_bucket& bucket, string& orig_oid, string& oid)
 {
@@ -980,8 +981,8 @@ public:
   int time_log_list(const string& oid, utime_t& start_time, utime_t& end_time,
                     int max_entries, list<cls_log_entry>& entries, string& marker, bool *truncated);
   int time_log_trim(const string& oid, utime_t& start_time, utime_t& end_time);
-  int log_lock_exclusive(const string& oid, utime_t& duration, string& owner_id);
-  int log_unlock(const string& oid, string& owner_id);
+  int lock_exclusive(rgw_bucket& pool, const string& oid, utime_t& duration, string& owner_id);
+  int unlock(rgw_bucket& pool, const string& oid, string& owner_id);
 
   /// clean up/process any temporary objects older than given date[/time]
   int remove_temp_objects(string date, string time);
diff --git a/src/rgw/rgw_rest_log.cc b/src/rgw/rgw_rest_log.cc
index 886dbbe979e..b37c88a3ad4 100644
--- a/src/rgw/rgw_rest_log.cc
+++ b/src/rgw/rgw_rest_log.cc
@@ -143,37 +143,54 @@ void RGWOp_MDLog_Delete::execute() {
   http_ret = meta_log->trim(shard_id, ut_st, ut_et);
 }
 
-int RGWOp_MDLog_Post::check_caps(RGWUserCaps& caps) {
-  if (caps.check_cap("mdlog", RGW_CAP_READ) &&
-      caps.check_cap("mdlog", RGW_CAP_WRITE)) {
-    return -EPERM;
+void RGWOp_MDLog_Lock::execute() {
+  string shard_id_str, duration_str, lock_id;
+  int shard_id;
+
+  http_ret = 0;
+
+  shard_id_str = s->info.args.get("id");
+  duration_str = s->info.args.get("length");
+  lock_id      = s->info.args.get("lock_id");
+
+  if (shard_id_str.empty() ||
+      (duration_str.empty()) ||
+      lock_id.empty()) {
+    dout(5) << "Error invalid parameter list" << dendl;
+    http_ret = -EINVAL;
+    return;
   }
-  return 0;
-}
 
-const char *RGWOp_MDLog_Post::name() {
-  int pt = get_post_type();
-  if (pt == MDLOG_POST_LOCK) 
-    return "lock mdlog object";
-  else if (pt == MDLOG_POST_UNLOCK)
-    return "unlock mdlog object";
-  return NULL;
+  string err;
+  shard_id = strict_strtol(shard_id_str.c_str(), 10, &err);
+  if (!err.empty()) {
+    dout(5) << "Error parsing shard_id param " << shard_id_str << dendl;
+    http_ret = -EINVAL;
+    return;
+  }
+
+  RGWMetadataLog *meta_log = store->meta_mgr->get_log();
+  int dur;
+  dur = strict_strtol(duration_str.c_str(), 10, &err);
+  if (!err.empty() || dur <= 0) {
+    dout(5) << "invalid length param " << duration_str << dendl;
+    http_ret = -EINVAL;
+    return;
+  }
+  utime_t time(dur, 0);
+  http_ret = meta_log->lock_exclusive(shard_id, time, lock_id);
 }
 
-void RGWOp_MDLog_Post::execute() {
-  string shard_id_str, duration_str, lock_id;
+void RGWOp_MDLog_Unlock::execute() {
+  string shard_id_str, lock_id;
   int shard_id;
-  int pt = get_post_type();
 
   http_ret = 0;
 
   shard_id_str = s->info.args.get("id");
-  if (pt == MDLOG_POST_LOCK)
-    duration_str = s->info.args.get("length");
   lock_id      = s->info.args.get("lock_id");
 
   if (shard_id_str.empty() ||
-      (pt == MDLOG_POST_LOCK && duration_str.empty()) ||
       lock_id.empty()) {
     dout(5) << "Error invalid parameter list" << dendl;
     http_ret = -EINVAL;
@@ -189,20 +206,7 @@ void RGWOp_MDLog_Post::execute() {
   }
 
   RGWMetadataLog *meta_log = store->meta_mgr->get_log();
-  if (pt == MDLOG_POST_LOCK) {
-    int dur;
-    dur = strict_strtol(duration_str.c_str(), 10, &err);
-    if (!err.empty() || dur <= 0) {
-      dout(5) << "invalid length param " << duration_str << dendl;
-      http_ret = -EINVAL;
-      return;
-    }
-    utime_t time(dur, 0);
-    http_ret = meta_log->lock_exclusive(shard_id, time, lock_id);
-  } else if (pt == MDLOG_POST_UNLOCK) {
-    http_ret = meta_log->unlock(shard_id, lock_id);
-  } else 
-    http_ret = -EINVAL;
+  http_ret = meta_log->unlock(shard_id, lock_id);
 }
 
 void RGWOp_BILog_List::execute() {
@@ -310,10 +314,10 @@ void RGWOp_BILog_Delete::execute() {
 }
 
 void RGWOp_DATALog_List::execute() {
-  string   shard = s->args.get("id");
+  string   shard = s->info.args.get("id");
 
-  string   st = s->args.get("start-time"),
-           et = s->args.get("end-time"),
+  string   st = s->info.args.get("start-time"),
+           et = s->info.args.get("end-time"),
            err;
   utime_t  ut_st, 
            ut_et;
@@ -380,37 +384,53 @@ void RGWOp_DATALog_GetShardsInfo::send_response() {
   flusher.flush();
 }
 
-int RGWOp_DATALog_Post::check_caps(RGWUserCaps& caps) {
-  if (caps.check_cap("datalog", RGW_CAP_READ) &&
-      caps.check_cap("datalog", RGW_CAP_WRITE)) {
-    return -EPERM;
+void RGWOp_DATALog_Lock::execute() {
+  string shard_id_str, duration_str, lock_id;
+  int shard_id;
+
+  http_ret = 0;
+
+  shard_id_str = s->info.args.get("id");
+  duration_str = s->info.args.get("length");
+  lock_id      = s->info.args.get("lock_id");
+
+  if (shard_id_str.empty() ||
+      (duration_str.empty()) ||
+      lock_id.empty()) {
+    dout(5) << "Error invalid parameter list" << dendl;
+    http_ret = -EINVAL;
+    return;
   }
-  return 0;
-}
 
-const char *RGWOp_DATALog_Post::name() {
-  int pt = get_post_type();
-  if (pt == DATALOG_POST_LOCK) 
-    return "lock datalog object";
-  else if (pt == DATALOG_POST_UNLOCK)
-    return "unlock datalog object";
-  return NULL;
+  string err;
+  shard_id = strict_strtol(shard_id_str.c_str(), 10, &err);
+  if (!err.empty()) {
+    dout(5) << "Error parsing shard_id param " << shard_id_str << dendl;
+    http_ret = -EINVAL;
+    return;
+  }
+
+  int dur;
+  dur = strict_strtol(duration_str.c_str(), 10, &err);
+  if (!err.empty() || dur <= 0) {
+    dout(5) << "invalid length param " << duration_str << dendl;
+    http_ret = -EINVAL;
+    return;
+  }
+  utime_t time(dur, 0);
+  http_ret = store->data_log->lock_exclusive(shard_id, time, lock_id);
 }
 
-void RGWOp_DATALog_Post::execute() {
-  string shard_id_str, duration_str, lock_id;
+void RGWOp_DATALog_Unlock::execute() {
+  string shard_id_str, lock_id;
   int shard_id;
-  int pt = get_post_type();
 
   http_ret = 0;
 
-  shard_id_str = s->args.get("id");
-  if (pt == DATALOG_POST_LOCK)
-    duration_str = s->args.get("length");
-  lock_id      = s->args.get("lock_id");
+  shard_id_str = s->info.args.get("id");
+  lock_id      = s->info.args.get("lock_id");
 
   if (shard_id_str.empty() ||
-      (pt == DATALOG_POST_LOCK && duration_str.empty()) ||
       lock_id.empty()) {
     dout(5) << "Error invalid parameter list" << dendl;
     http_ret = -EINVAL;
@@ -425,26 +445,13 @@ void RGWOp_DATALog_Post::execute() {
     return;
   }
 
-  if (pt == DATALOG_POST_LOCK) {
-    int dur;
-    dur = strict_strtol(duration_str.c_str(), 10, &err);
-    if (!err.empty() || dur <= 0) {
-      dout(5) << "invalid length param " << duration_str << dendl;
-      http_ret = -EINVAL;
-      return;
-    }
-    utime_t time(dur, 0);
-    http_ret = store->data_log->lock_exclusive(shard_id, time, lock_id);
-  } else if (pt == DATALOG_POST_UNLOCK) {
-    http_ret = store->data_log->unlock(shard_id, lock_id);
-  } else 
-    http_ret = -EINVAL;
+  http_ret = store->data_log->unlock(shard_id, lock_id);
 }
 
 void RGWOp_DATALog_Delete::execute() {
-  string   st = s->args.get("start-time"),
-           et = s->args.get("end-time"),
-           shard = s->args.get("id"),
+  string   st = s->info.args.get("start-time"),
+           et = s->info.args.get("end-time"),
+           shard = s->info.args.get("id"),
            err;
   utime_t  ut_st, 
            ut_et;
@@ -493,7 +500,7 @@ RGWOp *RGWHandler_Log::op_get() {
   } else if (type.compare("bucket-index") == 0) {
     return new RGWOp_BILog_List;
   } else if (type.compare("data") == 0) {
-    if (s->args.exists("id")) {
+    if (s->info.args.exists("id")) {
       return new RGWOp_DATALog_List;
     } else {
       return new RGWOp_DATALog_GetShardsInfo;
@@ -527,10 +534,17 @@ RGWOp *RGWHandler_Log::op_post() {
     return NULL;
   }
 
-  if (type.compare("metadata") == 0)
-    return new RGWOp_MDLog_Post;
-  else if (type.compare("data") == 0)
-    return new RGWOp_DATALog_Post;
+  if (type.compare("metadata") == 0) {
+    if (s->info.args.exists("lock"))
+      return new RGWOp_MDLog_Lock;
+    else if (s->info.args.exists("unlock"))
+      return new RGWOp_MDLog_Unlock;
+  } else if (type.compare("data") == 0) {
+    if (s->info.args.exists("lock"))
+      return new RGWOp_DATALog_Lock;
+    else if (s->info.args.exists("unlock"))
+      return new RGWOp_DATALog_Unlock;
+  }
   return NULL;
 }
 
diff --git a/src/rgw/rgw_rest_log.h b/src/rgw/rgw_rest_log.h
index e5b94954549..58207c59c32 100644
--- a/src/rgw/rgw_rest_log.h
+++ b/src/rgw/rgw_rest_log.h
@@ -90,29 +90,32 @@ public:
   }
 };
 
-class RGWOp_MDLog_Post : public RGWRESTOp {
-  enum {
-    MDLOG_POST_INVALID = 0,
-    MDLOG_POST_LOCK,
-    MDLOG_POST_UNLOCK
-  };
-  int get_post_type() {
-    bool exists;
-    s->info.args.get("lock", &exists);
-    if (exists) 
-      return MDLOG_POST_LOCK;
-    s->info.args.get("unlock", &exists);
-    if (exists)
-      return MDLOG_POST_UNLOCK;
-    return MDLOG_POST_INVALID;
+class RGWOp_MDLog_Lock : public RGWRESTOp {
+public:
+  RGWOp_MDLog_Lock() {}
+  ~RGWOp_MDLog_Lock() {}
+
+  int check_caps(RGWUserCaps& caps) {
+    return caps.check_cap("mdlog", RGW_CAP_WRITE);
+  }
+  void execute();
+  virtual const char *name() {
+    return "lock_mdlog_object";
   }
+};
+
+class RGWOp_MDLog_Unlock : public RGWRESTOp {
 public:
-  RGWOp_MDLog_Post() {}
-  ~RGWOp_MDLog_Post() {}
+  RGWOp_MDLog_Unlock() {}
+  ~RGWOp_MDLog_Unlock() {}
 
-  int check_caps(RGWUserCaps& caps);
+  int check_caps(RGWUserCaps& caps) {
+    return caps.check_cap("mdlog", RGW_CAP_WRITE);
+  }
   void execute();
-  virtual const char *name();
+  virtual const char *name() {
+    return "unlock_mdlog_object";
+  }
 };
 
 class RGWOp_MDLog_Delete : public RGWRESTOp {
@@ -169,29 +172,32 @@ public:
   }
 };
 
-class RGWOp_DATALog_Post : public RGWRESTOp {
-  enum {
-    DATALOG_POST_INVALID = 0,
-    DATALOG_POST_LOCK,
-    DATALOG_POST_UNLOCK
-  };
-  int get_post_type() {
-    bool exists;
-    s->args.get("lock", &exists);
-    if (exists) 
-      return DATALOG_POST_LOCK;
-    s->args.get("unlock", &exists);
-    if (exists)
-      return DATALOG_POST_UNLOCK;
-    return DATALOG_POST_INVALID;
+class RGWOp_DATALog_Lock : public RGWRESTOp {
+public:
+  RGWOp_DATALog_Lock() {}
+  ~RGWOp_DATALog_Lock() {}
+
+  int check_caps(RGWUserCaps& caps) {
+    return caps.check_cap("datalog", RGW_CAP_WRITE);
+  }
+  void execute();
+  virtual const char *name() {
+    return "lock_datalog_object";
   }
+};
+
+class RGWOp_DATALog_Unlock : public RGWRESTOp {
 public:
-  RGWOp_DATALog_Post() {}
-  ~RGWOp_DATALog_Post() {}
+  RGWOp_DATALog_Unlock() {}
+  ~RGWOp_DATALog_Unlock() {}
 
-  int check_caps(RGWUserCaps& caps);
+  int check_caps(RGWUserCaps& caps) {
+    return caps.check_cap("datalog", RGW_CAP_WRITE);
+  }
   void execute();
-  virtual const char *name();
+  virtual const char *name() {
+    return "unlock_datalog_object";
+  }
 };
 
 class RGWOp_DATALog_Delete : public RGWRESTOp {
diff --git a/src/rgw/rgw_rest_metadata.cc b/src/rgw/rgw_rest_metadata.cc
index 18d22f3c20e..69f8a5ccbc4 100644
--- a/src/rgw/rgw_rest_metadata.cc
+++ b/src/rgw/rgw_rest_metadata.cc
@@ -19,8 +19,9 @@
 #include "rgw_rest_metadata.h"
 #include "rgw_client_io.h"
 #include "common/errno.h"
-#define dout_subsys ceph_subsys_rgw
+#include "common/strtol.h"
 
+#define dout_subsys ceph_subsys_rgw
 
 const char *RGWOp_Metadata_Get::name() {
   return "get_metadata";
@@ -180,6 +181,58 @@ void RGWOp_Metadata_Delete::execute() {
   http_ret = 0;
 }
 
+void RGWOp_Metadata_Lock::execute() {
+  string duration_str, lock_id;
+  string metadata_key;
+
+  frame_metadata_key(s, metadata_key);
+
+  http_ret = 0;
+
+  duration_str = s->info.args.get("length");
+  lock_id      = s->info.args.get("lock_id");
+
+  if ((!s->info.args.exists("key")) || 
+      (duration_str.empty()) ||
+      lock_id.empty()) {
+    dout(5) << "Error invalid parameter list" << dendl;
+    http_ret = -EINVAL;
+    return;
+  }
+
+  int dur;
+  string err;
+
+  dur = strict_strtol(duration_str.c_str(), 10, &err);
+  if (!err.empty() || dur <= 0) {
+    dout(5) << "invalid length param " << duration_str << dendl;
+    http_ret = -EINVAL;
+    return;
+  }
+  utime_t time(dur, 0);
+  http_ret = store->meta_mgr->lock_exclusive(metadata_key, time, lock_id);
+}
+
+void RGWOp_Metadata_Unlock::execute() {
+  string lock_id;
+  string metadata_key;
+
+  frame_metadata_key(s, metadata_key);
+
+  http_ret = 0;
+
+  lock_id = s->info.args.get("lock_id");
+
+  if ((!s->info.args.exists("key")) || 
+      lock_id.empty()) {
+    dout(5) << "Error invalid parameter list" << dendl;
+    http_ret = -EINVAL;
+    return;
+  }
+
+  http_ret = store->meta_mgr->unlock(metadata_key, lock_id);
+}
+
 RGWOp *RGWHandler_Metadata::op_get() {
   if (s->info.args.exists("key"))
     return new RGWOp_Metadata_Get;
@@ -194,3 +247,13 @@ RGWOp *RGWHandler_Metadata::op_put() {
 RGWOp *RGWHandler_Metadata::op_delete() {
   return new RGWOp_Metadata_Delete;
 }
+
+RGWOp *RGWHandler_Metadata::op_post() {
+  if (s->info.args.exists("lock"))
+    return new RGWOp_Metadata_Lock;
+  else if (s->info.args.exists("unlock"))
+    return new RGWOp_Metadata_Unlock;
+
+  return NULL;
+}
+
diff --git a/src/rgw/rgw_rest_metadata.h b/src/rgw/rgw_rest_metadata.h
index 5bf47bb3a92..85993d08d58 100644
--- a/src/rgw/rgw_rest_metadata.h
+++ b/src/rgw/rgw_rest_metadata.h
@@ -63,11 +63,40 @@ public:
   virtual const char *name() { return "remove_metadata"; }
 };
 
+class RGWOp_Metadata_Lock : public RGWRESTOp {
+public:
+  RGWOp_Metadata_Lock() {}
+  ~RGWOp_Metadata_Lock() {}
+
+  int check_caps(RGWUserCaps& caps) {
+    return caps.check_cap("metadata", RGW_CAP_WRITE);
+  }
+  void execute();
+  virtual const char *name() {
+    return "lock_metadata_object";
+  }
+};
+
+class RGWOp_Metadata_Unlock : public RGWRESTOp {
+public:
+  RGWOp_Metadata_Unlock() {}
+  ~RGWOp_Metadata_Unlock() {}
+
+  int check_caps(RGWUserCaps& caps) {
+    return caps.check_cap("metadata", RGW_CAP_WRITE);
+  }
+  void execute();
+  virtual const char *name() {
+    return "unlock_metadata_object";
+  }
+};
+
 class RGWHandler_Metadata : public RGWHandler_Auth_S3 {
 protected:
   RGWOp *op_get();
   RGWOp *op_put();
   RGWOp *op_delete();
+  RGWOp *op_post();
 
   int read_permissions(RGWOp*) {
     return 0;
diff --git a/src/test/test_rgw_admin_log.cc b/src/test/test_rgw_admin_log.cc
index 10116610f3f..40d0f70c3bc 100644
--- a/src/test/test_rgw_admin_log.cc
+++ b/src/test/test_rgw_admin_log.cc
@@ -740,14 +740,15 @@ TEST(TestRGWAdmin, datalog_list) {
   EXPECT_EQ (parse_json_resp(parser), 0);
   JSONDecoder::decode_json("num_objects", num_objects, (JSONObj *)&parser);
   ASSERT_EQ(num_objects,g_ceph_context->_conf->rgw_data_log_num_shards);
-  
+ 
+  sleep(1);
   ASSERT_EQ(0, create_bucket());
   
   char *bucket_obj = (char *)malloc(TEST_BUCKET_OBJECT_SIZE);
   ASSERT_TRUE(bucket_obj != NULL);
   EXPECT_EQ(put_bucket_obj(TEST_BUCKET_OBJECT, bucket_obj, TEST_BUCKET_OBJECT_SIZE), 0);
   free(bucket_obj);
-  
+  sleep(1); 
   ss << "/admin/log?type=data&id=" << shard_id << "&start-time=" << start_time;
   rest_req = ss.str();
   g_test->send_request(string("GET"), rest_req);
@@ -760,7 +761,6 @@ TEST(TestRGWAdmin, datalog_list) {
     EXPECT_EQ(entry.entity_type, ENTITY_TYPE_BUCKET);
     EXPECT_EQ(entry.key.compare(TEST_BUCKET_NAME), 0);
   }
-
   ASSERT_EQ(0, delete_obj(TEST_BUCKET_OBJECT));
   sleep(1);
   ASSERT_EQ(get_formatted_time(end_time), 0);
@@ -900,11 +900,11 @@ TEST(TestRGWAdmin, datalog_lock_unlock) {
   ASSERT_EQ(0, caps_add(cname, perm));
   rest_req = "/admin/log?type=data&lock&id=1&length=3&lock_id=ceph";
   g_test->send_request(string("POST"), rest_req, read_dummy_post, NULL, sizeof(int));
-  EXPECT_EQ(200U, g_test->get_resp_code()); 
+  EXPECT_EQ(403U, g_test->get_resp_code()); 
   
   rest_req = "/admin/log?type=data&unlock&id=1&lock_id=ceph";
   g_test->send_request(string("POST"), rest_req, read_dummy_post, NULL, sizeof(int));
-  EXPECT_EQ(200U, g_test->get_resp_code()); 
+  EXPECT_EQ(403U, g_test->get_resp_code()); 
   
   ASSERT_EQ(0, caps_rm(cname, perm));
   perm = "write";
@@ -1310,11 +1310,11 @@ TEST(TestRGWAdmin, mdlog_lock_unlock) {
   ASSERT_EQ(0, caps_add(cname, perm));
   rest_req = "/admin/log?type=metadata&lock&id=1&length=3&lock_id=ceph";
   g_test->send_request(string("POST"), rest_req, read_dummy_post, NULL, sizeof(int));
-  EXPECT_EQ(200U, g_test->get_resp_code()); 
+  EXPECT_EQ(403U, g_test->get_resp_code()); 
   
   rest_req = "/admin/log?type=metadata&unlock&id=1&lock_id=ceph";
   g_test->send_request(string("POST"), rest_req, read_dummy_post, NULL, sizeof(int));
-  EXPECT_EQ(200U, g_test->get_resp_code()); 
+  EXPECT_EQ(403U, g_test->get_resp_code()); 
   
   ASSERT_EQ(0, caps_rm(cname, perm));
   perm = "write";
diff --git a/src/test/test_rgw_admin_meta.cc b/src/test/test_rgw_admin_meta.cc
index 5f811a97bf8..f4002fb7bf9 100644
--- a/src/test/test_rgw_admin_meta.cc
+++ b/src/test/test_rgw_admin_meta.cc
@@ -51,8 +51,9 @@ using namespace std;
 #define HTTP_RESPONSE_STR "RespCode"
 #define CEPH_CRYPTO_HMACSHA1_DIGESTSIZE 20
 #define RGW_ADMIN_RESP_PATH "/tmp/.test_rgw_admin_resp"
+#define CEPH_UID  "ceph"
 
-static string uid = "ceph";
+static string uid = CEPH_UID;
 static string display_name = "CEPH";
 static string meta_caps = "metadata";
 
@@ -350,7 +351,7 @@ int run_rgw_admin(string& cmd, string& resp) {
       resp = data;
       free(data);
       unlink(RGW_ADMIN_RESP_PATH);
-      /* cout << "radosgw-admin " << cmd << ": " << resp << std::endl; */
+      /* cout << "radosgw-admin " << cmd << ": " << resp << std::endl;*/
     }
   } else 
     return -1;
@@ -531,6 +532,13 @@ int compare_user_info(RGWUserInfo& i1, RGWUserInfo& i2) {
   return 0;
 }
 
+size_t read_dummy_post(void *ptr, size_t s, size_t n, void *ud) {
+  int dummy = 0;
+  memcpy(ptr, &dummy, sizeof(dummy));
+  return sizeof(dummy);
+}
+
+
 int parse_json_resp(JSONParser &parser) {
   string *resp;
   resp = (string *)g_test->get_response_data();
@@ -783,6 +791,101 @@ TEST(TestRGWAdmin, meta_put){
   ASSERT_EQ(0, user_rm(uid, display_name));
 }
 
+TEST(TestRGWAdmin, meta_lock_unlock) {
+  const char *perm = "*";
+  string rest_req;
+
+  ASSERT_EQ(0, user_create(uid, display_name));
+  ASSERT_EQ(0, meta_caps_add(perm));
+
+  rest_req = "/admin/metadata/user?key="CEPH_UID"&lock&length=3";
+  g_test->send_request(string("POST"), rest_req, read_dummy_post, NULL, sizeof(int));
+  EXPECT_EQ(400U, g_test->get_resp_code()); /*Bad request*/
+  
+  rest_req = "/admin/metadata/user?lock&length=3&lock_id=ceph";
+  g_test->send_request(string("POST"), rest_req, read_dummy_post, NULL, sizeof(int));
+  EXPECT_EQ(400U, g_test->get_resp_code()); /*Bad request*/
+
+  rest_req = "/admin/metadata/user?key="CEPH_UID"&unlock";
+  g_test->send_request(string("POST"), rest_req, read_dummy_post, NULL, sizeof(int));
+  EXPECT_EQ(400U, g_test->get_resp_code()); /*Bad request*/
+
+  rest_req = "/admin/metadata/user?unlock&lock_id=ceph";
+  g_test->send_request(string("POST"), rest_req, read_dummy_post, NULL, sizeof(int));
+  EXPECT_EQ(400U, g_test->get_resp_code()); /*Bad request*/
+  
+  rest_req = "/admin/metadata/user?key="CEPH_UID"&lock&length=3&lock_id=ceph";
+  g_test->send_request(string("POST"), rest_req, read_dummy_post, NULL, sizeof(int));
+  EXPECT_EQ(200U, g_test->get_resp_code()); 
+  
+  rest_req = "/admin/metadata/user?key="CEPH_UID"&unlock&lock_id=ceph";
+  g_test->send_request(string("POST"), rest_req, read_dummy_post, NULL, sizeof(int));
+  EXPECT_EQ(200U, g_test->get_resp_code()); 
+  
+  rest_req = "/admin/metadata/user?key="CEPH_UID"&lock&length=3&lock_id=ceph1";
+  g_test->send_request(string("POST"), rest_req, read_dummy_post, NULL, sizeof(int));
+  EXPECT_EQ(200U, g_test->get_resp_code()); 
+  
+  rest_req = "/admin/metadata/user?key="CEPH_UID"&unlock&lock_id=ceph1";
+  g_test->send_request(string("POST"), rest_req, read_dummy_post, NULL, sizeof(int));
+  EXPECT_EQ(200U, g_test->get_resp_code()); 
+  
+  rest_req = "/admin/metadata/user?key="CEPH_UID"&lock&length=3&lock_id=ceph";
+  g_test->send_request(string("POST"), rest_req, read_dummy_post, NULL, sizeof(int));
+  EXPECT_EQ(200U, g_test->get_resp_code()); 
+  utime_t sleep_time(3, 0);
+
+  rest_req = "/admin/metadata/user?key="CEPH_UID"&lock&length=3&lock_id=ceph1";
+  g_test->send_request(string("POST"), rest_req, read_dummy_post, NULL, sizeof(int));
+  EXPECT_EQ(500U, g_test->get_resp_code()); 
+
+  rest_req = "/admin/metadata/user?key="CEPH_UID"&lock&length=3&lock_id=ceph";
+  g_test->send_request(string("POST"), rest_req, read_dummy_post, NULL, sizeof(int));
+  EXPECT_EQ(409U, g_test->get_resp_code()); 
+  sleep_time.sleep();
+
+  rest_req = "/admin/metadata/user?key="CEPH_UID"&lock&length=3&lock_id=ceph1";
+  g_test->send_request(string("POST"), rest_req, read_dummy_post, NULL, sizeof(int));
+  EXPECT_EQ(200U, g_test->get_resp_code()); 
+  
+  rest_req = "/admin/metadata/user?key="CEPH_UID"&unlock&lock_id=ceph1";
+  g_test->send_request(string("POST"), rest_req, read_dummy_post, NULL, sizeof(int));
+  EXPECT_EQ(200U, g_test->get_resp_code()); 
+
+  ASSERT_EQ(0, meta_caps_rm(perm));
+  perm = "read";
+  ASSERT_EQ(0, meta_caps_add(perm));
+  rest_req = "/admin/metadata/user?key="CEPH_UID"&lock&length=3&lock_id=ceph";
+  g_test->send_request(string("POST"), rest_req, read_dummy_post, NULL, sizeof(int));
+  EXPECT_EQ(403U, g_test->get_resp_code()); 
+  
+  rest_req = "/admin/metadata/user?key="CEPH_UID"&unlock&lock_id=ceph";
+  g_test->send_request(string("POST"), rest_req, read_dummy_post, NULL, sizeof(int));
+  EXPECT_EQ(403U, g_test->get_resp_code()); 
+  
+  ASSERT_EQ(0, meta_caps_rm(perm));
+  perm = "write";
+  ASSERT_EQ(0, meta_caps_add(perm));
+  rest_req = "/admin/metadata/user?key="CEPH_UID"&lock&length=3&lock_id=ceph";
+  g_test->send_request(string("POST"), rest_req, read_dummy_post, NULL, sizeof(int));
+  EXPECT_EQ(200U, g_test->get_resp_code()); 
+  
+  rest_req = "/admin/metadata/user?key="CEPH_UID"&unlock&lock_id=ceph";
+  g_test->send_request(string("POST"), rest_req, read_dummy_post, NULL, sizeof(int));
+  EXPECT_EQ(200U, g_test->get_resp_code()); 
+  
+  ASSERT_EQ(0, meta_caps_rm(perm));
+  rest_req = "/admin/metadata/user?key="CEPH_UID"&lock&length=3&lock_id=ceph";
+  g_test->send_request(string("POST"), rest_req, read_dummy_post, NULL, sizeof(int));
+  EXPECT_EQ(403U, g_test->get_resp_code()); 
+  
+  rest_req = "/admin/metadata/user?key="CEPH_UID"&unlock&lock_id=ceph";
+  g_test->send_request(string("POST"), rest_req, read_dummy_post, NULL, sizeof(int));
+  EXPECT_EQ(403U, g_test->get_resp_code()); 
+  
+  ASSERT_EQ(0, user_rm(uid, display_name));
+}
+
 TEST(TestRGWAdmin, meta_delete){
   JSONParser parser;
   const char *perm = "*";