Add CONTROL_PLANE_TIMEOUT configuration and apply it across various components (#285)

- Introduced CONTROL_PLANE_TIMEOUT setting in SidecarConfig for controlling request timeouts.
- Updated aiohttp ClientSession and AsyncClient to utilize the new timeout configuration.
- Modified BlockingRequest to accept a timeout parameter, ensuring consistent timeout handling in blocking requests.
- Adjusted RemoteConfigFetcher and EnvApiKeyFetcher to incorporate the timeout setting for improved request management.

Author: danyi1212

horizon/config.py

@@ -34,6 +34,12 @@ def __new__(cls, *, prefix=None, is_model=True):  # noqa: ARG004
         description="URL to the control plane that manages this PDP, typically Permit.io cloud (api.permit.io)",
     )
 
+    CONTROL_PLANE_TIMEOUT = confi.float(
+        "CONTROL_PLANE_TIMEOUT",
+        75,
+        description="Timeout in seconds for control plane requests",
+    )
+
     CONTROL_PLANE_PDP_DELTAS_API = confi.str(
         "CONTROL_PLANE_PDP_DELTAS_API",
         "http://localhost:8000",

horizon/facts/client.py

@@ -26,6 +26,7 @@ def client(self) -> AsyncClient:
             env_api_key = get_env_api_key()
             self._client = AsyncClient(
                 base_url=sidecar_config.CONTROL_PLANE,
+                timeout=sidecar_config.CONTROL_PLANE_TIMEOUT,
                 headers={"Authorization": f"Bearer {env_api_key}"},
                 trust_env=True,
             )

horizon/startup/api_keys.py

@@ -19,9 +19,11 @@ class EnvApiKeyFetcher:
     def __init__(
         self,
         backend_url: str = sidecar_config.CONTROL_PLANE,
+        timeout: float = sidecar_config.CONTROL_PLANE_TIMEOUT,
         retry_config=None,
     ):
         self._backend_url = backend_url
+        self._timeout = timeout
         self._retry_config = retry_config or DEFAULT_RETRY_CONFIG
         self.api_key_level = self._get_api_key_level()
 
@@ -83,6 +85,7 @@ def _fetch_env_key(self, api_key: str, active_project_key: str, active_env_key:
         fetch_with_retry = retry(**self._retry_config)(
             lambda: BlockingRequest(
                 token=api_key,
+                timeout=self._timeout,
             ).get(url=api_key_url)
         )
         try:
@@ -105,6 +108,7 @@ def fetch_scope(self, api_key: str) -> dict | None:
         fetch_with_retry = retry(**self._retry_config)(
             lambda: BlockingRequest(
                 token=api_key,
+                timeout=self._timeout,
             ).get(url=api_key_url)
         )
         try:

horizon/startup/blocking_request.py

@@ -6,9 +6,10 @@
 
 
 class BlockingRequest:
-    def __init__(self, token: str | None, extra_headers: dict[str, Any] | None = None):
+    def __init__(self, token: str | None, extra_headers: dict[str, Any] | None = None, timeout: float = 60):
         self._token = token
         self._extra_headers = {k: v for k, v in (extra_headers or {}).items() if v is not None}
+        self._timeout = timeout
 
     def _headers(self) -> dict[str, str]:
         headers = {}
@@ -22,7 +23,7 @@ def get(self, url: str, params=None) -> dict:
         """
         utility method to send a *blocking* HTTP GET request and get the response back.
         """
-        response = requests.get(url, headers=self._headers(), params=params)
+        response = requests.get(url, headers=self._headers(), params=params, timeout=self._timeout)
 
         if response.status_code == 401:
             raise InvalidPDPTokenError()
@@ -33,7 +34,7 @@ def post(self, url: str, payload: dict | None = None, params=None) -> dict:
         """
         utility method to send a *blocking* HTTP POST request with a JSON payload and get the response back.
         """
-        response = requests.post(url, json=payload, headers=self._headers(), params=params)
+        response = requests.post(url, json=payload, headers=self._headers(), params=params, timeout=self._timeout)
 
         if response.status_code == 401:
             raise InvalidPDPTokenError()

horizon/startup/remote_config.py

@@ -50,6 +50,7 @@ def __init__(
         backend_url: str = sidecar_config.CONTROL_PLANE,
         remote_config_route: str = sidecar_config.REMOTE_CONFIG_ENDPOINT,
         shard_id: str | None = sidecar_config.SHARD_ID,
+        timeout: float = sidecar_config.CONTROL_PLANE_TIMEOUT,
         retry_config=None,
     ):
         """
@@ -63,6 +64,7 @@ def __init__(
         self._url = f"{backend_url}{remote_config_route}"
         self._backend_url = backend_url
         self._token = get_env_api_key()
+        self._timeout = timeout
         self._retry_config = retry_config if retry_config is not None else DEFAULT_RETRY_CONFIG
         self._shard_id = shard_id
 
@@ -91,8 +93,13 @@ def _fetch_config(self) -> RemoteConfig:
         However, this is ok because the RemoteConfigFetcher runs *once* when the sidecar starts.
         """
         try:
-            response = BlockingRequest(token=self._token, extra_headers={"X-Shard-ID": self._shard_id}).post(
-                url=self._url, payload=PersistentStateHandler.build_state_payload_sync()
+            response = BlockingRequest(
+                token=self._token,
+                extra_headers={"X-Shard-ID": self._shard_id},
+                timeout=self._timeout,
+            ).post(
+                url=self._url,
+                payload=PersistentStateHandler.build_state_payload_sync(),
             )
 
             try:

horizon/state.py

@@ -200,6 +200,7 @@ async def _report(self, state: PersistentState | None = None):
         config_url = f"{sidecar_config.CONTROL_PLANE}{sidecar_config.REMOTE_STATE_ENDPOINT}"
         async with aiohttp.ClientSession(
             trust_env=True,
+            timeout=aiohttp.ClientTimeout(total=sidecar_config.CONTROL_PLANE_TIMEOUT),
         ) as session:
             logger.info("Reporting status update to server...")
             response = await session.post(