rgw: bucket replica log, handle shard ids

bucket replica log can now save entries by the specified shard id

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

diff --git a/src/rgw/rgw_admin.cc b/src/rgw/rgw_admin.cc
index d0f9f5843dc7b34777e3a6d40dea667aabd60c5f..03b51a59685ee018101229254aef00611d3da37c 100644 (file)
--- a/src/rgw/rgw_admin.cc
+++ b/src/rgw/rgw_admin.cc
@@ -2565,7 +2565,7 @@ next:
       }
 
       RGWReplicaBucketLogger logger(store);
-      ret = logger.get_bounds(bucket, bounds);
+      ret = logger.get_bounds(bucket, shard_id, bounds);
       if (ret < 0)
         return -ret;
     } else { // shouldn't get here
@@ -2616,7 +2616,7 @@ next:
       }
 
       RGWReplicaBucketLogger logger(store);
-      ret = logger.delete_bound(bucket, daemon_id);
+      ret = logger.delete_bound(bucket, shard_id, daemon_id);
       if (ret < 0)
         return -ret;
     }

diff --git a/src/rgw/rgw_replica_log.cc b/src/rgw/rgw_replica_log.cc
index 961abc2e188061deec4170b255f699b13e4ba532..04547176041c18183439994648ad7c046d0c34ae 100644 (file)
--- a/src/rgw/rgw_replica_log.cc
+++ b/src/rgw/rgw_replica_log.cc
@@ -132,3 +132,15 @@ RGWReplicaBucketLogger::RGWReplicaBucketLogger(RGWRados *_store) :
   prefix = _store->ctx()->_conf->rgw_replica_log_obj_prefix;
   prefix.append(".");
 }
+
+string RGWReplicaBucketLogger::obj_name(const rgw_bucket& bucket, int shard_id)
+{
+  string s = prefix + bucket.name;
+
+  if (shard_id >= 0) {
+    char buf[16];
+    snprintf(buf, sizeof(buf), ".%d", shard_id);
+    s += buf;
+  }
+  return s;
+}

diff --git a/src/rgw/rgw_replica_log.h b/src/rgw/rgw_replica_log.h
index 456b230a6520e3a3f31d1659b0c281be367f12f3..8fec5f766af1bcba2b82d50e50c2afa2b8b615b5 100644 (file)
--- a/src/rgw/rgw_replica_log.h
+++ b/src/rgw/rgw_replica_log.h
@@ -97,20 +97,22 @@ public:
 class RGWReplicaBucketLogger : private RGWReplicaLogger {
   string pool;
   string prefix;
+
+  string obj_name(const rgw_bucket& bucket, int shard_id);
 public:
   RGWReplicaBucketLogger(RGWRados *_store);
-  int update_bound(const rgw_bucket& bucket, const string& daemon_id,
+  int update_bound(const rgw_bucket& bucket, int shard_id, const string& daemon_id,
                    const string& marker, const utime_t& time,
                    const list<RGWReplicaItemMarker> *entries) {
-    return RGWReplicaLogger::update_bound(prefix+bucket.name, pool,
+    return RGWReplicaLogger::update_bound(obj_name(bucket, shard_id), pool,
                                           daemon_id, marker, time, entries);
   }
-  int delete_bound(const rgw_bucket& bucket, const string& daemon_id) {
-    return RGWReplicaLogger::delete_bound(prefix+bucket.name, pool,
+  int delete_bound(const rgw_bucket& bucket, int shard_id, const string& daemon_id) {
+    return RGWReplicaLogger::delete_bound(obj_name(bucket, shard_id), pool,
                                           daemon_id);
   }
-  int get_bounds(const rgw_bucket& bucket, RGWReplicaBounds& bounds) {
-    return RGWReplicaLogger::get_bounds(prefix+bucket.name, pool,
+  int get_bounds(const rgw_bucket& bucket, int shard_id, RGWReplicaBounds& bounds) {
+    return RGWReplicaLogger::get_bounds(obj_name(bucket, shard_id), pool,
                                         bounds);
   }
 };

diff --git a/src/rgw/rgw_rest_replica_log.cc b/src/rgw/rgw_rest_replica_log.cc
index e7dd962f0f7647bc0a93e2d3b156a56941a336d5..0309a2cabd25b56bed87cb46fa109978acec7195 100644 (file)
--- a/src/rgw/rgw_rest_replica_log.cc
+++ b/src/rgw/rgw_rest_replica_log.cc
@@ -181,6 +181,13 @@ void RGWOp_BILog_SetBounds::execute() {
     return;
   }
 
+  int shard_id;
+  http_ret = rgw_bucket_parse_bucket_instance(bucket_instance, &bucket_instance, &shard_id);
+  if (http_ret < 0) {
+    dout(5) << "failed to parse bucket instance" << dendl;
+    return;
+  }
+
   rgw_bucket bucket;
   if ((http_ret = bucket_instance_to_bucket(store, bucket_instance, bucket)) < 0) 
     return;
@@ -194,7 +201,7 @@ void RGWOp_BILog_SetBounds::execute() {
     return;
   }
 
-  http_ret = rl.update_bound(bucket, daemon_id, marker, ut, &markers);
+  http_ret = rl.update_bound(bucket, shard_id, daemon_id, marker, ut, &markers);
 }
 
 void RGWOp_BILog_GetBounds::execute() {
@@ -206,12 +213,19 @@ void RGWOp_BILog_GetBounds::execute() {
     return;
   }
 
+  int shard_id;
+  http_ret = rgw_bucket_parse_bucket_instance(bucket_instance, &bucket_instance, &shard_id);
+  if (http_ret < 0) {
+    dout(5) << "failed to parse bucket instance" << dendl;
+    return;
+  }
+
   rgw_bucket bucket;
   if ((http_ret = bucket_instance_to_bucket(store, bucket_instance, bucket)) < 0) 
     return;
 
   RGWReplicaBucketLogger rl(store);
-  http_ret = rl.get_bounds(bucket, bounds);
+  http_ret = rl.get_bounds(bucket, shard_id, bounds);
 }
 
 void RGWOp_BILog_GetBounds::send_response() {
@@ -237,12 +251,19 @@ void RGWOp_BILog_DeleteBounds::execute() {
     return;
   }
   
+  int shard_id;
+  http_ret = rgw_bucket_parse_bucket_instance(bucket_instance, &bucket_instance, &shard_id);
+  if (http_ret < 0) {
+    dout(5) << "failed to parse bucket instance" << dendl;
+    return;
+  }
+
   rgw_bucket bucket;
   if ((http_ret = bucket_instance_to_bucket(store, bucket_instance, bucket)) < 0) 
     return;
   
   RGWReplicaBucketLogger rl(store);
-  http_ret = rl.delete_bound(bucket, daemon_id);
+  http_ret = rl.delete_bound(bucket, shard_id, daemon_id);
 }
 
 RGWOp *RGWHandler_ReplicaLog::op_get() {