Add configurable httpx request timeout and increase default

boefjes/boefjes/clients/bytes_client.py

@@ -9,6 +9,7 @@
 import structlog
 from httpx import Client, HTTPStatusError, HTTPTransport, Response
 
+from boefjes.config import settings
 from boefjes.job_models import BoefjeMeta, NormalizerMeta, RawDataMeta
 
 BYTES_API_CLIENT_VERSION = "0.3"
@@ -38,6 +39,7 @@ def __init__(self, base_url: str, username: str, password: str):
             base_url=base_url,
             headers={"User-Agent": f"bytes-api-client/{BYTES_API_CLIENT_VERSION}"},
             transport=(HTTPTransport(retries=6)),
+            timeout=settings.outgoing_request_timeout,
         )
 
         self.credentials = {"username": username, "password": password}

boefjes/boefjes/clients/scheduler_client.py

@@ -5,6 +5,7 @@
 from httpx import Client, HTTPTransport, Response
 from pydantic import BaseModel, TypeAdapter
 
+from boefjes.config import settings
 from boefjes.job_models import BoefjeMeta, NormalizerMeta
 
 
@@ -57,7 +58,9 @@ def push_item(self, p_item: Task) -> None:
 
 class SchedulerAPIClient(SchedulerClientInterface):
     def __init__(self, base_url: str):
-        self._session = Client(base_url=base_url, transport=HTTPTransport(retries=6))
+        self._session = Client(
+            base_url=base_url, transport=HTTPTransport(retries=6), timeout=settings.outgoing_request_timeout
+        )
 
     @staticmethod
     def _verify_response(response: Response) -> None:

boefjes/boefjes/config.py

@@ -122,6 +122,8 @@ class Settings(BaseSettings):
 
     logging_format: Literal["text", "json"] = Field("text", description="Logging format")
 
+    outgoing_request_timeout: int = Field(30, description="Timeout for outgoing HTTP requests")
+
     model_config = SettingsConfigDict(env_prefix="BOEFJES_")
 
     @classmethod

boefjes/boefjes/job_handler.py

@@ -33,14 +33,15 @@
 
 
 def get_octopoes_api_connector(org_code: str) -> OctopoesAPIConnector:
-    return OctopoesAPIConnector(str(settings.octopoes_api), org_code)
+    return OctopoesAPIConnector(str(settings.octopoes_api), org_code, timeout=settings.outgoing_request_timeout)
 
 
 def get_environment_settings(boefje_meta: BoefjeMeta, schema: dict | None = None) -> dict[str, str]:
     try:
         katalogus_api = str(settings.katalogus_api).rstrip("/")
         response = httpx.get(
-            f"{katalogus_api}/v1/organisations/{boefje_meta.organization}/{boefje_meta.boefje.id}/settings", timeout=30
+            f"{katalogus_api}/v1/organisations/{boefje_meta.organization}/{boefje_meta.boefje.id}/settings",
+            timeout=settings.outgoing_request_timeout,
         )
         response.raise_for_status()
     except HTTPError:

octopoes/octopoes/config/settings.py

@@ -68,6 +68,8 @@ class Settings(BaseSettings):
 
     logging_format: Literal["text", "json"] = Field("text", description="Logging format")
 
+    outgoing_request_timeout: int = Field(30, description="Timeout for outgoing HTTP requests")
+
     model_config = SettingsConfigDict(env_prefix="OCTOPOES_")
 
     @classmethod

octopoes/octopoes/connector/octopoes.py

