With mdlog lock and unlock functionality, listing and trimming base on shard_id

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

9 files changed, 384 insertions, 104 deletions

diff --git a/src/cls/log/cls_log.cc b/src/cls/log/cls_log.cc
index e8196660bf5..82616fd9712 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.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_admin.cc b/src/rgw/rgw_admin.cc
index 344cbdbdc69..9cbd4405b54 100644
--- a/src/rgw/rgw_admin.cc
+++ b/src/rgw/rgw_admin.cc
@@ -642,6 +642,7 @@ int main(int argc, char **argv)
   string start_marker;
   string end_marker;
   int max_entries = -1;
+  int shard_id = 0;
 
   std::string val;
   std::ostringstream errs;
@@ -707,6 +708,8 @@ int main(int argc, char **argv)
       start_date = val;
     } else if (ceph_argparse_witharg(args, i, &val, "--end-date", "--end-time", (char*)NULL)) {
       end_date = val;
+    } else if (ceph_argparse_witharg(args, i, &val, "--shard-id", (char*)NULL)) {
+      shard_id = atoi(val.c_str());
     } else if (ceph_argparse_witharg(args, i, &val, "--access", (char*)NULL)) {
       access = val;
       perm_mask = rgw_str_to_perm(access.c_str());
@@ -1761,7 +1764,7 @@ next:
 
     RGWMetadataLog *meta_log = store->meta_mgr->get_log();
 
-    meta_log->init_list_entries(store, start_time, end_time, &handle);
+    meta_log->init_list_entries(shard_id, start_time, end_time, &handle);
 
     bool truncated;
 
@@ -1797,7 +1800,7 @@ next:
 
     RGWMetadataLog *meta_log = store->meta_mgr->get_log();
 
-    ret = meta_log->trim(store, start_time, end_time);
+    ret = meta_log->trim(shard_id, start_time, end_time);
     if (ret < 0) {
       cerr << "ERROR: meta_log->trim(): " << cpp_strerror(-ret) << std::endl;
       return -ret;
diff --git a/src/rgw/rgw_metadata.cc b/src/rgw/rgw_metadata.cc
index d3e39daab19..105a3465461 100644
--- a/src/rgw/rgw_metadata.cc
+++ b/src/rgw/rgw_metadata.cc
@@ -79,14 +79,15 @@ int RGWMetadataLog::add_entry(RGWRados *store, string& section, string& key, buf
   return store->time_log_add(oid, now, section, key, bl);
 }
 
-void RGWMetadataLog::init_list_entries(RGWRados *store, utime_t& from_time, utime_t& end_time, void **handle)
+void RGWMetadataLog::init_list_entries(int shard_id, utime_t& from_time, utime_t& end_time, void **handle)
 {
-  LogListCtx *ctx = new LogListCtx(store);
+  LogListCtx *ctx = new LogListCtx();
 
+  ctx->cur_shard = shard_id;
   ctx->from_time = from_time;
   ctx->end_time = end_time;
 
-  get_shard_oid(0, ctx->cur_oid);
+  get_shard_oid(ctx->cur_shard, ctx->cur_oid);
 
   *handle = (void *)ctx;
 }
@@ -102,65 +103,51 @@ int RGWMetadataLog::list_entries(void *handle,
                  bool *truncated) {
   LogListCtx *ctx = (LogListCtx *)handle;
 
-  if (ctx->done || !max_entries) {
+  if (!max_entries) {
     *truncated = false;
     return 0;
   }
 
   entries.clear();
 
-  do {
-    list<cls_log_entry> ents;
-    bool is_truncated;
-    int ret = store->time_log_list(ctx->cur_oid, ctx->from_time, ctx->end_time,
-                                 max_entries - entries.size(), ents, ctx->marker, &is_truncated);
-    if (ret == -ENOENT) {
-      is_truncated = false;
-      ret = 0;
-    }
-    if (ret < 0)
-      return ret;
-
-    if (ents.size()) {
-      entries.splice(entries.end(), ents);
-    }
-
-    if (!is_truncated) {
-      ++ctx->cur_shard;
-      if (ctx->cur_shard < cct->_conf->rgw_md_log_max_shards) {
-        get_shard_oid(ctx->cur_shard, ctx->cur_oid);
-        ctx->marker.clear();
-      } else {
-        ctx->done = true;
-        break;
-      }
-    }
-  } while (entries.size() < (size_t)max_entries);
-
-  *truncated = !ctx->done;
-
+  int ret = store->time_log_list(ctx->cur_oid, ctx->from_time, ctx->end_time,
+                                 max_entries, entries, ctx->marker, truncated);
+  if ((ret < 0) && (ret != -ENOENT))
+    return ret;
   return 0;
 }
 
-int RGWMetadataLog::trim(RGWRados *store, utime_t& from_time, utime_t& end_time)
+int RGWMetadataLog::trim(int shard_id, utime_t& from_time, utime_t& end_time)
 {
   string oid;
-  for (int shard = 0; shard < cct->_conf->rgw_md_log_max_shards; shard++) {
-    get_shard_oid(shard, oid);
+  get_shard_oid(shard_id, oid);
 
-    int ret;
+  int ret;
 
-    ret = store->time_log_trim(oid, from_time, end_time);
+  ret = store->time_log_trim(oid, from_time, end_time);
 
-    if (ret == -ENOENT)
-      ret = 0;
+  if (ret == -ENOENT)
+    ret = 0;
 
-    if (ret < 0)
-      return ret;
-  }
+  if (ret < 0)
+    return ret;
 
   return 0;
 }
+  
+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);
+}
+
+int RGWMetadataLog::unlock(int shard_id, string& owner_id) {
+  string oid;
+  get_shard_oid(shard_id, oid);
+
+  return store->log_unlock(oid, owner_id);
+}
 
 obj_version& RGWMetadataObject::get_version()
 {
diff --git a/src/rgw/rgw_metadata.h b/src/rgw/rgw_metadata.h
index 46a90b826e8..299b3ff35b3 100644
--- a/src/rgw/rgw_metadata.h
+++ b/src/rgw/rgw_metadata.h
@@ -76,7 +76,6 @@ public:
   int add_entry(RGWRados *store, string& section, string& key, bufferlist& bl);
 
   struct LogListCtx {
-    RGWRados *store;
     int cur_shard;
     string marker;
     utime_t from_time;
@@ -86,17 +85,19 @@ public:
 
     bool done;
 
-    LogListCtx(RGWRados *_store) : store(_store), cur_shard(0), done(false) {}
+    LogListCtx() : done(false) {}
   };
 
-  void init_list_entries(RGWRados *store, utime_t& from_time, utime_t& end_time, void **handle);
+  void init_list_entries(int shard_id, utime_t& from_time, utime_t& end_time, void **handle);
   void complete_list_entries(void *handle);
   int list_entries(void *handle,
                    int max_entries,
                    list<cls_log_entry>& entries,
                    bool *truncated);
 
-  int trim(RGWRados *store, utime_t& from_time, utime_t& end_time);
+  int trim(int shard_id, utime_t& from_time, utime_t& end_time);
+  int lock_exclusive(int shard_id, utime_t& duration, string& owner_id);
+  int unlock(int shard_id, string& owner_id);
 };
 
 class RGWMetadataLogData;
diff --git a/src/rgw/rgw_rados.cc b/src/rgw/rgw_rados.cc
index 9f0914249c0..a85dd4c8c25 100644
--- a/src/rgw/rgw_rados.cc
+++ b/src/rgw/rgw_rados.cc
@@ -19,6 +19,7 @@
 #include "cls/refcount/cls_refcount_client.h"
 #include "cls/version/cls_version_client.h"
 #include "cls/log/cls_log_client.h"
+#include "cls/lock/cls_lock_client.h"
 
 #include "rgw_tools.h"
 
@@ -57,7 +58,7 @@ 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;
 
@@ -1185,6 +1186,31 @@ 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) {
+  librados::IoCtx io_ctx;
+
+  const char *log_pool = zone.log_pool.name.c_str();
+  int r = rados->ioctx_create(log_pool, io_ctx);
+  if (r < 0)
+    return r;
+  rados::cls::lock::Lock l(rgw_log_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) {
+  librados::IoCtx io_ctx;
+
+  const char *log_pool = zone.log_pool.name.c_str();
+  int r = rados->ioctx_create(log_pool, io_ctx);
+  if (r < 0)
+    return r;
+  rados::cls::lock::Lock l(rgw_log_lock_name);
+  l.set_cookie(owner_id);
+  return l.unlock(&io_ctx, oid);
+}
+
 int RGWRados::decode_policy(bufferlist& bl, ACLOwner *owner)
 {
   bufferlist::iterator i = bl.begin();
diff --git a/src/rgw/rgw_rados.h b/src/rgw/rgw_rados.h
index 01d2724fd83..33269c6e239 100644
--- a/src/rgw/rgw_rados.h
+++ b/src/rgw/rgw_rados.h
@@ -966,6 +966,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);
 
   /// 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 07bd813f1e7..2a7dbf1b403 100644
--- a/src/rgw/rgw_rest_log.cc
+++ b/src/rgw/rgw_rest_log.cc
@@ -36,31 +36,42 @@ static int parse_date_str(string& in, utime_t& out) {
 }
 
 void RGWOp_MDLog_List::execute() {
-  string   st = s->args.get("start-time"),
-           et = s->args.get("end-time");
-  utime_t  ut_st, 
-           ut_et;
-  void    *handle;
-  list<cls_log_entry> entries;
+  string   shard = s->args.get("id");
+
+  if (!shard.empty()) {
+    string   st = s->args.get("start-time"),
+             et = s->args.get("end-time"),
+             err;
+    utime_t  ut_st, 
+             ut_et;
+    void    *handle;
+    int      shard_id;
+    list<cls_log_entry> entries;
+
+    shard_id = strict_strtol(shard.c_str(), 10, &err);
+    if (!err.empty()) {
+      dout(5) << "Error parsing shard_id " << shard << dendl;
+      http_ret = -EINVAL;
+      return;
+    }
 
-  if (parse_date_str(st, ut_st) < 0) {
-    http_ret = -EINVAL;
-    return;
-  }
+    if (parse_date_str(st, ut_st) < 0) {
+      http_ret = -EINVAL;
+      return;
+    }
 
-  if (parse_date_str(et, ut_et) < 0) {
-    http_ret = -EINVAL;
-    return;
-  }
+    if (parse_date_str(et, ut_et) < 0) {
+      http_ret = -EINVAL;
+      return;
+    }
 
-  RGWMetadataLog *meta_log = store->meta_mgr->get_log();
+    RGWMetadataLog *meta_log = store->meta_mgr->get_log();
 
-  meta_log->init_list_entries(store, ut_st, ut_et, &handle);
+    meta_log->init_list_entries(shard_id, ut_st, ut_et, &handle);
 
-  bool truncated;
+    bool truncated;
 
-  s->formatter->open_array_section("entries");
-  do {
+    s->formatter->open_array_section("entries");
     http_ret = meta_log->list_entries(handle, 1000, entries, &truncated);
     if (http_ret < 0) {
       return;
@@ -72,10 +83,16 @@ void RGWOp_MDLog_List::execute() {
       store->meta_mgr->dump_log_entry(entry, s->formatter);
       s->formatter->flush(out_stream);
     }
-  } while (truncated);
 
-  s->formatter->close_section();
-  s->formatter->flush(out_stream);
+    s->formatter->close_section();
+    s->formatter->flush(out_stream);
+  } else {
+    unsigned num_objects = s->cct->_conf->rgw_md_log_max_shards;
+    s->formatter->open_object_section("num_ojects");
+    s->formatter->dump_unsigned("num_objects", num_objects);
+    s->formatter->close_section();
+    s->formatter->flush(out_stream);
+  }
 
   http_ret = 0;
 }
@@ -92,11 +109,21 @@ void RGWOp_MDLog_List::send_response() {
 
 void RGWOp_MDLog_Delete::execute() {
   string   st = s->args.get("start-time"),
-           et = s->args.get("end-time");
+           et = s->args.get("end-time"),
+           shard = s->args.get("id"),
+           err;
   utime_t  ut_st, 
            ut_et;
+  int     shard_id;
 
   http_ret = 0;
+
+  shard_id = strict_strtol(shard.c_str(), 10, &err);
+  if (!err.empty()) {
+    dout(5) << "Error parsing shard_id " << shard << dendl;
+    http_ret = -EINVAL;
+    return;
+  }
   if (st.empty() || et.empty()) {
     http_ret = -EINVAL;
     return;
@@ -113,7 +140,69 @@ void RGWOp_MDLog_Delete::execute() {
   }
   RGWMetadataLog *meta_log = store->meta_mgr->get_log();
 
-  http_ret = meta_log->trim(store, ut_st, ut_et);
+  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;
+  }
+  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;
+}
+
+void RGWOp_MDLog_Post::execute() {
+  string shard_id_str, duration_str, lock_id;
+  int shard_id;
+  int pt = get_post_type();
+
+  http_ret = 0;
+
+  shard_id_str = s->args.get("id");
+  if (pt == MDLOG_POST_LOCK)
+    duration_str = s->args.get("length");
+  lock_id      = s->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;
+    return;
+  }
+
+  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();
+  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;
 }
 
 void RGWOp_BILog_List::execute() {
@@ -235,3 +324,17 @@ RGWOp *RGWHandler_Log::op_delete() {
     return new RGWOp_BILog_Delete;
   return NULL;
 }
+
+RGWOp *RGWHandler_Log::op_post() {
+  bool exists;
+  string type = s->args.get("type", &exists);
+
+  if (!exists) {
+    return NULL;
+  }
+
+  if (type.compare("metadata") == 0)
+    return new RGWOp_MDLog_Post;
+  return NULL;
+}
+
diff --git a/src/rgw/rgw_rest_log.h b/src/rgw/rgw_rest_log.h
index b3ef0690501..2471943c295 100644
--- a/src/rgw/rgw_rest_log.h
+++ b/src/rgw/rgw_rest_log.h
@@ -68,6 +68,31 @@ 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->args.get("lock", &exists);
+    if (exists) 
+      return MDLOG_POST_LOCK;
+    s->args.get("unlock", &exists);
+    if (exists)
+      return MDLOG_POST_UNLOCK;
+    return MDLOG_POST_INVALID;
+  }
+public:
+  RGWOp_MDLog_Post() {}
+  ~RGWOp_MDLog_Post() {}
+
+  int check_caps(RGWUserCaps& caps);
+  void execute();
+  virtual const char *name();
+};
+
 class RGWOp_MDLog_Delete : public RGWRESTOp {
 public:
   RGWOp_MDLog_Delete() {}
@@ -86,6 +111,7 @@ class RGWHandler_Log : public RGWHandler_Auth_S3 {
 protected:
   RGWOp *op_get();
   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 e0432d41756..a2a3c2ad56b 100644
--- a/src/test/test_rgw_admin_log.cc
+++ b/src/test/test_rgw_admin_log.cc
@@ -34,6 +34,8 @@ extern "C"{
 #include "global/global_init.h"
 #include "rgw/rgw_common.h"
 #include "rgw/rgw_rados.h"
+#include "include/utime.h"
+#include "include/object.h"
 #define GTEST
 #ifdef GTEST
 #include <gtest/gtest.h>
@@ -482,6 +484,12 @@ static int delete_bucket(void){
   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);
+}
+
 size_t read_bucket_object(void *ptr, size_t s, size_t n, void *ud) {
   memcpy(ptr, ud, TEST_BUCKET_OBJECT_SIZE);
   return TEST_BUCKET_OBJECT_SIZE;
@@ -660,6 +668,13 @@ static int get_bilog_list(list<cls_bilog_entry> &entries) {
   return 0;
 }
 
+unsigned get_shard_id(string& key, int max_shards) {
+  string section = "user";
+  uint32_t val = ceph_str_hash_linux(key.c_str(), key.size());
+  val ^= ceph_str_hash_linux(section.c_str(), section.size());
+  return (unsigned)(val % max_shards);
+}
+
 TEST(TestRGWAdmin, mdlog_list) {
   string start_time, 
          end_time,
@@ -667,13 +682,25 @@ TEST(TestRGWAdmin, mdlog_list) {
   const char *cname = "mdlog",
              *perm = "*";
   string rest_req;
+  unsigned shard_id = get_shard_id(uid, g_ceph_context->_conf->rgw_md_log_max_shards);
+  stringstream ss;
 
   ASSERT_EQ(get_formatted_time(start_time), 0);
   ASSERT_EQ(0, user_create(uid, display_name));
   ASSERT_EQ(0, caps_add(cname, perm));
 
-  rest_req = "/admin/log?type=metadata&start-time=";
-  rest_req.append(start_time);
+  rest_req = "/admin/log?type=metadata";
+  g_test->send_request(string("GET"), rest_req);
+  EXPECT_EQ(200U, g_test->get_resp_code());
+  JSONParser parser;
+  int num_objects;
+  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_md_log_max_shards);
+
+  ss.str("");
+  ss << "/admin/log?type=metadata&id=" << shard_id << "&start-time=" << start_time;
+  rest_req = ss.str();
   g_test->send_request(string("GET"), rest_req);
   EXPECT_EQ(200U, g_test->get_resp_code());
   
@@ -705,8 +732,9 @@ TEST(TestRGWAdmin, mdlog_list) {
   ASSERT_EQ(0, caps_rm(cname, perm));
   perm="read";
   ASSERT_EQ(0, caps_add(cname, perm));
-  rest_req = "/admin/log?type=metadata&start-time=";
-  rest_req.append(start_time_2);
+  ss.str("");
+  ss << "/admin/log?type=metadata&id=" << shard_id << "&start-time=" << start_time_2;
+  rest_req = ss.str();
   g_test->send_request(string("GET"), rest_req);
   EXPECT_EQ(200U, g_test->get_resp_code());
  
@@ -741,8 +769,9 @@ TEST(TestRGWAdmin, mdlog_list) {
   perm = "*";
   ASSERT_EQ(0, caps_add(cname, perm));
 
-  rest_req = "/admin/log?type=metadata&start-time=";
-  rest_req.append(start_time_2);
+  ss.str("");
+  ss << "/admin/log?type=metadata&id=" << shard_id << "&start-time=" << start_time_2;
+  rest_req = ss.str();
   g_test->send_request(string("GET"), rest_req);
   EXPECT_EQ(200U, g_test->get_resp_code());
   
@@ -769,10 +798,10 @@ TEST(TestRGWAdmin, mdlog_list) {
 
   sleep(1);
   ASSERT_EQ(get_formatted_time(end_time), 0);
-  rest_req = "/admin/log?type=metadata&start-time=";
-  rest_req.append(start_time);
-  rest_req.append("&end-time=");
-  rest_req.append(end_time);
+  ss.str("");
+  ss << "/admin/log?type=metadata&id=" << shard_id << "&start-time=" << start_time 
+    << "&end-time=" << end_time;
+  rest_req = ss.str();
   g_test->send_request(string("GET"), rest_req);
   EXPECT_EQ(200U, g_test->get_resp_code());
   entries.clear();
@@ -781,8 +810,9 @@ TEST(TestRGWAdmin, mdlog_list) {
 
 
   ASSERT_EQ(0, caps_rm(cname, perm));
-  rest_req = "/admin/log?type=metadata&start-time=";
-  rest_req.append(start_time);
+  ss.str("");
+  ss << "/admin/log?type=metadata&id=" << shard_id << "&start-time=" << start_time;
+  rest_req = ss.str();
   g_test->send_request(string("GET"), rest_req);
   EXPECT_EQ(403U, g_test->get_resp_code());
 
@@ -790,10 +820,10 @@ TEST(TestRGWAdmin, mdlog_list) {
   ASSERT_EQ(0, caps_add(cname, perm));
   sleep(1);
   ASSERT_EQ(get_formatted_time(end_time), 0);
-  rest_req = "/admin/log?type=metadata&start-time=";
-  rest_req.append(start_time);
-  rest_req.append("&end-time=");
-  rest_req.append(end_time);
+  ss.str("");
+  ss << "/admin/log?type=metadata&id=" << shard_id << "&start-time=" << start_time 
+    << "&end-time=" << end_time;
+  rest_req = ss.str();
   g_test->send_request(string("DELETE"), rest_req);
   EXPECT_EQ(200U, g_test->get_resp_code());
 
@@ -808,14 +838,17 @@ TEST(TestRGWAdmin, mdlog_trim) {
              *perm = "*";
   string rest_req;
   list<cls_log_entry_json> entries;
+  unsigned shard_id = get_shard_id(uid, g_ceph_context->_conf->rgw_md_log_max_shards);
+  ostringstream ss;
 
   sleep(1);
   ASSERT_EQ(get_formatted_time(start_time), 0);
   ASSERT_EQ(0, user_create(uid, display_name));
   ASSERT_EQ(0, caps_add(cname, perm));
 
-  rest_req = "/admin/log?type=metadata&start-time=";
-  rest_req.append(start_time);
+  ss.str("");
+  ss << "/admin/log?type=metadata&id=" << shard_id << "&start-time=" << start_time;
+  rest_req = ss.str();
   g_test->send_request(string("DELETE"), rest_req);
   EXPECT_EQ(400U, g_test->get_resp_code()); /*Bad request*/
 
@@ -826,15 +859,15 @@ TEST(TestRGWAdmin, mdlog_trim) {
 
   sleep(1);
   ASSERT_EQ(get_formatted_time(end_time), 0);
-  rest_req = "/admin/log?type=metadata&start-time=";
-  rest_req.append(start_time);
-  rest_req.append("&end-time=");
-  rest_req.append(end_time);
+  ss.str("");
+  ss << "/admin/log?type=metadata&id=" << shard_id << "&start-time=" << start_time << "&end-time=" << end_time;
+  rest_req = ss.str();
   g_test->send_request(string("DELETE"), rest_req);
   EXPECT_EQ(200U, g_test->get_resp_code());
 
-  rest_req = "/admin/log?type=metadata&start-time=";
-  rest_req.append(start_time);
+  ss.str("");
+  ss << "/admin/log?type=metadata&id=" << shard_id << "&start-time=" << start_time;
+  rest_req = ss.str();
   g_test->send_request(string("GET"), rest_req);
   EXPECT_EQ(200U, g_test->get_resp_code());
   entries.clear();
@@ -845,10 +878,9 @@ TEST(TestRGWAdmin, mdlog_trim) {
   perm="write";
   ASSERT_EQ(0, caps_add(cname, perm));
   ASSERT_EQ(get_formatted_time(end_time), 0);
-  rest_req = "/admin/log?type=metadata&start-time=";
-  rest_req.append(start_time);
-  rest_req.append("&end-time=");
-  rest_req.append(end_time);
+  ss.str("");
+  ss << "/admin/log?type=metadata&id=" << shard_id << "&start-time=" << start_time << "&end-time=" << end_time;
+  rest_req = ss.str();
   g_test->send_request(string("DELETE"), rest_req);
   EXPECT_EQ(200U, g_test->get_resp_code());
 
@@ -858,6 +890,106 @@ TEST(TestRGWAdmin, mdlog_trim) {
   ASSERT_EQ(0, user_rm(uid, display_name));
 }
 
+TEST(TestRGWAdmin, mdlog_lock_unlock) {
+  const char *cname = "mdlog",
+             *perm = "*";
+  string rest_req;
+
+  ASSERT_EQ(0, user_create(uid, display_name));
+  ASSERT_EQ(0, caps_add(cname, perm));
+
+  rest_req = "/admin/log?type=metadata&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/log?type=metadata&lock&id=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/log?type=metadata&lock&length=3&id=1";
+  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/log?type=metadata&unlock&id=1";
+  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/log?type=metadata&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/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()); 
+  
+  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()); 
+  
+  rest_req = "/admin/log?type=metadata&lock&id=1&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/log?type=metadata&unlock&id=1&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/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()); 
+  utime_t sleep_time(3, 0);
+
+  rest_req = "/admin/log?type=metadata&lock&id=1&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/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(409U, g_test->get_resp_code()); 
+  sleep_time.sleep();
+
+  rest_req = "/admin/log?type=metadata&lock&id=1&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/log?type=metadata&unlock&id=1&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, caps_rm(cname, perm));
+  perm = "read";
+  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()); 
+  
+  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()); 
+  
+  ASSERT_EQ(0, caps_rm(cname, perm));
+  perm = "write";
+  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()); 
+  
+  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()); 
+  
+  ASSERT_EQ(0, caps_rm(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(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(403U, g_test->get_resp_code()); 
+  
+  ASSERT_EQ(0, user_rm(uid, display_name));
+}
+
 TEST(TestRGWAdmin, bilog_list) {
   const char *cname = "bilog",
              *perm = "*";