feat(backend): add task manager retention settings [INFP-507]

backend/infrahub/config.py

@@ -28,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:
@@ -2100,6 +2105,10 @@ def trace(self) -> TraceSettings:
     def experimental_features(self) -> ExperimentalFeaturesSettings:
         return self.active_settings.experimental_features
 
+    @property
+    def task_manager(self) -> TaskManagerSettings:
+        return self.active_settings.task_manager
+
     @property
     def enterprise_features(self) -> list[EnterpriseFeatures]:
         """Returns a list of enterprise features that are enabled based on the settings."""
@@ -2130,6 +2139,8 @@ class Settings(BaseSettings):
     trace: TraceSettings = TraceSettings()
     experimental_features: ExperimentalFeaturesSettings = ExperimentalFeaturesSettings()
     log_forwarding: LogForwardingSettings = LogForwardingSettings()
+    # Built when the configuration loads, so an invalid retention fails the load instead of the module import.
+    task_manager: TaskManagerSettings = Field(default_factory=TaskManagerSettings)
 
     @model_validator(mode="after")
     def validate_git_branch_deletion_requires_branch_deletion(self) -> Self:
@@ -2210,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

@@ -1,13 +1,30 @@
 from __future__ import annotations
 
 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,
+    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__)
 
 router = APIRouter(prefix="/infrahub")
 
@@ -29,7 +46,47 @@ async def _init_prefect() -> None:
         await init_prefect()
 
 
+def apply_infrahub_settings_to_prefect() -> None:
+    """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()
+    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}")
+
+
+def create_prefect_app_with_infrahub_routes() -> FastAPI:
+    """Build Prefect's server app with Infrahub's routes, leaving Prefect's settings as the environment sets them."""
+    app = create_app()
+    api_app: FastAPI = app.__dict__["api_app"]
+    api_app.include_router(router=router)
+
+    return app
+
+
 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"
@@ -40,8 +97,4 @@ def create_infrahub_prefect() -> FastAPI:
         config.SETTINGS.initialize_and_exit()
         asyncio.run(_init_prefect())
 
-    app = create_app()
-    api_app: FastAPI = app.__dict__["api_app"]
-    api_app.include_router(router=router)
-
-    return app
+    return create_prefect_app_with_infrahub_routes()