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

@@ -1,6 +1,8 @@
 import asyncio
 from typing import Any, Dict, List, Optional, Tuple
 
+import aiohttp
+from fastapi import HTTPException
 from opal_client.config import opal_client_config
 from opal_client.policy_store.base_policy_store_client import JsonableValue
 from opal_common.config import opal_common_config
@@ -9,12 +11,20 @@
 from opal_common.fetcher.providers.http_fetch_provider import HttpFetcherConfig
 from opal_common.logger import logger
 from opal_common.utils import get_authorization_header, tuple_to_dict
+from tenacity import retry
 
 
 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:
@@ -36,6 +46,9 @@ def __init__(self, default_data_url: str = None, token: str = None):
         self._default_fetcher_config = HttpFetcherConfig(
             headers=self._auth_headers, is_json=True
         )
+        self._retry_config = (
+            retry_config if retry_config is not None else self.DEFAULT_RETRY_CONFIG
+        )
 
     async def __aenter__(self):
         await self.start()
@@ -65,13 +78,39 @@ async def handle_url(
             return None
 
         logger.info("Fetching data from url: {url}", url=url)
+        response = None
+
+        @retry(**self._retry_config)
+        async def handle_url_with_retry():
+            nonlocal response
+            try:
+                # ask the engine to get our data
+                response = await self._engine.handle_url(url, config=config)
+                if (
+                    isinstance(response, aiohttp.ClientResponse)
+                    and response.status >= 400
+                ):
+                    logger.error(
+                        "error while fetching url: {url}, got response code: {code}",
+                        url=url,
+                        code=response.status,
+                    )
+                    raise HTTPException(
+                        status_code=response.status,
+                        detail=f"bad status code(>=400) returned accessing url: {url}",
+                    )
+                return response
+            except asyncio.TimeoutError as e:
+                logger.exception("Timeout while fetching url: {url}", url=url)
+                raise
+
         try:
-            # ask the engine to get our data
-            response = await self._engine.handle_url(url, config=config)
+            await handle_url_with_retry()
+        except HTTPException:
             return response
         except asyncio.TimeoutError as e:
-            logger.exception("Timeout while fetching url: {url}", url=url)
             raise
+        return response
 
     async def handle_urls(
         self, urls: List[Tuple[str, FetcherConfig, Optional[JsonableValue]]] = None

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)",