Be more wary of exceptions in full data sync

Catch more broadly when syncing a bucket, and pass any further uncaught exceptions up through the results queue

Signed-off-by: Josh Durgin <jdurgin@redhat.com>

diff --git a/radosgw_agent/sync.py b/radosgw_agent/sync.py
index 67ebea6b5ab663f9c6412aa514fe3e309cd7301b..18792d4d7aee2f79db2a65d3a5e4abdbeee97a76 100644 (file)
--- a/radosgw_agent/sync.py
+++ b/radosgw_agent/sync.py
@@ -157,8 +157,13 @@ class Syncer(object):
                 log.debug('synced item %r successfully', item)
                 self.complete_item(shard_num, retries)
             else:
-                log.error('error syncing shard %d', shard_num)
-                retries.append(shard_num)
+                try:
+                    if isinstance(result, Exception):
+                        raise result
+                except Exception:
+                    log.exception('error syncing shard %d', shard_num)
+                else:
+                    log.error('error syncing shard %d', shard_num)
 
             log.info('%d/%d items processed', i + 1, num_items)
         if retries:

diff --git a/radosgw_agent/worker.py b/radosgw_agent/worker.py
index 40435eaf2a2fca63a49c88bbaf5adfcdd1673954..1a5215576125661e273668ad570c2a73f8aa732e 100644 (file)
--- a/radosgw_agent/worker.py
+++ b/radosgw_agent/worker.py
@@ -364,45 +364,49 @@ class DataWorkerFull(DataWorker):
             objects = client.list_objects_in_bucket(self.src_conn, bucket)
             if not objects:
                 return True
-        except Exception as e:
-            log.error('error preparing for full sync of bucket "%s": %s',
-                      bucket, e)
-            return False
 
-        retries = self.sync_bucket(bucket, objects)
+            retries = self.sync_bucket(bucket, objects)
 
-        result = self.set_bound(instance, marker, retries, 'bucket-index')
-        return not retries and result == RESULT_SUCCESS
+            result = self.set_bound(instance, marker, retries, 'bucket-index')
+            return not retries and result == RESULT_SUCCESS
+        except Exception as e:
+            log.exception('error preparing for full sync of bucket "%s"',
+                          bucket)
+            return False
 
     def run(self):
         self.prepare_lock()
         while True:
-            item = self.work_queue.get()
-            if item is None:
-                log.info('No more entries in queue, exiting')
-                break
-
-            shard_num, buckets = item
-
-            # first, lock the log
             try:
-                self.lock_shard(shard_num)
-            except SkipShard:
-                continue
+                item = self.work_queue.get()
+                if item is None:
+                    log.info('No more entries in queue, exiting')
+                    break
 
-            # attempt to sync each bucket, add to a list to retry
-            # during incremental sync if sync fails
-            retry_buckets = []
-            for bucket in buckets:
-                if not self.full_sync_bucket(bucket):
-                    retry_buckets.append(bucket)
+                shard_num, buckets = item
 
-            # unlock shard and report buckets to retry during incremental sync
-            self.unlock_shard()
-            self.result_queue.put((RESULT_SUCCESS, (shard_num, retry_buckets)))
-            log.info('finished syncing shard %d', shard_num)
-            log.info('incremental sync will need to retry buckets: %s',
-                     retry_buckets)
+                # first, lock the log
+                try:
+                    self.lock_shard(shard_num)
+                except SkipShard:
+                    continue
+
+                # attempt to sync each bucket, add to a list to retry
+                # during incremental sync if sync fails
+                retry_buckets = []
+                for bucket in buckets:
+                    if not self.full_sync_bucket(bucket):
+                        retry_buckets.append(bucket)
+
+                # unlock shard and report buckets to retry during incremental sync
+                self.unlock_shard()
+                self.result_queue.put((RESULT_SUCCESS, (shard_num, retry_buckets)))
+                log.info('finished syncing shard %d', shard_num)
+                log.info('incremental sync will need to retry buckets: %s',
+                         retry_buckets)
+            except Exception as e:
+                log.exception('exception syncing shard %d', shard_num)
+                self.result_queue.put((e, (shard_num, [])))
 
 class MetadataWorker(Worker):