Merge branch 'tickets/DM-51389'

Author: kfindeisen

python/activator/activator.py

@@ -89,6 +89,8 @@
 bucket_notification_kafka_offset_reset = os.environ.get("BUCKET_NOTIFICATION_KAFKA_OFFSET_RESET", "latest")
 # Max requests to handle before restarting. 0 means no limit.
 max_requests = int(os.environ.get("WORKER_RESTART_FREQ", 0))
+# The number of seconds to delay retrying connections to the Redis stream.
+redis_retry = float(os.environ.get("REDIS_RETRY_DELAY", 30))
 
 # Conditionally load keda environment variables
 if platform == "keda":
@@ -234,7 +236,12 @@ def _make_redis_streams_client(self):
         redis_client : `redis.Redis`
             Initialized Redis client.
         """
-        return redis.Redis(host=self.host)
+        policy = redis.retry.Retry(redis.backoff.ConstantBackoff(redis_retry),
+                                   1,
+                                   # Bare ConnectionError covers things like DNS problems
+                                   (redis.exceptions.ConnectionError, ),
+                                   )
+        return redis.Redis(host=self.host, retry=policy)
 
     @staticmethod
     def _close_on_error(func):

python/activator/middleware_interface.py

@@ -84,7 +84,13 @@
 repo_retry = float(os.environ.get("REPO_RETRY_DELAY", 30))
 
 
-@connect.retry(2, sqlalchemy.exc.OperationalError, wait=repo_retry)
+# TODO: revisit which cases should be retried after DM-50934
+# TODO: catch ButlerConnectionError once it's available
+SQL_EXCEPTIONS = (sqlalchemy.exc.OperationalError, sqlalchemy.exc.InterfaceError)
+DATASTORE_EXCEPTIONS = SQL_EXCEPTIONS + (botocore.exceptions.ClientError, )
+
+
+@connect.retry(2, SQL_EXCEPTIONS, wait=repo_retry)
 def get_central_butler(central_repo: str, instrument_class: str):
     """Provide a Butler that can access the given repository and read and write
     data for the given instrument.
@@ -360,7 +366,7 @@ def _init_visit_definer(self):
         define_visits_config.groupExposures = "one-to-one"
         self.define_visits = lsst.obs.base.DefineVisitsTask(config=define_visits_config, butler=self.butler)
 
-    @connect.retry(2, (sqlalchemy.exc.OperationalError, botocore.exceptions.ClientError), wait=repo_retry)
+    @connect.retry(2, DATASTORE_EXCEPTIONS, wait=repo_retry)
     def _init_governor_datasets(self, timestamp, skymap):
         """Load and store the camera and skymap for later use.
 
@@ -537,7 +543,7 @@ def prep_butler(self) -> None:
                         detector=self.visit.detector,
                         group=self.visit.groupId)
 
-    @connect.retry(2, sqlalchemy.exc.OperationalError, wait=repo_retry)
+    @connect.retry(2, SQL_EXCEPTIONS, wait=repo_retry)
     def _find_data_to_preload(self, region):
         """Identify the datasets to export from the central repo.
 
@@ -912,7 +918,7 @@ def _find_init_outputs(self):
             _log.debug("Found %d new init-output datasets from %s.", n_datasets, run)
         return datasets
 
-    @connect.retry(2, (sqlalchemy.exc.OperationalError, botocore.exceptions.ClientError), wait=repo_retry)
+    @connect.retry(2, DATASTORE_EXCEPTIONS, wait=repo_retry)
     def _transfer_data(self, datasets, calibs):
         """Transfer datasets and all associated collections from the central
         repo to the local repo.
@@ -1602,7 +1608,7 @@ def _get_safe_dataset_types(butler):
         return [dstype.name for dstype in butler.registry.queryDatasetTypes(...)
                 if "detector" in dstype.dimensions]
 
-    @connect.retry(2, (sqlalchemy.exc.OperationalError, botocore.exceptions.ClientError), wait=repo_retry)
+    @connect.retry(2, DATASTORE_EXCEPTIONS, wait=repo_retry)
     def _export_subset(self, exposure_ids: set[int],
                        dataset_types: typing.Any, in_collections: typing.Any) -> None:
         """Copy datasets associated with a processing run back to the

python/initializer/write_init_outputs.py

@@ -49,6 +49,10 @@
 repo_retry = float(os.environ.get("REPO_RETRY_DELAY", 30))
 
 
+SQL_EXCEPTIONS = (sqlalchemy.exc.OperationalError, sqlalchemy.exc.InterfaceError)
+DATASTORE_EXCEPTIONS = SQL_EXCEPTIONS + (botocore.exceptions.ClientError, )
+
+
 def _config_from_yaml(yaml_string):
     """Initialize a PipelinesConfig from a YAML-formatted string.
 
@@ -79,7 +83,7 @@ def make_parser():
     return parser
 
 
-@connect_utils.retry(2, sqlalchemy.exc.OperationalError, wait=repo_retry)
+@connect_utils.retry(2, SQL_EXCEPTIONS, wait=repo_retry)
 def _connect_butler(repo):
     """Connect to a particular repo.
 
@@ -146,7 +150,7 @@ def _get_current_day_obs() -> str:
     return run_utils.get_day_obs(astropy.time.Time.now())
 
 
-@connect_utils.retry(2, (sqlalchemy.exc.OperationalError, botocore.exceptions.ClientError), wait=repo_retry)
+@connect_utils.retry(2, DATASTORE_EXCEPTIONS, wait=repo_retry)
 def _make_init_outputs(base_butler: lsst.daf.butler.Butler,
                        instrument: lsst.obs.base.Instrument,
                        apdb: str,
@@ -200,7 +204,7 @@ def _make_init_outputs(base_butler: lsst.daf.butler.Butler,
     return run
 
 
-@connect_utils.retry(2, sqlalchemy.exc.OperationalError, wait=repo_retry)
+@connect_utils.retry(2, SQL_EXCEPTIONS, wait=repo_retry)
 def _make_output_chain(butler: lsst.daf.butler.Butler,
                        instrument: lsst.obs.base.Instrument,
                        runs: collections.abc.Sequence[str],