Merge pull request ceph#64933 from soumyakoduri/wip-skoduri-restore-progress

rgw/restore: Persistently store the restore state for cloud-s3 tier

Reviewed-by: Jiffin Tony Thottan <[email protected]>
Reviewed-by: Matt Benjamin <[email protected]>

Author: soumyakoduri

qa/suites/rgw/cloud-transition/overrides.yaml

@@ -5,13 +5,15 @@ overrides:
         setuser: ceph
         setgroup: ceph
         debug rgw: 20
+        debug rgw_restore: 20
         rgw crypt s3 kms backend: testing
         rgw crypt s3 kms encryption keys: testkey-1=YmluCmJvb3N0CmJvb3N0LWJ1aWxkCmNlcGguY29uZgo= testkey-2=aWIKTWFrZWZpbGUKbWFuCm91dApzcmMKVGVzdGluZwo=
         rgw crypt require ssl: false
         rgw sts key: abcdefghijklmnop
         rgw s3 auth use sts: true
         rgw lc debug interval: 10
         rgw_restore_debug_interval: 20
+        rgw_restore_processor_period: 10
   rgw:
     storage classes:
       LUKEWARM:

qa/suites/rgw/cloud-transition/tasks/restore/cloud_restore_s3tests.yaml

@@ -7,6 +7,7 @@ tasks:
     client.1:
       port: 8001
 - rgw-cloudtier:
+    client.1:
     client.0:
       cloud_storage_class: CLOUDTIER-CLIENT0
       cloud_client: client.1
@@ -27,5 +28,6 @@ tasks:
       extra_attrs: ["cloud_restore"]
       lc_debug_interval: 10
       rgw_restore_debug_interval: 20
+      rgw_restore_processor_period: 10
       lifecycle_tests: True
       cloudtier_tests: True

qa/suites/rgw/cloud-transition/tasks/transition/cloud_transition_s3tests.yaml

@@ -11,6 +11,7 @@ tasks:
     #client.3:
       #port: 8003
 - rgw-cloudtier:
+    client.1:
     client.0:
       # cloudtier storage class params
       # retain_head_object = false

qa/tasks/rgw_cloudtier.py

@@ -54,8 +54,6 @@ def setup(self):
         clients = self.config.keys() # http://tracker.ceph.com/issues/20417
         for client in clients:
             client_config = self.config.get(client)
-            if client_config is None:
-                client_config = {}
 
             if client_config is not None:
                 log.info('client %s - cloudtier config is -----------------%s ', client, client_config)
@@ -118,13 +116,14 @@ def setup(self):
 
                 log.info('Finished Configuring rgw cloudtier ...')
 
-                cluster_name, daemon_type, client_id = teuthology.split_role(client)
-                client_with_id = daemon_type + '.' + client_id
-                self.ctx.daemons.get_daemon('rgw', client_with_id, cluster_name).restart()
-                log.info('restarted rgw daemon ...')
+        for client in clients:
+            cluster_name, daemon_type, client_id = teuthology.split_role(client)
+            client_with_id = daemon_type + '.' + client_id
+            self.ctx.daemons.get_daemon('rgw', client_with_id, cluster_name).restart()
+            log.info('restarted rgw daemon ...')
 
-                (remote,) = self.ctx.cluster.only(client).remotes.keys()
-                wait_for_radosgw(endpoint.url(), remote)
+            (remote,) = self.ctx.cluster.only(client).remotes.keys()
+            wait_for_radosgw(endpoint.url(), remote)
 
 
 task = RGWCloudTier

qa/tasks/s3tests.py

@@ -372,6 +372,10 @@ def configure(ctx, config):
         if rgw_restore_debug_interval:
             s3tests_conf['s3 main']['rgw_restore_debug_interval'] = rgw_restore_debug_interval
 
+        rgw_restore_processor_period = properties.get('rgw_restore_processor_period')
+        if rgw_restore_processor_period:
+            s3tests_conf['s3 main']['rgw_restore_processor_period'] = rgw_restore_processor_period
+
         if ctx.rgw_cloudtier is not None:
             log.info(' ctx.rgw_cloudtier config  is %s ...', ctx.rgw_cloudtier.config)
             client_rgw_config = ctx.rgw_cloudtier.config.get(client)

src/rgw/driver/rados/rgw_rados.cc

