introduce retry for fetching data source configurations

documentation/docs/getting-started/configuration.mdx

@@ -127,6 +127,7 @@ Please use this table as a reference.
 | POLICY_STORE_AUTH_OAUTH_CLIENT_ID     | The client id OPAL will use to authenticate against the OAuth server.                                                                                                                     |         |
 | POLICY_STORE_AUTH_OAUTH_CLIENT_SECRET | The client secret OPAL will use to authenticate against the OAuth server.                                                                                                                 |         |
 | POLICY_STORE_CONN_RETRY               | Retry options when connecting to the policy store (i.e. the agent that handles the policy, e.g. OPA).                                                                                     |         |
+| DATA_STORE_CONN_RETRY                 | Retry options when connecting to the base data source (e.g. an external API server which returns data snapshot).                                                                          |         |
 | POLICY_STORE_POLICY_PATHS_TO_IGNORE   | Which policy paths pushed to the client should be ignored. List of glob style paths, or paths without wildcards but ending with "/\*\*" indicating a parent path (ignoring all under it). |         |
 | POLICY_UPDATER_CONN_RETRY             | Retry options when connecting to the policy source (e.g. the policy bundle server).                                                                                                       |         |
 | INLINE_OPA_ENABLED                    | Whether or not OPAL should run OPA by itself in the same container.                                                                                                                       |         |

packages/opal-client/opal_client/config.py

@@ -1,7 +1,7 @@
 from enum import Enum
 
 from opal_client.engine.options import CedarServerOptions, OpaServerOptions
-from opal_client.policy.options import PolicyConnRetryOptions
+from opal_client.policy.options import ConnRetryOptions
 from opal_client.policy_store.schemas import PolicyStoreAuth, PolicyStoreTypes
 from opal_common.confi import Confi, confi
 from opal_common.config import opal_common_config
@@ -49,16 +49,16 @@ class OpalClientConfig(Confi):
         description="the client secret OPAL will use to authenticate against the OAuth server.",
     )
 
-    POLICY_STORE_CONN_RETRY: PolicyConnRetryOptions = confi.model(
+    POLICY_STORE_CONN_RETRY: ConnRetryOptions = confi.model(
         "POLICY_STORE_CONN_RETRY",
-        PolicyConnRetryOptions,
+        ConnRetryOptions,
         # defaults are being set according to PolicyStoreConnRetryOptions pydantic definitions (see class)
         {},
         description="retry options when connecting to the policy store (i.e. the agent that handles the policy, e.g. OPA)",
     )
-    POLICY_UPDATER_CONN_RETRY: PolicyConnRetryOptions = confi.model(
+    POLICY_UPDATER_CONN_RETRY: ConnRetryOptions = confi.model(
         "POLICY_UPDATER_CONN_RETRY",
-        PolicyConnRetryOptions,
+        ConnRetryOptions,
         {
             "wait_strategy": "random_exponential",
             "max_wait": 10,
@@ -68,6 +68,18 @@ class OpalClientConfig(Confi):
         description="retry options when connecting to the policy source (e.g. the policy bundle server)",
     )
 
+    DATA_STORE_CONN_RETRY: ConnRetryOptions = confi.model(
+        "DATA_STORE_CONN_RETRY",
+        ConnRetryOptions,
+        {
+            "wait_strategy": "random_exponential",
+            "max_wait": 10,
+            "attempts": 5,
+            "wait_time": 1,
+        },
+        description="retry options when connecting to the base data source (e.g. an external API server which returns data snapshot)",
+    )
+
     POLICY_STORE_POLICY_PATHS_TO_IGNORE = confi.list(
         "POLICY_STORE_POLICY_PATHS_TO_IGNORE",
         [],

packages/opal-client/opal_client/data/fetcher.py

@@ -14,7 +14,14 @@
 class DataFetcher:
     """fetches policy data from backend."""
 
-    def __init__(self, default_data_url: str = None, token: str = None):
+    # Use as default config the configuration provider by opal_client_config.DATA_STORE_CONN_RETRY
+    # Add reraise as true (an option not available for control from the higher-level config)
+    DEFAULT_RETRY_CONFIG = opal_client_config.DATA_STORE_CONN_RETRY.toTenacityConfig()
+    DEFAULT_RETRY_CONFIG["reraise"] = True
+
+    def __init__(
+        self, default_data_url: str = None, token: str = None, retry_config=None
+    ):
         """
 
         Args:
@@ -24,11 +31,15 @@ def __init__(self, default_data_url: str = None, token: str = None):
         # defaults
         default_data_url: str = default_data_url or opal_client_config.DEFAULT_DATA_URL
         token: str = token or opal_client_config.CLIENT_TOKEN
+        self._retry_config = (
+            retry_config if retry_config is not None else self.DEFAULT_RETRY_CONFIG
+        )
         # The underlying fetching engine
         self._engine = FetchingEngine(
             worker_count=opal_common_config.FETCHING_WORKER_COUNT,
             callback_timeout=opal_common_config.FETCHING_CALLBACK_TIMEOUT,
             enqueue_timeout=opal_common_config.FETCHING_ENQUEUE_TIMEOUT,
+            retry_config=self._retry_config,
         )
         self._data_url = default_data_url
         self._token = token

packages/opal-client/opal_client/policy/options.py

@@ -20,7 +20,7 @@ class WaitStrategy(str, Enum):
     random_exponential = "random_exponential"
 
 
-class PolicyConnRetryOptions(BaseModel):
+class ConnRetryOptions(BaseModel):
     wait_strategy: WaitStrategy = Field(
         WaitStrategy.fixed,
         description="waiting strategy (e.g. fixed for fixed-time waiting, exponential for exponential back-off) (default fixed)",

packages/opal-common/opal_common/fetcher/engine/fetching_engine.py

@@ -36,6 +36,7 @@ def __init__(
         worker_count: int = DEFAULT_WORKER_COUNT,
         callback_timeout: int = DEFAULT_CALLBACK_TIMEOUT,
         enqueue_timeout: int = DEFAULT_ENQUEUE_TIMEOUT,
+        retry_config=None,
     ) -> None:
         # The internal task queue (created at start_workers)
         self._queue: asyncio.Queue = None
@@ -51,6 +52,7 @@ def __init__(
         self._callback_timeout = callback_timeout
         # time in seconds before time out on adding a task to queue (when full)
         self._enqueue_timeout = enqueue_timeout
+        self._retry_config = retry_config
 
     def start_workers(self):
         if self._queue is None:
@@ -145,7 +147,9 @@ async def queue_url(
             fetcher = config.fetcher
 
         # init a URL event
-        event = FetchEvent(url=url, fetcher=fetcher, config=config)
+        event = FetchEvent(
+            url=url, fetcher=fetcher, config=config, retry=self._retry_config
+        )
         return await self.queue_fetch_event(event, callback)
 
     async def queue_fetch_event(

packages/opal-common/opal_common/fetcher/providers/http_fetch_provider.py

@@ -71,7 +71,9 @@ async def __aenter__(self):
         headers = {}
         if self._event.config.headers is not None:
             headers = self._event.config.headers
-        self._session = await ClientSession(headers=headers).__aenter__()
+        self._session = await ClientSession(
+            headers=headers, raise_for_status=True
+        ).__aenter__()
         return self
 
     async def __aexit__(self, exc_type=None, exc_val=None, tb=None):