RESTful APIs for data changes log implemented with test cases

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

6 files changed, 626 insertions, 6 deletions

diff --git a/src/cls/log/cls_log.cc b/src/cls/log/cls_log.cc
index 82616fd9712..73169edb924 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 >= op.from_time));
+  bool use_time_boundary = (!op.from_time.is_zero() && !op.to_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 6048359b821..b22ceef97e0 100644
--- a/src/rgw/rgw_bucket.cc
+++ b/src/rgw/rgw_bucket.cc
@@ -1195,6 +1195,15 @@ int RGWDataChangesLog::list_entries(utime_t& start_time, utime_t& end_time, int
   return 0;
 }
 
+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;
+
+  return 0;
+}
+
 int RGWDataChangesLog::trim_entries(utime_t& start_time, utime_t& end_time)
 {
   for (int shard = 0; shard < num_shards; shard++) {
diff --git a/src/rgw/rgw_bucket.h b/src/rgw/rgw_bucket.h
index 2de140db3fd..f1fbd184df0 100644
--- a/src/rgw/rgw_bucket.h
+++ b/src/rgw/rgw_bucket.h
@@ -341,8 +341,14 @@ public:
   int renew_entries();
   int list_entries(int shard, utime_t& start_time, utime_t& end_time, int max_entries,
                list<rgw_data_change>& entries, string& marker, bool *truncated);
+  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);
+  }
+  int unlock(int shard_id, string& owner_id) {
+    return store->log_unlock(oids[shard_id], owner_id);
+  }
   struct LogMarker {
     int shard;
     string marker;
diff --git a/src/rgw/rgw_rest_log.cc b/src/rgw/rgw_rest_log.cc
index 97c4b288d27..886dbbe979e 100644
--- a/src/rgw/rgw_rest_log.cc
+++ b/src/rgw/rgw_rest_log.cc
@@ -309,6 +309,173 @@ void RGWOp_BILog_Delete::execute() {
   return;
 }
 
+void RGWOp_DATALog_List::execute() {
+  string   shard = s->args.get("id");
+
+  string   st = s->args.get("start-time"),
+           et = s->args.get("end-time"),
+           err;
+  utime_t  ut_st, 
+           ut_et;
+  int      shard_id;
+
+  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(et, ut_et) < 0) {
+    http_ret = -EINVAL;
+    return;
+  }
+
+  string marker;
+  bool truncated;
+#define DATALOG_LIST_MAX_ENTRIES 1000
+
+  http_ret = store->data_log->list_entries(shard_id, ut_st, ut_et, 
+                               DATALOG_LIST_MAX_ENTRIES, entries, marker, &truncated);
+}
+
+void RGWOp_DATALog_List::send_response() {
+  set_req_state_err(s, http_ret);
+  dump_errno(s);
+  end_header(s);
+
+  if (http_ret < 0)
+    return;
+
+  s->formatter->open_array_section("entries");
+  for (list<rgw_data_change>::iterator iter = entries.begin(); 
+       iter != entries.end(); ++iter) {
+    rgw_data_change& entry = *iter;
+    encode_json("entry", entry, s->formatter);
+    flusher.flush();
+  }
+  s->formatter->close_section();
+  flusher.flush();
+}
+
+
+void RGWOp_DATALog_GetShardsInfo::execute() {
+  num_objects = s->cct->_conf->rgw_data_log_num_shards;
+  http_ret = 0;
+}
+
+void RGWOp_DATALog_GetShardsInfo::send_response() {
+  set_req_state_err(s, http_ret);
+  dump_errno(s);
+  end_header(s);
+
+  s->formatter->open_object_section("num_objects");
+  s->formatter->dump_unsigned("num_objects", num_objects);
+  s->formatter->close_section();
+  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;
+  }
+  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;
+}
+
+void RGWOp_DATALog_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 == DATALOG_POST_LOCK)
+    duration_str = s->args.get("length");
+  lock_id      = s->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;
+    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;
+  }
+
+  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;
+}
+
+void RGWOp_DATALog_Delete::execute() {
+  string   st = s->args.get("start-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;
+  }
+
+  if (parse_date_str(st, ut_st) < 0) {
+    http_ret = -EINVAL;
+    return;
+  }
+
+  if (parse_date_str(et, ut_et) < 0) {
+    http_ret = -EINVAL;
+    return;
+  }
+
+  http_ret = store->data_log->trim_entries(shard_id, ut_st, ut_et);
+}
+
 RGWOp *RGWHandler_Log::op_get() {
   bool exists;
   string type = s->info.args.get("type", &exists);
@@ -325,6 +492,12 @@ 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")) {
+      return new RGWOp_DATALog_List;
+    } else {
+      return new RGWOp_DATALog_GetShardsInfo;
+    }
   }
   return NULL;
 }