@@ -38,10 +38,12 @@ class OctopoesAPIConnector:
         - connector.RemoteException if an error occurs inside Octopoes API
     """
 
-    def __init__(self, base_uri: str, client: str):
+    def __init__(self, base_uri: str, client: str, timeout: int = 30):
         self.base_uri = base_uri
         self.client = client
-        self.session = httpx.Client(base_url=base_uri, timeout=30, event_hooks={"response": [self._verify_response]})
+        self.session = httpx.Client(
+            base_url=base_uri, timeout=timeout, event_hooks={"response": [self._verify_response]}
+        )
         self.logger = structlog.get_logger("octopoes-connector", organisation_code=client)
 
     @staticmethod

octopoes/octopoes/xtdb/client.py

@@ -10,11 +10,13 @@
 from httpx import HTTPError, HTTPStatusError, Response, codes
 from pydantic import BaseModel, ConfigDict, Field, TypeAdapter
 
+from octopoes.config.settings import Settings
 from octopoes.models.transaction import TransactionRecord
 from octopoes.xtdb.exceptions import NodeNotFound, XTDBException
 from octopoes.xtdb.query import Query
 
 logger = structlog.get_logger(__name__)
+settings = Settings()
 
 
 class OperationType(Enum):
@@ -46,7 +48,10 @@ class XTDBStatus(BaseModel):
 @functools.cache
 def _get_xtdb_http_session(base_url: str) -> httpx.Client:
     return httpx.Client(
-        base_url=base_url, headers={"Accept": "application/json"}, transport=(httpx.HTTPTransport(retries=3))
+        base_url=base_url,
+        headers={"Accept": "application/json"},
+        transport=(httpx.HTTPTransport(retries=3)),
+        timeout=settings.outgoing_request_timeout,
     )
 
 

rocky/account/mixins.py

@@ -109,7 +109,9 @@ def setup(self, request, *args, **kwargs):
         if self.organization_member.blocked:
             raise PermissionDenied()
 
-        self.octopoes_api_connector = OctopoesAPIConnector(settings.OCTOPOES_API, organization_code)
+        self.octopoes_api_connector = OctopoesAPIConnector(
+            settings.OCTOPOES_API, organization_code, timeout=settings.ROCKY_OUTGOING_REQUEST_TIMEOUT
+        )
         self.bytes_client = get_bytes_client(organization_code)
 
     def get_context_data(self, **kwargs):
@@ -250,7 +252,9 @@ def organization(self) -> Organization:
 
     @cached_property
     def octopoes_api_connector(self) -> OctopoesAPIConnector:
-        return OctopoesAPIConnector(settings.OCTOPOES_API, self.organization.code)
+        return OctopoesAPIConnector(
+            settings.OCTOPOES_API, self.organization.code, timeout=settings.ROCKY_OUTGOING_REQUEST_TIMEOUT
+        )
 
     @cached_property
     def bytes_client(self) -> BytesClient:

rocky/crisis_room/views.py

@@ -58,7 +58,9 @@ def sort_by_severity(
 
     def get_finding_type_severity_count(self, organization: Organization) -> dict[str, int]:
         try:
-            api_connector = OctopoesAPIConnector(settings.OCTOPOES_API, organization.code)
+            api_connector = OctopoesAPIConnector(
+                settings.OCTOPOES_API, organization.code, timeout=settings.ROCKY_OUTGOING_REQUEST_TIMEOUT
+            )
             return api_connector.count_findings_by_severity(valid_time=self.observed_at)
         except ConnectorException:
             messages.add_message(

rocky/katalogus/client.py

@@ -130,7 +130,7 @@ def __init__(self, error: httpx.HTTPStatusError):
 
 class KATalogusClientV1:
     def __init__(self, base_uri: str, organization: str | None):
-        self.session = httpx.Client(base_url=base_uri)
+        self.session = httpx.Client(base_url=base_uri, timeout=settings.ROCKY_OUTGOING_REQUEST_TIMEOUT)
         self.organization = valid_organization_code(organization) if organization else organization
         self.organization_uri = f"/v1/organisations/{organization}"
 

rocky/reports/runner/report_runner.py

@@ -22,7 +22,9 @@ def __init__(self, bytes_client: BytesClient, valid_time: datetime | None = None
     def run(self, report_task: ReportTask) -> None:
         valid_time = self.valid_time or datetime.now(timezone.utc)
 
-        connector = OctopoesAPIConnector(settings.OCTOPOES_API, report_task.organisation_id)
+        connector = OctopoesAPIConnector(
+            settings.OCTOPOES_API, report_task.organisation_id, timeout=settings.ROCKY_OUTGOING_REQUEST_TIMEOUT
+        )
         recipe = connector.get(Reference.from_str(f"ReportRecipe|{report_task.report_recipe_id}"), valid_time)
         report_types = [get_report_by_id(report_type_id) for report_type_id in recipe.report_types]
 

rocky/rocky/bytes_client.py

@@ -18,7 +18,7 @@
 class BytesClient:
     def __init__(self, base_url: str, username: str, password: str, organization: str | None):
         self.credentials = {"username": username, "password": password}
-        self.session = httpx.Client(base_url=base_url)
+        self.session = httpx.Client(base_url=base_url, timeout=settings.ROCKY_OUTGOING_REQUEST_TIMEOUT)
         self.organization = organization
 
     def health(self) -> ServiceHealth:

rocky/rocky/keiko.py

@@ -34,7 +34,7 @@ class GeneratingReportFailed(ReportException):
 
 class KeikoClient:
     def __init__(self, base_uri: str, timeout: int = 60):
-        self.session = httpx.Client(base_url=base_uri)
+        self.session = httpx.Client(base_url=base_uri, timeout=settings.ROCKY_OUTGOING_REQUEST_TIMEOUT)
         self._timeout = timeout
 
     def generate_report(self, template: str, data: dict, glossary: str) -> str:

rocky/rocky/scheduler.py

@@ -265,7 +265,7 @@ class SchedulerHTTPError(SchedulerError):
 
 class SchedulerClient:
     def __init__(self, base_uri: str, organization_code: str | None):
-        self._client = httpx.Client(base_url=base_uri)
+        self._client = httpx.Client(base_url=base_uri, timeout=settings.ROCKY_OUTGOING_REQUEST_TIMEOUT)
         self.organization_code = organization_code
 
     def list_schedules(self, **kwargs) -> PaginatedSchedulesResponse:

rocky/rocky/settings.py

@@ -495,3 +495,5 @@ def immutable_file_test(path, url):
 POLL_INTERVAL = env.int("POLL_INTERVAL", default=10)
 # Seconds to wait before checking the workers when queues are full
 WORKER_HEARTBEAT = env.int("WORKER_HEARTBEAT", default=5)
+
+ROCKY_OUTGOING_REQUEST_TIMEOUT = env.int("ROCKY_OUTGOING_REQUEST_TIMEOUT", default=30)

rocky/rocky/views/organization_list.py

@@ -41,7 +41,9 @@ def post(self, request: HttpRequest, *args, **kwargs) -> HttpResponse:
             for organization in organizations:
                 try:
                     logger.info("Recalculating bits", event_code=920000, organization_code=organization.code)
-                    number_of_bits += OctopoesAPIConnector(settings.OCTOPOES_API, organization.code).recalculate_bits()
+                    number_of_bits += OctopoesAPIConnector(
+                        settings.OCTOPOES_API, organization.code, timeout=settings.ROCKY_OUTGOING_REQUEST_TIMEOUT
+                    ).recalculate_bits()
                 except Exception as exc:
                     failed.append(f"{organization}, ({str(exc)})")
                     logging.warning("Failed recalculating bits for %s, %s", organization, exc)

rocky/tools/management/commands/generate_report.py

@@ -74,7 +74,13 @@ def handle(self, *args, **options):
 
     @staticmethod
     def get_findings_metadata(organization, valid_time, severities) -> list[dict[str, Any]]:
-        findings = FindingList(OctopoesAPIConnector(settings.OCTOPOES_API, organization.code), valid_time, severities)
+        findings = FindingList(
+            OctopoesAPIConnector(
+                settings.OCTOPOES_API, organization.code, timeout=settings.ROCKY_OUTGOING_REQUEST_TIMEOUT
+            ),
+            valid_time,
+            severities,
+        )
 
         return generate_findings_metadata(findings, severities)
 

rocky/tools/models.py

@@ -192,7 +192,9 @@ def _get_healthy_katalogus(organization_code: str) -> KATalogusClientV1:
 
     @staticmethod
     def _get_healthy_octopoes(organization_code: str) -> OctopoesAPIConnector:
-        octopoes_client = OctopoesAPIConnector(settings.OCTOPOES_API, client=organization_code)
+        octopoes_client = OctopoesAPIConnector(
+            settings.OCTOPOES_API, client=organization_code, timeout=settings.ROCKY_OUTGOING_REQUEST_TIMEOUT
+        )
         try:
             health = octopoes_client.root_health()
         except HTTPError as e:

rocky/tools/viewsets.py

@@ -67,7 +67,9 @@ def set_indemnification(self, request, pk=None):
     def recalculate_bits(self, request, pk=None):
         organization = self.get_object()
         logger.info("Recalculating bits", event_code=920000, organization_code=organization.code)
-        connector = OctopoesAPIConnector(settings.OCTOPOES_API, organization.code)
+        connector = OctopoesAPIConnector(
+            settings.OCTOPOES_API, organization.code, timeout=settings.ROCKY_OUTGOING_REQUEST_TIMEOUT
+        )
         number_of_bits = connector.recalculate_bits()
 
         return Response({"number_of_bits": number_of_bits})