rgw/multisite: track shard sync status objects per generation

Signed-off-by: Yuval Lifshitz <ylifshit@redhat.com>

diff --git a/src/rgw/rgw_admin.cc b/src/rgw/rgw_admin.cc
index af5fbfaeecf9b401f68f8ea5e1542fe19dd14e9a..bc38d592d78ed7860869eb19d61cdfb107ef2790 100644 (file)
--- a/src/rgw/rgw_admin.cc
+++ b/src/rgw/rgw_admin.cc
@@ -2307,7 +2307,7 @@ static int bucket_source_sync_status(const DoutPrefixProvider *dpp, rgw::sal::Ra
   }
 
   std::vector<rgw_bucket_shard_sync_info> status;
-  r = rgw_read_bucket_inc_sync_status(dpp, store, pipe, bucket_info, &source_bucket->get_info(), &status);
+  r = rgw_read_bucket_inc_sync_status(dpp, store, pipe, bucket_info, &source_bucket->get_info(), full_status.incremental_gen,  &status);
   if (r < 0) {
     lderr(store->ctx()) << "failed to read bucket incremental sync status: " << cpp_strerror(r) << dendl;
     return r;
@@ -8304,6 +8304,7 @@ next:
       cerr << "ERROR: sync.init() returned ret=" << ret << std::endl;
       return -ret;
     }
+
     ret = sync.read_sync_status(dpp());
     if (ret < 0) {
       cerr << "ERROR: sync.read_sync_status() returned ret=" << ret << std::endl;

diff --git a/src/rgw/rgw_data_sync.cc b/src/rgw/rgw_data_sync.cc
index 84e9bf0752cda8349c3fdd1d10ed7e632d5fc6ca..8295b1a59744e00e7e48865a15a24628ec01a1eb 100644 (file)
--- a/src/rgw/rgw_data_sync.cc
+++ b/src/rgw/rgw_data_sync.cc
@@ -2768,7 +2768,7 @@ public:
                                         bool exclusive)
     : RGWCoroutine(_sc->cct), sc(_sc), sync_env(_sc->env),
       sync_pair(_sync_pair),
-      sync_status_oid(RGWBucketPipeSyncStatusManager::inc_status_oid(sc->source_zone, _sync_pair)),
+      sync_status_oid(RGWBucketPipeSyncStatusManager::inc_status_oid(sc->source_zone, _sync_pair, _info.latest_gen)),
       status(_status), objv_tracker(objv_tracker), info(_info), marker_mgr(_marker_mgr), exclusive(exclusive)
   {}
 
@@ -2915,9 +2915,10 @@ public:
   RGWReadBucketPipeSyncStatusCoroutine(RGWDataSyncCtx *_sc,
                                    const rgw_bucket_sync_pair_info& sync_pair,
                                    rgw_bucket_shard_sync_info *_status,
-                                   RGWObjVersionTracker* objv_tracker)
+                                   RGWObjVersionTracker* objv_tracker,
+                                   uint64_t gen)
     : RGWCoroutine(_sc->cct), sc(_sc), sync_env(_sc->env),
-      oid(RGWBucketPipeSyncStatusManager::inc_status_oid(sc->source_zone, sync_pair)),
+      oid(RGWBucketPipeSyncStatusManager::inc_status_oid(sc->source_zone, sync_pair, gen)),
       status(_status), objv_tracker(objv_tracker)
   {}
   int operate(const DoutPrefixProvider *dpp) override;
@@ -2951,7 +2952,7 @@ class CheckBucketShardStatusIsIncremental : public RGWReadBucketPipeSyncStatusCo
   CheckBucketShardStatusIsIncremental(RGWDataSyncCtx* sc,
                                       const rgw_bucket_sync_pair_info& sync_pair,
                                       bool* result)
-    : RGWReadBucketPipeSyncStatusCoroutine(sc, sync_pair, &status, nullptr),
+    : RGWReadBucketPipeSyncStatusCoroutine(sc, sync_pair, &status, nullptr, 0 /*no gen in compat mode*/),
       result(result)
   {}
 
