Improve handling of fail queues (#6781)

Author: achave11-ucsc

scripts/reindex.py

@@ -7,6 +7,7 @@
 import sys
 
 from azul import (
+    R,
     config,
     reject,
 )
@@ -143,6 +144,10 @@ def main(argv: list[str]):
     reject(args.deindex and (args.delete or args.create),
            '--deindex is incompatible with --create and --delete.')
 
+    assert azul.is_valid_failed_queues_state(), R(
+        'Operation canceled due to non-empty failed queues. Running manage_queues.py is '
+        'required to empty the queues.')
+
     deindex = args.deindex or (args.delete and not every_source)
     delete = args.delete and every_source
 
@@ -184,6 +189,9 @@ def main(argv: list[str]):
                             args.prefix, args.catalogs)
             else:
                 azul.wait_for_indexer()
+                assert azul.is_valid_failed_queues_state(), R(
+                    'Operation failed due to non empty failed queues. Use manage_queues.py to '
+                    'amend this state before retrying.')
 
 
 if __name__ == '__main__':

src/azul/azulclient.py

@@ -600,6 +600,14 @@ def message(source: SourceRef):
         messages = map(message, sources)
         self.queue_mirror_messages(messages)
 
+    def is_valid_failed_queues_state(self) -> bool:
+        queues = set()
+        for failed_queue in config.fail_queue_names:
+            log.info('Checking whether queue %r is empty', failed_queue)
+            queues.add(self.is_queue_empty(failed_queue))
+        assert len(queues) > 0, R('Failed queues not checked', queues)
+        return all(queues)
+
 
 class AzulClientError(RuntimeError):
     pass

test/integration_test.py

@@ -83,6 +83,7 @@
 from azul import (
     CatalogName,
     Config,
+    R,
     RequirementError,
     cache,
     cached_property,
@@ -429,6 +430,7 @@ class Catalog:
 
         def _wait_for_indexer():
             self.azul_client.wait_for_indexer()
+            self._empty_failed_queues_check()
 
         flags = config.it_flags
         index, delete = ['no_' + flag not in flags for flag in ['index', 'delete']]
@@ -438,6 +440,7 @@ def _wait_for_indexer():
         else:
             log.warning('Will skip indexing due to overriding IT flag.')
 
+        self._empty_failed_queues_check()
         catalogs: list[Catalog] = []
         for catalog in config.integration_test_catalogs:
             if index:
@@ -495,6 +498,7 @@ def _wait_for_indexer():
             self._test_mirroring()
 
     def _reset_indexer(self):
+        self._empty_failed_queues_check()
         # While it's OK to erase the integration test catalog, the queues are
         # shared by all catalogs and we can't afford to trash them in a stable
         # deployment like production.
@@ -505,6 +509,11 @@ def _reset_indexer(self):
                                        delete_indices=True,
                                        create_indices=True)
 
+    def _empty_failed_queues_check(self):
+        assert self.azul_client.is_valid_failed_queues_state(), R(
+            'Failed queues are not empty. Use manage_queues.py to amend this state '
+            'before retrying.')
+
     def _test_other_endpoints(self):
         catalog = config.default_catalog
         if config.is_hca_enabled(catalog):