@@ -5554,6 +5554,7 @@ int RGWRados::restore_obj_from_cloud(RGWLCCloudTierCtx& tier_ctx,
     ret = rgw_cloud_tier_get_object(tier_ctx, false,  headers,
                                 &set_mtime, etag, accounted_size,
                                 attrs, &cb);
+    in_progress = false;
   }
 
   if (ret < 0) { 

src/rgw/rgw_restore.cc

@@ -341,13 +341,14 @@ int Restore::process(int index, int max_secs, optional_yield y)
       ret = process_restore_entry(entry, y);
 
       if (!ret && entry.status == rgw::sal::RGWRestoreStatus::RestoreAlreadyInProgress) {
-	 r_entries.push_back(entry);
+      	 r_entries.push_back(entry);
          ldpp_dout(this, 20) << __PRETTY_FUNCTION__ << ": re-pushing entry: '" << entry
 		 	 << "' on shard:"
   	  	         << obj_names[index] << dendl;	 
       }
 
-      if (ret < 0)
+      // Skip the entry of object/bucket which no longer exists
+      if (ret < 0 && (ret != -ENOENT))
         goto done;
 
       ///process all entries, trim and re-add
@@ -391,14 +392,16 @@ int Restore::process(int index, int max_secs, optional_yield y)
 int Restore::process_restore_entry(RestoreEntry& entry, optional_yield y)
 {
   int ret = 0;
-  bool in_progress = true;
   std::unique_ptr<rgw::sal::Bucket> bucket;
   std::unique_ptr<rgw::sal::Object> obj;
   std::unique_ptr<rgw::sal::PlacementTier> tier;
   std::optional<uint64_t> days = entry.days;
   rgw::sal::RGWRestoreStatus restore_status = rgw::sal::RGWRestoreStatus::None;
   rgw_placement_rule target_placement;
 
+  // mark in_progress as false if the entry is being processed first time
+  bool in_progress = ((entry.status == rgw::sal::RGWRestoreStatus::None) ? false : true);
+
   // Ensure its the same source zone processing temp entries as we do not
   // replicate temp restored copies
   if (days) { // temp copy
@@ -429,6 +432,7 @@ int Restore::process_restore_entry(RestoreEntry& entry, optional_yield y)
     return ret;
   }
 
+  ldpp_dout(this, 10) << "Restore:: Processing restore entry of object(" << obj->get_key() << ") entry: " << entry << dendl;
   target_placement.inherit_from(bucket->get_placement_rule());
 
   auto& attrs = obj->get_attrs();
@@ -439,6 +443,7 @@ int Restore::process_restore_entry(RestoreEntry& entry, optional_yield y)
     using ceph::decode;
     decode(restore_status, iter);
   }
+  // check if its still in Progress state
   if (restore_status != rgw::sal::RGWRestoreStatus::RestoreAlreadyInProgress) {
     ldpp_dout(this, 5) << __PRETTY_FUNCTION__ << ": Restore of object " << obj->get_key()
 	   	       << " not in progress state" << dendl;
@@ -450,11 +455,13 @@ int Restore::process_restore_entry(RestoreEntry& entry, optional_yield y)
   attr_iter = attrs.find(RGW_ATTR_STORAGE_CLASS);
   if (attr_iter != attrs.end()) {
     target_placement.storage_class = attr_iter->second.to_str();
+  } else {
+    ldpp_dout(this, -1) << __PRETTY_FUNCTION__ << ": ERROR: Attr RGW_ATTR_STORAGE_CLASS not found for object: " << obj->get_key() << dendl;
   }
   ret = driver->get_zone()->get_zonegroup().get_placement_tier(target_placement, &tier);
 
   if (ret < 0) {
-    ldpp_dout(this, -1) << __PRETTY_FUNCTION__ << ": ERROR: failed to fetch tier placement handle, ret = " << ret << dendl;
+    ldpp_dout(this, -1) << __PRETTY_FUNCTION__ << ": ERROR: failed to fetch tier placement handle, target_placement = " << target_placement << ", for zonegroup = " << driver->get_zone()->get_zonegroup().get_name() << ", ret = " << ret << dendl;
     goto done;	  
   } else {
     ldpp_dout(this, 20) << __PRETTY_FUNCTION__ << ": getting tier placement handle"
@@ -625,52 +632,37 @@ int Restore::restore_obj_from_cloud(rgw::sal::Bucket* pbucket,
     return ret;
   }
 
-  // now go ahead with restoring object
-  bool in_progress = false;
-  ret = pobj->restore_obj_from_cloud(pbucket, tier, cct, days, in_progress, dpp, y);
+  // now add the entry to the restore list to be processed by Restore worker thread
+  // asynchronoudly
+  RestoreEntry entry;
+  entry.bucket = pbucket->get_key();
+  entry.obj_key = pobj->get_key();
+  // for first time mark status as None
+  entry.status = rgw::sal::RGWRestoreStatus::None;
+  entry.days = days;
+  entry.zone_id = driver->get_zone()->get_id(); 
+ 
+  ldpp_dout(this, 10) << "Restore:: Adding restore entry of object(" << pobj->get_key() << ") entry: " << entry << dendl;
+
+  int index = choose_oid(entry);
+  ldpp_dout(this, 10) << __PRETTY_FUNCTION__ << ": Adding restore entry of object(" << pobj->get_key() << ") entry: " << entry << ", to shard:" << obj_names[index] << dendl;
+
+  std::vector<rgw::restore::RestoreEntry> r_entries;
+  r_entries.push_back(entry);
+  ret = sal_restore->add_entries(this, y, index, r_entries);
 
   if (ret < 0) {
-   ldpp_dout(this, -1) << __PRETTY_FUNCTION__ << ": ERROR: object " << pobj->get_key() << " fetching failed" << ret << dendl;	  
-    auto reset_ret = set_cloud_restore_status(this, pobj, y, rgw::sal::RGWRestoreStatus::RestoreFailed);
+    ldpp_dout(this, -1) << __PRETTY_FUNCTION__ << ": ERROR: Adding restore entry of object(" << pobj->get_key() << ") failed" << ret << dendl;	    
 
+    auto reset_ret = set_cloud_restore_status(this, pobj, y, rgw::sal::RGWRestoreStatus::RestoreFailed);
     if (reset_ret < 0) {
-      ldpp_dout(this, -1) << __PRETTY_FUNCTION__ << ": Setting restore status to RestoreFailed failed for object(" << pobj->get_key() << ") " << reset_ret << dendl;	    
+      ldpp_dout(this, -1) << __PRETTY_FUNCTION__ << ": Setting restore status as RestoreFailed failed for object(" << pobj->get_key() << ") " << reset_ret << dendl;	      
     }
 
     return ret;
   }
 
-  if (in_progress) {
-    // add restore entry to the list
-    RestoreEntry entry;
-    entry.bucket = pbucket->get_key();
-    entry.obj_key = pobj->get_key();
-    entry.status = rgw::sal::RGWRestoreStatus::RestoreAlreadyInProgress;
-    entry.days = days;
-    entry.zone_id = driver->get_zone()->get_id(); 
-
-    ldpp_dout(this, 10) << "Restore:: Adding restore entry of object(" << pobj->get_key() << ") entry: " << entry << dendl;
-
-    int index = choose_oid(entry);
-    ldpp_dout(this, 10) << __PRETTY_FUNCTION__ << ": Adding restore entry of object(" << pobj->get_key() << ") entry: " << entry << ", to shard:" << obj_names[index] << dendl;
-
-    std::vector<rgw::restore::RestoreEntry> r_entries;
-    r_entries.push_back(entry);
-    ret = sal_restore->add_entries(this, y, index, r_entries);
-
-    if (ret < 0) {
-      ldpp_dout(this, -1) << __PRETTY_FUNCTION__ << ": ERROR: Adding restore entry of object(" << pobj->get_key() << ") failed" << ret << dendl;	    
-
-      auto reset_ret = set_cloud_restore_status(this, pobj, y, rgw::sal::RGWRestoreStatus::RestoreFailed);
-      if (reset_ret < 0) {
-        ldpp_dout(this, -1) << __PRETTY_FUNCTION__ << ": Setting restore status as RestoreFailed failed for object(" << pobj->get_key() << ") " << reset_ret << dendl;	      
-      }
-
-      return ret;
-    }
-  }
-
-  ldpp_dout(this, 10) << __PRETTY_FUNCTION__ << ": Restore of object " << pobj->get_key() << (in_progress ? " is in progress" : " succeeded") << dendl;  
+  ldpp_dout(this, 10) << __PRETTY_FUNCTION__ << ": Restore of object " << pobj->get_key() << " is in progress." << dendl;  
   return ret;
 }