@@ -3045,7 +3046,7 @@ class InitBucketShardStatusCR : public RGWCoroutine {
       while (--tries) {
         objv.clear();
         // read current status and objv
-        yield call(new ReadCR(sc, pair, &status, &objv));
+        yield call(new ReadCR(sc, pair, &status, &objv, info.latest_gen));
         if (retcode < 0) {
           return set_cr_error(retcode);
         }
@@ -3384,7 +3385,7 @@ RGWCoroutine *RGWRemoteBucketManager::read_sync_status_cr(int num, rgw_bucket_sh
     return nullptr;
   }
 
-  return new RGWReadBucketPipeSyncStatusCoroutine(&sc, sync_pairs[num], sync_status, nullptr);
+  return new RGWReadBucketPipeSyncStatusCoroutine(&sc, sync_pairs[num], sync_status, nullptr, full_status.incremental_gen);
 }
 
 RGWBucketPipeSyncStatusManager::RGWBucketPipeSyncStatusManager(rgw::sal::RadosStore* _store,
@@ -5077,7 +5078,7 @@ public:
     : RGWCoroutine(_sc->cct), sc(_sc), sync_env(_sc->env),
       lease_cr(std::move(lease_cr)), sync_pair(_sync_pair),
       sync_pipe(sync_pipe), bucket_state(bucket_state), generation(generation), progress(progress),
-      shard_status_oid(RGWBucketPipeSyncStatusManager::inc_status_oid(sc->source_zone, sync_pair)),
+      shard_status_oid(RGWBucketPipeSyncStatusManager::inc_status_oid(sc->source_zone, sync_pair, generation)),
       bucket_status_obj(sc->env->svc->zone->get_zone_params().log_pool,
                  RGWBucketPipeSyncStatusManager::full_status_oid(sc->source_zone,
                                                                  sync_pair.source_bs.bucket,
@@ -5091,7 +5092,7 @@ public:
 int RGWSyncBucketShardCR::operate(const DoutPrefixProvider *dpp)
 {
   reenter(this) {
-    yield call(new RGWReadBucketPipeSyncStatusCoroutine(sc, sync_pair, &sync_status, &objv_tracker));
+    yield call(new RGWReadBucketPipeSyncStatusCoroutine(sc, sync_pair, &sync_status, &objv_tracker, generation));
     if (retcode < 0 && retcode != -ENOENT) {
       tn->log(0, "ERROR: failed to read sync status for bucket");
       return set_cr_error(retcode);
@@ -5477,13 +5478,20 @@ string RGWBucketPipeSyncStatusManager::full_status_oid(const rgw_zone_id& source
   }
 }
 
+inline std::string generation_token(uint64_t gen) {
+  return (gen == 0) ? "" : (":" + std::to_string(gen));
+}
+
 string RGWBucketPipeSyncStatusManager::inc_status_oid(const rgw_zone_id& source_zone,
-                                                      const rgw_bucket_sync_pair_info& sync_pair)
+                                                      const rgw_bucket_sync_pair_info& sync_pair,
+                                                      uint64_t gen)
 {
   if (sync_pair.source_bs.bucket == sync_pair.dest_bucket) {
-    return bucket_status_oid_prefix + "." + source_zone.id + ":" + sync_pair.source_bs.get_key();
+    return bucket_status_oid_prefix + "." + source_zone.id + ":" + sync_pair.source_bs.get_key() + 
+      generation_token(gen);
   } else {
-    return bucket_status_oid_prefix + "." + source_zone.id + ":" + sync_pair.dest_bucket.get_key() + ":" + sync_pair.source_bs.get_key();
+    return bucket_status_oid_prefix + "." + source_zone.id + ":" + sync_pair.dest_bucket.get_key() + ":" + sync_pair.source_bs.get_key() +
+      generation_token(gen);
   }
 }
 
@@ -5532,6 +5540,8 @@ class RGWCollectBucketSyncStatusCR : public RGWShardCollectCR {
   rgw::sal::RadosStore* const store;
   RGWDataSyncCtx *const sc;
   RGWDataSyncEnv *const env;
+  const uint64_t gen;
+
   rgw_bucket_sync_pair_info sync_pair;
   using Vector = std::vector<rgw_bucket_shard_sync_info>;
   Vector::iterator i, end;
@@ -5550,9 +5560,10 @@ class RGWCollectBucketSyncStatusCR : public RGWShardCollectCR {
   RGWCollectBucketSyncStatusCR(rgw::sal::RadosStore* store, RGWDataSyncCtx *sc,
                                const RGWBucketInfo& source_bucket_info,
                                const RGWBucketInfo& dest_bucket_info,
+                               uint64_t gen,
                                Vector *status)
     : RGWShardCollectCR(sc->cct, max_concurrent_shards),
-      store(store), sc(sc), env(sc->env),
+      store(store), sc(sc), env(sc->env), gen(gen),
       i(status->begin()), end(status->end())
   {
     sync_pair.source_bs = rgw_bucket_shard(source_bucket_info.bucket, source_bucket_info.layout.current_index.layout.normal.num_shards > 0 ? 0 : -1);
@@ -5563,7 +5574,7 @@ class RGWCollectBucketSyncStatusCR : public RGWShardCollectCR {
     if (i == end) {
       return false;
     }
-    spawn(new RGWReadBucketPipeSyncStatusCoroutine(sc, sync_pair, &*i, nullptr), false);
+    spawn(new RGWReadBucketPipeSyncStatusCoroutine(sc, sync_pair, &*i, nullptr, gen), false);
     ++i;
     ++sync_pair.source_bs.shard_id;
     return true;
@@ -5606,6 +5617,7 @@ int rgw_read_bucket_inc_sync_status(const DoutPrefixProvider *dpp,
                                     const rgw_sync_bucket_pipe& pipe,
                                     const RGWBucketInfo& dest_bucket_info,
                                     const RGWBucketInfo *psource_bucket_info,
+                                    uint64_t gen,
                                     std::vector<rgw_bucket_shard_sync_info> *status)
 {
   if (!pipe.source.zone ||
@@ -5649,5 +5661,6 @@ int rgw_read_bucket_inc_sync_status(const DoutPrefixProvider *dpp,
   return crs.run(dpp, new RGWCollectBucketSyncStatusCR(store, &sc,
                                                   *psource_bucket_info,
                                                   dest_bucket_info,
+                                                  gen,
                                                   status));
 }

diff --git a/src/rgw/rgw_data_sync.h b/src/rgw/rgw_data_sync.h
index 538a692d5b2336df5b3a163e6bff63031e9b579c..e1b5b2d9cb0a5680a7b937477fc1a0eeaea4742f 100644 (file)
--- a/src/rgw/rgw_data_sync.h
+++ b/src/rgw/rgw_data_sync.h
@@ -749,7 +749,8 @@ public:
                                     const rgw_bucket& source_bucket,
                                     const rgw_bucket& dest_bucket);
   static std::string inc_status_oid(const rgw_zone_id& source_zone,
-                                   const rgw_bucket_sync_pair_info& bs);
+                                   const rgw_bucket_sync_pair_info& bs,
+                                   uint64_t gen);
   // specific source obj sync status, can be used by sync modules
   static std::string obj_status_oid(const rgw_bucket_sync_pipe& sync_pipe,
                                const rgw_zone_id& source_zone, const rgw::sal::Object* obj); /* specific source obj sync status,
@@ -777,6 +778,7 @@ int rgw_read_bucket_inc_sync_status(const DoutPrefixProvider *dpp,
                                     const rgw_sync_bucket_pipe& pipe,
                                     const RGWBucketInfo& dest_bucket_info,
                                     const RGWBucketInfo *psource_bucket_info,
+                                    uint64_t gen,
                                     std::vector<rgw_bucket_shard_sync_info> *status);
 
 class RGWDefaultSyncModule : public RGWSyncModule {

diff --git a/src/rgw/rgw_rest_log.cc b/src/rgw/rgw_rest_log.cc
index 891003533491a0268ab885cfc23d7cfed94c985d..58b6e81c6a1c56bcd18c1887896972aa6b5e1f3f 100644 (file)
--- a/src/rgw/rgw_rest_log.cc
+++ b/src/rgw/rgw_rest_log.cc
@@ -1015,6 +1015,7 @@ void RGWOp_BILog_Status::execute(optional_yield y)
       pipe,
       bucket->get_info(),
       nullptr,
+      status.sync_status.incremental_gen,
       &status.inc_status);
     if (op_ret < 0) {
       ldpp_dout(this, -1) << "ERROR: rgw_read_bucket_inc_sync_status() on pipe=" << pipe << " returned ret=" << op_ret << dendl;
@@ -1076,7 +1077,7 @@ void RGWOp_BILog_Status::execute(optional_yield y)
       }
     }
     int r = rgw_read_bucket_inc_sync_status(this, static_cast<rgw::sal::RadosStore*>(store),
-                                           pipe, *pinfo, &bucket->get_info(), &current_status);
+                                           pipe, *pinfo, &bucket->get_info(), status.sync_status.incremental_gen, &current_status);
     if (r < 0) {
       ldpp_dout(this, -1) << "ERROR: rgw_read_bucket_inc_sync_status() on pipe=" << pipe << " returned ret=" << r << dendl;
       op_ret = r;

diff --git a/src/rgw/rgw_sync_checkpoint.cc b/src/rgw/rgw_sync_checkpoint.cc
index 0c581aae14153dd7688d1a2f634f999905cba002..3835f5a7990e00321b81579a566b572f06ce57c5 100644 (file)
--- a/src/rgw/rgw_sync_checkpoint.cc
+++ b/src/rgw/rgw_sync_checkpoint.cc
@@ -145,7 +145,7 @@ int bucket_source_sync_checkpoint(const DoutPrefixProvider* dpp,
   std::vector<rgw_bucket_shard_sync_info> status;
   status.resize(std::max<size_t>(1, num_shards));
   r = rgw_read_bucket_inc_sync_status(dpp, store, pipe, bucket_info,
-                                      &source_bucket_info, &status);
+                                      &source_bucket_info, full_status.incremental_gen, &status);
   if (r < 0) {
     return r;
   }
@@ -161,7 +161,7 @@ int bucket_source_sync_checkpoint(const DoutPrefixProvider* dpp,
         << "    remote markers: " << remote_markers << dendl;
     std::this_thread::sleep_until(delay_until);
     r = rgw_read_bucket_inc_sync_status(dpp, store, pipe, bucket_info,
-                                        &source_bucket_info, &status);
+                                        &source_bucket_info, full_status.incremental_gen, &status);
     if (r < 0) {
       return r;
     }