rgw: orphans find: don't process stale bucket instances

As a large bucket might have resharded multiple times, check the cur bucket info
and ensure that no reshard is in progress before we attempt to log bucket index
entries. On a large sized bucket, since a bucket would have undergone reshard
multiple times, this avoids wasteful processing of stale bucket instance entries

Signed-off-by: Abhishek Lekshmanan <abhishek@suse.com>

src/rgw/rgw_orphan.cc

@@ -11,6 +11,7 @@
 #include "rgw_rados.h"
 #include "rgw_orphan.h"
 #include "rgw_zone.h"
+#include "rgw_bucket.h"
 
 #include "services/svc_zone.h"
 #include "services/svc_sys_obj.h"
@@ -485,11 +486,47 @@ int RGWOrphanSearch::pop_and_handle_stat_op(map<int, list<string> >& oids, std::
 
 int RGWOrphanSearch::build_linked_oids_for_bucket(const string& bucket_instance_id, map<int, list<string> >& oids)
 {
-  ldout(store->ctx(), 10) << "building linked oids for bucket instance: " << bucket_instance_id << dendl;
-  RGWBucketInfo bucket_info;
   RGWObjectCtx obj_ctx(store);
   auto sysobj_ctx = store->svc.sysobj->init_obj_ctx();
-  int ret = store->get_bucket_instance_info(sysobj_ctx, bucket_instance_id, bucket_info, NULL, NULL);
+
+  rgw_bucket orphan_bucket;
+  int shard_id;
+  int ret = rgw_bucket_parse_bucket_key(store->ctx(), bucket_instance_id,
+                                        &orphan_bucket, &shard_id);
+  if (ret < 0) {
+    ldout(store->ctx(),0) << __func__ << " failed to parse bucket instance: "
+                 << bucket_instance_id << " skipping" << dendl;
+    return ret;
+  }
+
+  RGWBucketInfo cur_bucket_info;
+  ret = store->get_bucket_info(sysobj_ctx, orphan_bucket.tenant,
+			       orphan_bucket.name, cur_bucket_info, nullptr);
+  if (ret < 0) {
+    if (ret == -ENOENT) {
+      /* probably raced with bucket removal */
+      return 0;
+    }
+    lderr(store->ctx()) << __func__ << ": ERROR: RGWRados::get_bucket_instance_info() returned ret=" << ret << dendl;
+    return ret;
+  }
+
+  if (cur_bucket_info.bucket.bucket_id != orphan_bucket.bucket_id) {
+    ldout(store->ctx(), 0) << __func__ << ": Skipping stale bucket instance: "
+                           << orphan_bucket.name << ": "
+                           << orphan_bucket.bucket_id << dendl;
+    return 0;
+  }
+
+  if (cur_bucket_info.reshard_status == CLS_RGW_RESHARD_IN_PROGRESS) {
+    ldout(store->ctx(), 0) << __func__ << ": reshard in progress. Skipping "
+                           << orphan_bucket.name << ": "
+                           << orphan_bucket.bucket_id << dendl;
+    return 0;
+  }
+
+  RGWBucketInfo bucket_info;
+  ret = store->get_bucket_instance_info(sysobj_ctx, bucket_instance_id, bucket_info, nullptr, nullptr);
   if (ret < 0) {
     if (ret == -ENOENT) {
       /* probably raced with bucket removal */
@@ -499,6 +536,7 @@ int RGWOrphanSearch::build_linked_oids_for_bucket(const string& bucket_instance_
     return ret;
   }
 
+  ldout(store->ctx(), 10) << "building linked oids for bucket instance: " << bucket_instance_id << dendl;
   RGWRados::Bucket target(store, bucket_info);
   RGWRados::Bucket::List list_op(&target);