@@ -341,6 +514,8 @@ RGWOp *RGWHandler_Log::op_delete() {
     return new RGWOp_MDLog_Delete;
   else if (type.compare("bucket-index") == 0) 
     return new RGWOp_BILog_Delete;
+  else if (type.compare("data") == 0)
+    return new RGWOp_DATALog_Delete;
   return NULL;
 }
 
@@ -354,6 +529,8 @@ RGWOp *RGWHandler_Log::op_post() {
 
   if (type.compare("metadata") == 0)
     return new RGWOp_MDLog_Post;
+  else if (type.compare("data") == 0)
+    return new RGWOp_DATALog_Post;
   return NULL;
 }
 
diff --git a/src/rgw/rgw_rest_log.h b/src/rgw/rgw_rest_log.h
index 635758d77cb..e5b94954549 100644
--- a/src/rgw/rgw_rest_log.h
+++ b/src/rgw/rgw_rest_log.h
@@ -129,6 +129,85 @@ public:
   }
 };
 
+class RGWOp_DATALog_List : public RGWRESTOp {
+  list<rgw_data_change> entries;
+  int http_ret;
+public:
+  RGWOp_DATALog_List() : http_ret(0) {}
+  ~RGWOp_DATALog_List() {}
+
+  int check_caps(RGWUserCaps& caps) {
+    return caps.check_cap("datalog", RGW_CAP_READ);
+  }
+  int verify_permission() {
+    return check_caps(s->user.caps);
+  }
+  void execute();
+  virtual void send_response();
+  virtual const char *name() {
+    return "list_data_changes_log";
+  }
+};
+
+class RGWOp_DATALog_GetShardsInfo : public RGWRESTOp {
+  unsigned num_objects;
+  int http_ret;
+public:
+  RGWOp_DATALog_GetShardsInfo() : num_objects(0), http_ret(0) {}
+  ~RGWOp_DATALog_GetShardsInfo() {}
+
+  int check_caps(RGWUserCaps& caps) {
+    return caps.check_cap("datalog", RGW_CAP_READ);
+  }
+  int verify_permission() {
+    return check_caps(s->user.caps);
+  }
+  void execute();
+  virtual void send_response();
+  virtual const char *name() {
+    return "get_data_changes_log_shards_info";
+  }
+};
+
+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;
+  }
+public:
+  RGWOp_DATALog_Post() {}
+  ~RGWOp_DATALog_Post() {}
+
+  int check_caps(RGWUserCaps& caps);
+  void execute();
+  virtual const char *name();
+};
+
+class RGWOp_DATALog_Delete : public RGWRESTOp {
+public:
+  RGWOp_DATALog_Delete() {}
+  ~RGWOp_DATALog_Delete() {}
+
+  int check_caps(RGWUserCaps& caps) {
+    return caps.check_cap("datalog", RGW_CAP_WRITE);
+  }
+  void execute();
+  virtual const char *name() {
+    return "trim_data_changes_log";
+  }
+};
+
 class RGWHandler_Log : public RGWHandler_Auth_S3 {
 protected:
   RGWOp *op_get();
diff --git a/src/test/test_rgw_admin_log.cc b/src/test/test_rgw_admin_log.cc
index d3d312b7c5f..10116610f3f 100644
--- a/src/test/test_rgw_admin_log.cc
+++ b/src/test/test_rgw_admin_log.cc
@@ -33,6 +33,7 @@ extern "C"{
 #include "common/Finisher.h"
 #include "global/global_init.h"
 #include "rgw/rgw_common.h"
+#include "rgw/rgw_bucket.h"
 #include "rgw/rgw_rados.h"
 #include "include/utime.h"
 #include "include/object.h"
@@ -243,7 +244,7 @@ void get_date(string& d){
           days[tm.tm_wday], 
           tm.tm_mday, months[tm.tm_mon], 
           tm.tm_year + 1900,
-          tm.tm_hour, tm.tm_min, 0 /*tm.tm_sec*/);
+          tm.tm_hour, tm.tm_min, tm.tm_sec);
   d = date;
 }
 
