fix(backend): fix review findings on the task manager retention settings [INFP-507]

backend/infrahub/config.py

@@ -5,7 +5,6 @@
 import sys
 import tomllib
 from dataclasses import dataclass
-from datetime import timedelta
 from enum import StrEnum
 from pathlib import Path
 from typing import TYPE_CHECKING, Annotated, Any
@@ -18,9 +17,7 @@
     EmailStr,
     Field,
     PrivateAttr,
-    TypeAdapter,
     ValidationError,
-    ValidationInfo,
     computed_field,
     field_validator,
     model_validator,
@@ -31,6 +28,11 @@
 from infrahub.constants.database import DatabaseType
 from infrahub.exceptions import InitializationError, ProcessingError
 from infrahub.log import get_logger
+from infrahub.prefect_server.retention_settings import (
+    TaskManagerRetentionSettings,  # noqa: F401 - re-exported with the other settings models
+    TaskManagerSettings,
+    print_configuration_errors,
+)
 from infrahub.tls.bundle import resolve_ca_bundle
 
 if TYPE_CHECKING:
@@ -685,104 +687,6 @@ def api_endpoint(self) -> str:
         return url
 
 
-_RETENTION_IN_DAYS = re.compile(r"(\d+)d")
-_ISO_8601_DURATION = re.compile(
-    r"P(?=\d|T\d)(?:\d+Y)?(?:\d+M)?(?:\d+W)?(?:\d+D)?(?:T(?=\d)(?:\d+H)?(?:\d+M)?(?:\d+(?:\.\d+)?S)?)?"
-)
-_MINIMUM_RETENTION = timedelta(days=1)
-# Prefect's cutoff of now minus the retention must stay a valid date, which a century keeps far from overflowing.
-_MAXIMUM_RETENTION = timedelta(days=36500)
-_TIMEDELTA_ADAPTER: TypeAdapter[timedelta] = TypeAdapter(timedelta)
-
-
-def _days(count: str) -> timedelta:
-    try:
-        return timedelta(days=int(count))
-    except (OverflowError, ValueError):
-        # Still a number of days, so it is refused by the upper bound rather than as an unreadable value.
-        return timedelta.max
-
-
-def _parse_retention(value: Any) -> timedelta | None:
-    if isinstance(value, timedelta):
-        return value
-    if not isinstance(value, str):
-        return None
-    text = value.strip()
-    if match := _RETENTION_IN_DAYS.fullmatch(text):
-        return _days(count=match.group(1))
-    # Pydantic also reads forms such as "1 day, 00:00:00", which are not part of the documented contract.
-    if not text.startswith("P"):
-        return None
-    try:
-        return _TIMEDELTA_ADAPTER.validate_python(text)
-    except ValidationError:
-        # A well-formed duration too large to read is refused by the upper bound rather than as an unreadable value.
-        return timedelta.max if _ISO_8601_DURATION.fullmatch(text) else None
-
-
-class TaskManagerRetentionSettings(BaseSettings):
-    model_config = SettingsConfigDict(env_prefix="INFRAHUB_TASK_MANAGER_RETENTION_")
-
-    task_history: timedelta = Field(
-        default="30d",
-        validate_default=True,
-        description=(
-            "How long finished task runs are kept, with their logs and artifacts, as a number of days (`30d`) "
-            "or an ISO 8601 duration (`P30D`), from 1 to 36500 days. `PREFECT_SERVER_SERVICES_DB_VACUUM_ENABLED` and "
-            "`PREFECT_SERVER_SERVICES_DB_VACUUM_RETENTION_PERIOD` take precedence when set."
-        ),
-    )
-    activity_log: timedelta = Field(
-        default="7d",
-        validate_default=True,
-        description=(
-            "How long the events of the activity log are kept, as a number of days (`7d`) or an ISO 8601 "
-            "duration (`P7D`), from 1 to 36500 days. `PREFECT_SERVER_EVENTS_RETENTION_PERIOD`, or its legacy name "
-            "`PREFECT_EVENTS_RETENTION_PERIOD`, takes precedence when set."
-        ),
-    )
-    prefect_own_events: timedelta = Field(
-        default="7d",
-        validate_default=True,
-        description=(
-            "How long the task manager's own Prefect events are kept, as a number of days (`7d`) or an ISO 8601 "
-            "duration (`P7D`), from 1 to 36500 days; a value longer than `activity_log` is capped to it. "
-            "`PREFECT_SERVER_SERVICES_DB_VACUUM_EVENT_RETENTION_OVERRIDES` takes precedence when set."
-        ),
-    )
-
-    @field_validator("task_history", "activity_log", "prefect_own_events", mode="before")
-    @classmethod
-    def validate_retention_from_one_to_36500_days(cls, value: Any, info: ValidationInfo) -> timedelta:
-        """Read a number of days such as `30d` or an ISO 8601 duration such as `P30D`.
-
-        Raises:
-            ValueError: When the value is in neither form, is shorter than 1 day or is longer than 36500 days.
-
-        """
-        retention = _parse_retention(value)
-        if retention is None:
-            raise ValueError(
-                f"Invalid task manager retention: {info.field_name} must be a number of days such as 30d "
-                "or an ISO 8601 duration such as P30D"
-            )
-        if retention < _MINIMUM_RETENTION:
-            raise ValueError(f"Invalid task manager retention: {info.field_name} must be at least 1 day")
-        if retention > _MAXIMUM_RETENTION:
-            raise ValueError(f"Invalid task manager retention: {info.field_name} must be at most 36500 days")
-        return retention
-
-
-class TaskManagerSettings(BaseSettings):
-    """How long the task manager keeps its task history and activity log."""
-
-    model_config = SettingsConfigDict(env_prefix="INFRAHUB_TASK_MANAGER_")
-    retention: TaskManagerRetentionSettings = Field(
-        default_factory=TaskManagerRetentionSettings, description="How long each kind of record is kept."
-    )
-
-
 class ApiSettings(BaseSettings):
     model_config = SettingsConfigDict(env_prefix="INFRAHUB_API_")
     cors_allow_origins: list[str] = Field(
@@ -2317,10 +2221,7 @@ def load_and_exit(config_file_name: Path | str = "infrahub.toml", config_data: d
     try:
         SETTINGS.settings = load(config_file_name=config_file_name, config_data=config_data)
     except ValidationError as err:
-        print(f"Configuration not valid, found {len(err.errors())} error(s)")
-        for error in err.errors():
-            error_locations = [str(location) for location in error["loc"]]
-            print(f"  {'/'.join(error_locations)} | {error['msg']} ({error['type']})")
+        print_configuration_errors(error=err)
         sys.exit(1)
 
 

backend/infrahub/prefect_server/app.py

@@ -3,14 +3,26 @@
 import asyncio
 import logging
 import os
+import sys
+from pathlib import Path
 
 from fastapi import APIRouter, FastAPI
 from prefect.context import refresh_global_settings_context
 from prefect.server.api.server import create_app
+from prefect.settings import get_current_settings
+from pydantic import ValidationError
 
 from . import events
 from .bootstrap import init_prefect
-from .retention import apply_prefect_retention_env, prefect_retention_env_in_effect
+from .retention import (
+    apply_prefect_retention_env,
+    background_services_warning,
+    prefect_retention_already_set,
+    prefect_retention_in_effect,
+    remove_blank_vacuum_switch,
+    turned_off_cleanup_warnings,
+)
+from .retention_settings import load_task_manager_settings, print_configuration_errors
 
 log = logging.getLogger(__name__)
 
@@ -35,16 +47,28 @@ async def _init_prefect() -> None:
 
 
 def apply_infrahub_settings_to_prefect() -> None:
-    """Load Infrahub's configuration into Prefect's retention settings and log them, exiting when it is invalid."""
-    # The original Prefect container used by some test suites has no infrahub package.
-    from infrahub import config  # noqa: PLC0415
-
-    config.SETTINGS.initialize_and_exit()
-    for warning in apply_prefect_retention_env(environ=os.environ, settings=config.SETTINGS.task_manager.retention):
+    """Read the task manager section of Infrahub's configuration into Prefect's retention settings and log them, exiting when it is invalid."""
+    # Sets up the log handler and level that the retention lines need, and is imported here because the stock
+    # Prefect container of some test suites has no infrahub package.
+    import infrahub.log  # noqa: F401, PLC0415
+
+    config_file = Path(os.environ.get("INFRAHUB_CONFIG", "infrahub.toml")).resolve()
+    try:
+        retention = load_task_manager_settings(config_file=config_file).retention
+    except ValidationError as error:
+        print_configuration_errors(error=error)
+        sys.exit(1)
+    remove_blank_vacuum_switch(environ=os.environ)
+    # Read before any write, since a variable written here would count as one the operator set.
+    already_set = prefect_retention_already_set()
+    for warning in apply_prefect_retention_env(environ=os.environ, settings=retention, already_set=already_set):
         log.warning(warning)
     # Prefect reads its settings from the environment once at import, so new values apply only after a refresh.
     refresh_global_settings_context()
-    for name, value in prefect_retention_env_in_effect(environ=os.environ).items():
+    prefect_settings = get_current_settings()
+    for warning in turned_off_cleanup_warnings(vacuum=prefect_settings.server.services.db_vacuum):
+        log.warning(warning)
+    for name, value in prefect_retention_in_effect(settings=prefect_settings).items():
         log.info(f"Task manager retention: {name}={value}")
 
 
@@ -60,12 +84,17 @@ def create_prefect_app_with_infrahub_routes() -> FastAPI:
 def create_infrahub_prefect() -> FastAPI:
     """Build the task manager app with Prefect's settings derived from Infrahub's configuration."""
     apply_infrahub_settings_to_prefect()
+    if warning := background_services_warning(environ=os.environ):
+        log.warning(warning)
 
     if (
         os.getenv("PREFECT_API_BLOCKS_REGISTER_ON_START") == "false"
         and os.getenv("PREFECT_API_DATABASE_MIGRATE_ON_START") == "false"
     ):
         # We are probably running distributed mode
+        from infrahub import config  # noqa: PLC0415
+
+        config.SETTINGS.initialize_and_exit()
         asyncio.run(_init_prefect())
 
     return create_prefect_app_with_infrahub_routes()