@@ -668,13 +669,360 @@ static int get_bilog_list(list<cls_bilog_entry> &entries) {
   return 0;
 }
 
-unsigned get_shard_id(string& key, int max_shards) {
+static int decode_json(JSONObj *obj, rgw_data_change& ret) {
+  string entity;
+
+  JSONDecoder::decode_json("entity_type", entity, obj);
+  if (entity.compare("bucket") == 0)
+    ret.entity_type = ENTITY_TYPE_BUCKET;
+  JSONDecoder::decode_json("key", ret.key, obj);
+  return 0;
+}
+
+static int get_datalog_list(list<rgw_data_change> &entries) {
+  JSONParser parser;
+
+  if (parse_json_resp(parser) != 0)
+    return -1;
+  if (!parser.is_array()) 
+    return -1;
+
+  vector<string> l;
+  l = parser.get_array_elements();
+  int loop = 0;
+  for(vector<string>::iterator it = l.begin();
+      it != l.end(); it++, loop++) {
+    JSONParser jp;
+    rgw_data_change entry;
+
+    if(!jp.parse((*it).c_str(), (*it).length())) {
+      cerr << "Error parsing log json object" << std::endl;
+      return -1;
+    }
+    EXPECT_EQ(decode_json((JSONObj *)&jp, entry), 0);
+    entries.push_back(entry);
+  }
+  return 0;
+}
+
+unsigned get_mdlog_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);
 }
 
+unsigned get_datalog_shard_id(const char *bucket_name, int max_shards) {
+  uint32_t r = ceph_str_hash_linux(bucket_name, strlen(bucket_name)) % max_shards;
+  return (int)r;
+}
+
+TEST(TestRGWAdmin, datalog_list) {
+  string start_time, 
+         end_time,
+         start_time_2;
+  const char *cname = "datalog",
+             *perm = "*";
+  string rest_req;
+  unsigned shard_id = get_datalog_shard_id(TEST_BUCKET_NAME, g_ceph_context->_conf->rgw_data_log_num_shards);
+  stringstream ss;
+  list<rgw_data_change> entries;
+
+  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=data";
+  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_data_log_num_shards);
+  
+  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);
+  
+  ss << "/admin/log?type=data&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();
+  get_datalog_list(entries);
+  EXPECT_EQ(1U, entries.size());
+  if (entries.size() == 1) {
+    rgw_data_change entry = *(entries.begin());
+    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);
+  ss.str("");
+  ss << "/admin/log?type=data&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();
+  get_datalog_list(entries);
+  EXPECT_EQ(1U, entries.size());
+  if (entries.size() == 1) {
+    list<rgw_data_change>::iterator it = (entries.begin());
+    EXPECT_EQ((*it).entity_type, ENTITY_TYPE_BUCKET);
+    EXPECT_EQ((*it).key.compare(TEST_BUCKET_NAME), 0);
+  }
+
+  sleep(1);
+  EXPECT_EQ(put_bucket_obj(TEST_BUCKET_OBJECT, bucket_obj, TEST_BUCKET_OBJECT_SIZE), 0);
+  sleep(20);
+  ss.str("");
+  ss << "/admin/log?type=data&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();
+  get_datalog_list(entries);
+  EXPECT_EQ(2U, entries.size());
+  if (entries.size() == 2) {
+    list<rgw_data_change>::iterator it = (entries.begin());
+    EXPECT_EQ((*it).entity_type, ENTITY_TYPE_BUCKET);
+    EXPECT_EQ((*it).key.compare(TEST_BUCKET_NAME), 0);
+    it++; 
+    EXPECT_EQ((*it).entity_type, ENTITY_TYPE_BUCKET);
+    EXPECT_EQ((*it).key.compare(TEST_BUCKET_NAME), 0);
+  }
+
+  ss.str("");
+  ss << "/admin/log?type=data&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();
+  get_datalog_list(entries);
+  EXPECT_EQ(1U, entries.size());
+
+  ASSERT_EQ(0, caps_rm(cname, perm));
+  perm = "read";
+  ASSERT_EQ(0, caps_add(cname, perm));
+  ss.str("");
+  ss << "/admin/log?type=data&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());
+  ASSERT_EQ(0, caps_rm(cname, perm));
+  ss.str("");
+  ss << "/admin/log?type=data&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());
+
+  ASSERT_EQ(0, delete_obj(TEST_BUCKET_OBJECT));
+  ASSERT_EQ(0, delete_bucket());
+  ASSERT_EQ(0, user_rm(uid, display_name));
+}
+
+TEST(TestRGWAdmin, datalog_lock_unlock) {
+  const char *cname = "datalog",
+             *perm = "*";
+  string rest_req;
+
+  ASSERT_EQ(0, user_create(uid, display_name));
+  ASSERT_EQ(0, caps_add(cname, perm));
+
+  rest_req = "/admin/log?type=data&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=data&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=data&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=data&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=data&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=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()); 
+  
+  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()); 
+  
+  rest_req = "/admin/log?type=data&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=data&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=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()); 
+  utime_t sleep_time(3, 0);
+
+  rest_req = "/admin/log?type=data&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=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(409U, g_test->get_resp_code()); 
+  sleep_time.sleep();
+
+  rest_req = "/admin/log?type=data&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=data&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=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()); 
+  
+  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()); 
+  
+  ASSERT_EQ(0, caps_rm(cname, perm));
+  perm = "write";
+  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()); 
+  
+  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()); 
+  
+  ASSERT_EQ(0, caps_rm(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(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(403U, g_test->get_resp_code()); 
+  
+  ASSERT_EQ(0, user_rm(uid, display_name));
+}
+
+TEST(TestRGWAdmin, datalog_trim) {
+  string start_time, 
+         end_time;
+  const char *cname = "datalog",
+             *perm = "*";
+  string rest_req;
+  unsigned shard_id = get_datalog_shard_id(TEST_BUCKET_NAME, g_ceph_context->_conf->rgw_data_log_num_shards);
+  stringstream ss;
+  list<rgw_data_change> entries;
+
+  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=data";
+  g_test->send_request(string("DELETE"), rest_req);
+  EXPECT_EQ(400U, g_test->get_resp_code()); /*Bad request*/
+  
+  ss.str("");
+  ss << "/admin/log?type=data&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*/
+ 
+  ss.str("");
+  ss << "/admin/log?type=data&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*/
+  
+  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);
+  ASSERT_EQ(0, delete_obj(TEST_BUCKET_OBJECT));
+  sleep(1);
+  EXPECT_EQ(put_bucket_obj(TEST_BUCKET_OBJECT, bucket_obj, TEST_BUCKET_OBJECT_SIZE), 0);
+  ASSERT_EQ(0, delete_obj(TEST_BUCKET_OBJECT));
+  sleep(20);
+  free(bucket_obj);
+
+  ASSERT_EQ(get_formatted_time(end_time), 0);
+  ss.str("");
+  ss << "/admin/log?type=data&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();
+  get_datalog_list(entries);
+  EXPECT_TRUE(entries.size() > 0);
+
+  ss.str("");
+  ss << "/admin/log?type=data&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());
+
+  ss.str("");
+  ss << "/admin/log?type=data&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();
+  get_datalog_list(entries);
+  EXPECT_TRUE(entries.size() == 0);
+
+  ASSERT_EQ(0, caps_rm(cname, perm));
+  perm = "write";
+  ASSERT_EQ(0, caps_add(cname, perm));
+  ss.str("");
+  ss << "/admin/log?type=data&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());
+
+  ASSERT_EQ(0, caps_rm(cname, perm));
+  perm = "";
+  ASSERT_EQ(0, caps_add(cname, perm));
+  ss.str("");
+  ss << "/admin/log?type=data&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(403U, g_test->get_resp_code());
+
+  ASSERT_EQ(0, delete_bucket());
+  ASSERT_EQ(0, user_rm(uid, display_name));
+}
+
 TEST(TestRGWAdmin, mdlog_list) {
   string start_time, 
          end_time,
@@ -682,9 +1030,10 @@ 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);
+  unsigned shard_id = get_mdlog_shard_id(uid, g_ceph_context->_conf->rgw_md_log_max_shards);
   stringstream ss;
 
+  sleep(2);
   ASSERT_EQ(get_formatted_time(start_time), 0);
   ASSERT_EQ(0, user_create(uid, display_name));
   ASSERT_EQ(0, caps_add(cname, perm));
@@ -838,7 +1187,7 @@ 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);
+  unsigned shard_id = get_mdlog_shard_id(uid, g_ceph_context->_conf->rgw_md_log_max_shards);
   ostringstream ss;
 
   sleep(1);