diff --git a/backend/infrahub/config.py b/backend/infrahub/config.py index 1e0cd2bc86d..b5eaaf84107 100644 --- a/backend/infrahub/config.py +++ b/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) diff --git a/backend/infrahub/prefect_server/app.py b/backend/infrahub/prefect_server/app.py index 602e937ea97..7ca4ea3b203 100644 --- a/backend/infrahub/prefect_server/app.py +++ b/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() diff --git a/backend/infrahub/prefect_server/retention.py b/backend/infrahub/prefect_server/retention.py index ee68ac742fa..277068e3f1a 100644 --- a/backend/infrahub/prefect_server/retention.py +++ b/backend/infrahub/prefect_server/retention.py @@ -12,9 +12,10 @@ if TYPE_CHECKING: from collections.abc import Mapping, MutableMapping + from prefect.settings import Settings from pydantic_settings import BaseSettings - from infrahub.config import TaskManagerRetentionSettings + from .retention_settings import TaskManagerRetentionSettings _STATE_NAMES = ( "AwaitingConcurrencySlot", @@ -30,6 +31,7 @@ "NotReady", "Paused", "Pending", + "Rerunning", "Resuming", "Retrying", "RolledBack", @@ -47,6 +49,8 @@ "prefect.flow-run.pull-step.executed", "prefect.flow-run.pull-step.failed", "prefect.runner.cancelled-flow-run", + "prefect.workspace.transfer.started", + "prefect.workspace.transfer.completed", "prefect.worker.started", "prefect.worker.stopped", "prefect.worker.submitted-flow-run", @@ -110,13 +114,14 @@ VACUUM_RETENTION_PERIOD = "PREFECT_SERVER_SERVICES_DB_VACUUM_RETENTION_PERIOD" EVENTS_RETENTION_PERIOD = "PREFECT_SERVER_EVENTS_RETENTION_PERIOD" EVENT_RETENTION_OVERRIDES = "PREFECT_SERVER_SERVICES_DB_VACUUM_EVENT_RETENTION_OVERRIDES" +WEBSERVER_ONLY = "PREFECT__SERVER_WEBSERVER_ONLY" @dataclass(frozen=True) class _PrefectSetting: model: type[BaseSettings] field: str - retention_key: str + overridden_retention: str | None def environment_names(self) -> list[str]: """Every environment variable Prefect reads the setting from, in the order Prefect tries them.""" @@ -125,17 +130,18 @@ def environment_names(self) -> list[str]: _PREFECT_SETTINGS: dict[str, _PrefectSetting] = { - VACUUM_ENABLED: _PrefectSetting( - model=ServerServicesDBVacuumSettings, field="enabled", retention_key="task_history" - ), + # The switch turns cleanups on or off and replaces no retention. + VACUUM_ENABLED: _PrefectSetting(model=ServerServicesDBVacuumSettings, field="enabled", overridden_retention=None), VACUUM_RETENTION_PERIOD: _PrefectSetting( - model=ServerServicesDBVacuumSettings, field="retention_period", retention_key="task_history" + model=ServerServicesDBVacuumSettings, field="retention_period", overridden_retention="task_history" ), EVENTS_RETENTION_PERIOD: _PrefectSetting( - model=ServerEventsSettings, field="retention_period", retention_key="activity_log" + model=ServerEventsSettings, field="retention_period", overridden_retention="activity_log" ), EVENT_RETENTION_OVERRIDES: _PrefectSetting( - model=ServerServicesDBVacuumSettings, field="event_retention_overrides", retention_key="prefect_own_events" + model=ServerServicesDBVacuumSettings, + field="event_retention_overrides", + overridden_retention="prefect_own_events", ), } @@ -150,11 +156,6 @@ def _iso_8601(duration: timedelta) -> str: return f"{days}T{seconds}S" -def _own_events_retention(settings: TaskManagerRetentionSettings) -> timedelta: - # Prefect deletes every event older than the global event retention whatever its per-type override says. - return min(settings.prefect_own_events, settings.activity_log) - - def _name_set_in(environ: Mapping[str, str], setting: _PrefectSetting) -> str | None: # Prefect matches environment names case-insensitively and reads the first accepted name present. names_in_environ = {name.upper(): name for name in environ} @@ -164,14 +165,24 @@ def _name_set_in(environ: Mapping[str, str], setting: _PrefectSetting) -> str | return None +def remove_blank_vacuum_switch(environ: MutableMapping[str, str]) -> None: + """Remove the vacuum switch when it is present but empty or blank, under any name Prefect accepts for it.""" + # A compose file passes an unset host variable as an empty value, which Prefect reads as false for this switch + # and so deletes no task runs, while an empty retention variable already made the import of Prefect fail. + accepted = {name.upper() for name in _PREFECT_SETTINGS[VACUUM_ENABLED].environment_names()} + for name in [name for name in environ if name.upper() in accepted and not environ[name].strip()]: + del environ[name] + + def build_prefect_retention_env(settings: TaskManagerRetentionSettings, event_types: tuple[str, ...]) -> dict[str, str]: """Return the Prefect variables that enforce the retention. Args: event_types: Prefect event types kept for the own-event retention instead of the activity log retention. + No type gives an empty mapping, which also replaces Prefect's default override for heartbeats. """ - own_events = _iso_8601(_own_events_retention(settings)) + own_events = _iso_8601(settings.prefect_own_events) return { VACUUM_ENABLED: "events,flow_runs", VACUUM_RETENTION_PERIOD: _iso_8601(settings.task_history), @@ -180,52 +191,114 @@ def build_prefect_retention_env(settings: TaskManagerRetentionSettings, event_ty } -def apply_prefect_retention_env(environ: MutableMapping[str, str], settings: TaskManagerRetentionSettings) -> list[str]: - """Set the retention variables that the environment does not already set, and return the warnings to log. +def prefect_retention_already_set() -> frozenset[str]: + """Return the retention variables whose Prefect setting a Prefect source already sets. - A Prefect setting already present under any name Prefect accepts for it is left as it is, so an operator's - explicit Prefect setting keeps precedence. + Prefect reads a setting from the environment, a `.env` file, `prefect.toml`, the `[tool.prefect]` table of + `pyproject.toml` or the active profile, and marks a field as set only when one of them supplies it. An empty + vacuum switch counts as set, so the caller removes a blank one first. """ - warnings: list[str] = [] - own_events = _own_events_retention(settings) - if own_events < settings.prefect_own_events: - warnings.append( - f"task_manager.retention.prefect_own_events {_iso_8601(settings.prefect_own_events)} is longer than " - f"task_manager.retention.activity_log {_iso_8601(settings.activity_log)}, " - f"Prefect's own events are kept for {_iso_8601(own_events)}" - ) + return frozenset( + name for name, setting in _PREFECT_SETTINGS.items() if setting.field in setting.model().model_fields_set + ) + - for name, value in build_prefect_retention_env(settings=settings, event_types=PREFECT_EVENT_TYPES).items(): +def apply_prefect_retention_env( + environ: MutableMapping[str, str], settings: TaskManagerRetentionSettings, already_set: frozenset[str] +) -> list[str]: + """Set the retention variables whose Prefect setting no Prefect source sets, and return the warnings to log. + + A Prefect setting that is already set is left as it is, so an operator's explicit Prefect setting keeps + precedence, with a warning when it replaces a retention. The per-type event overrides are written as an empty + mapping when they cannot shorten the event retention that Infrahub writes. + + Args: + already_set: Retention variables whose Prefect setting the environment, a Prefect settings file or the + active profile already sets, read before any variable is written. + + """ + # Prefect deletes every event older than the general retention anyway, so overrides that are not shorter only + # add delete queries. + write_overrides = EVENTS_RETENTION_PERIOD in already_set or settings.prefect_own_events < settings.activity_log + event_types = PREFECT_EVENT_TYPES if write_overrides else () + warnings: list[str] = [] + for name, value in build_prefect_retention_env(settings=settings, event_types=event_types).items(): + if name not in already_set: + environ[name] = value + continue setting = _PREFECT_SETTINGS[name] - name_set = _name_set_in(environ=environ, setting=setting) - if name_set is not None: - warnings.append(f"{name_set} is set and overrides task_manager.retention.{setting.retention_key}") + if setting.overridden_retention is None: continue - environ[name] = value + overrides = f"overrides task_manager.retention.{setting.overridden_retention}" + name_set = _name_set_in(environ=environ, setting=setting) + if name_set is None: + warnings.append(f"{name} is set in a Prefect settings file or profile and {overrides}") + else: + warnings.append(f"{name_set} is set and {overrides}") return warnings -def _count_overrides_per_retention(value: str) -> str: +def background_services_warning(environ: Mapping[str, str]) -> str | None: + """Return the warning to log when Prefect's background services, which delete old records, run in another process.""" + # Prefect's server reads any non-empty value of this variable as true, "false" included. + if not environ.get(WEBSERVER_ONLY): + return None + return ( + f"Task manager retention: {WEBSERVER_ONLY} is set, so this process does not run Prefect's background " + "services, which delete old task runs and events. The process that runs them, such as " + "`prefect server services start`, does not read task_manager.retention: unless it sets the same PREFECT_* " + "variables, it deletes no task runs and keeps events for Prefect's default of 7 days." + ) + + +def turned_off_cleanup_warnings(vacuum: ServerServicesDBVacuumSettings) -> list[str]: + """Return a warning for each of Prefect's cleanups that the vacuum switch turns off, naming the retention not applied.""" try: - overrides = json.loads(value) - except json.JSONDecodeError: - return value - if not isinstance(overrides, dict): - return value - per_retention = Counter(str(retention) for retention in overrides.values()) - return ", ".join(f"{retention} for {count} event types" for retention, count in sorted(per_retention.items())) + enabled = vacuum.enabled_vacuum_types + except ValueError as exc: + # Prefect checks the vacuum types only when its services read them, so an unknown one is named here and + # the failure is left to Prefect. + return [f"{VACUUM_ENABLED} is not valid: {exc}"] + warnings: list[str] = [] + if "flow_runs" not in enabled: + warnings.append( + f"{VACUUM_ENABLED} does not turn on flow_runs, so old task runs are not deleted and " + "task_manager.retention.task_history is not applied" + ) + if "events" not in enabled: + warnings.append( + f"{VACUUM_ENABLED} does not turn on events, so old events are not deleted and " + "task_manager.retention.activity_log and task_manager.retention.prefect_own_events are not applied" + ) + return warnings -def prefect_retention_env_in_effect(environ: Mapping[str, str]) -> dict[str, str]: - """Return the variable Prefect reads for each retention setting present in the environment, with its value. +def _vacuum_switch(enabled: set[str] | bool | None) -> str: + if isinstance(enabled, bool): + return "true" if enabled else "false" + if enabled is None: + return "null" + return ",".join(sorted(enabled)) or "[]" + + +def _count_overrides_per_retention(overrides: Mapping[str, timedelta]) -> str: + if not overrides: + return "{}" + per_retention = Counter(overrides.values()) + return ", ".join( + f"{_iso_8601(retention)} for {count} event types" for retention, count in sorted(per_retention.items()) + ) + + +def prefect_retention_in_effect(settings: Settings) -> dict[str, str]: + """Return the value Prefect applies for each retention setting, keyed by the variable Infrahub derives for it. The per-type overrides are counted per retention rather than listed, as the list holds over a hundred types. """ - in_effect: dict[str, str] = {} - for name, setting in _PREFECT_SETTINGS.items(): - name_set = _name_set_in(environ=environ, setting=setting) - if name_set is None: - continue - value = environ[name_set] - in_effect[name_set] = _count_overrides_per_retention(value) if name == EVENT_RETENTION_OVERRIDES else value - return in_effect + vacuum = settings.server.services.db_vacuum + return { + VACUUM_ENABLED: _vacuum_switch(enabled=vacuum.enabled), + VACUUM_RETENTION_PERIOD: _iso_8601(vacuum.retention_period), + EVENTS_RETENTION_PERIOD: _iso_8601(settings.server.events.retention_period), + EVENT_RETENTION_OVERRIDES: _count_overrides_per_retention(overrides=vacuum.event_retention_overrides), + } diff --git a/backend/infrahub/prefect_server/retention_settings.py b/backend/infrahub/prefect_server/retention_settings.py new file mode 100644 index 00000000000..bac3465ff33 --- /dev/null +++ b/backend/infrahub/prefect_server/retention_settings.py @@ -0,0 +1,167 @@ +from __future__ import annotations + +import re +import tomllib +from datetime import timedelta +from typing import TYPE_CHECKING, Any, Self + +from pydantic import ( + BaseModel, + ConfigDict, + Field, + TypeAdapter, + ValidationError, + ValidationInfo, + field_validator, + model_validator, +) +from pydantic_settings import BaseSettings, SettingsConfigDict + +if TYPE_CHECKING: + from pathlib import Path + +_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_RETENTION_PERIOD` takes precedence when set. When " + "`PREFECT_SERVER_SERVICES_DB_VACUUM_ENABLED` is set to a value that does not turn on `flow_runs`, old " + "task runs are not deleted." + ), + ) + 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, and at least `prefect_own_events`. " + "`PREFECT_SERVER_EVENTS_RETENTION_PERIOD`, or its legacy name `PREFECT_EVENTS_RETENTION_PERIOD`, takes " + "precedence when set. When `PREFECT_SERVER_SERVICES_DB_VACUUM_ENABLED` is set to a value that does not " + "turn on `events`, old events are not deleted." + ), + ) + 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, and at most `activity_log`. " + "`PREFECT_SERVER_SERVICES_DB_VACUUM_EVENT_RETENTION_OVERRIDES` takes precedence when set. When " + "`PREFECT_SERVER_SERVICES_DB_VACUUM_ENABLED` is set to a value that does not turn on `events`, old " + "events are not deleted." + ), + ) + + @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 + + @model_validator(mode="after") + def validate_prefect_own_events_within_activity_log(self) -> Self: + """Require Prefect's own events to be kept no longer than the activity log. + + Raises: + ValueError: When `prefect_own_events` is longer than `activity_log`. + + """ + if self.prefect_own_events > self.activity_log: + raise ValueError( + "Invalid task manager retention: prefect_own_events must not be longer than activity_log, " + "so set prefect_own_events to at most the activity_log value" + ) + return self + + +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 _TaskManagerSection(BaseModel): + model_config = ConfigDict(extra="ignore") + + task_manager: TaskManagerSettings = Field(default_factory=TaskManagerSettings) + + +def load_task_manager_settings(config_file: Path) -> TaskManagerSettings: + """Read the task manager settings from the configuration file and the environment, ignoring every other section. + + Raises: + ValidationError: When a task manager setting is invalid. + tomllib.TOMLDecodeError: When the configuration file is not valid TOML. + UnicodeDecodeError: When the configuration file is not UTF-8 text. + OSError: When the configuration file exists but cannot be read. + + """ + data = tomllib.loads(config_file.read_text(encoding="utf-8")) if config_file.exists() else {} + return _TaskManagerSection.model_validate(data).task_manager + + +def print_configuration_errors(error: ValidationError) -> None: + """Print each error of an invalid configuration with its location.""" + print(f"Configuration not valid, found {len(error.errors())} error(s)") + for details in error.errors(): + error_locations = [str(location) for location in details["loc"]] + print(f" {'/'.join(error_locations)} | {details['msg']} ({details['type']})") diff --git a/backend/tests/component/task_manager/test_event_retention.py b/backend/tests/component/task_manager/test_event_retention.py index 6c12ef24bd3..8d3b7542513 100644 --- a/backend/tests/component/task_manager/test_event_retention.py +++ b/backend/tests/component/task_manager/test_event_retention.py @@ -7,6 +7,7 @@ import pytest import sqlalchemy as sa from prefect.server.database import provide_database_interface +from prefect.server.database.dependencies import temporary_database_interface from prefect.server.services.db_vacuum import vacuum_events_with_retention_overrides, vacuum_old_events from prefect.settings import temporary_settings from prefect.settings.models.server.events import ServerEventsSettings @@ -20,7 +21,10 @@ from infrahub.prefect_server.retention import PREFECT_EVENT_TYPES, build_prefect_retention_env if TYPE_CHECKING: - from collections.abc import Generator + from collections.abc import AsyncGenerator, Generator + from pathlib import Path + + from prefect.server.database import PrefectDBInterface @pytest.fixture @@ -41,6 +45,21 @@ def year_of_activity_log(monkeypatch: pytest.MonkeyPatch) -> Generator[None, Non yield +@pytest.fixture +async def database(tmp_path: Path) -> AsyncGenerator[PrefectDBInterface, None]: + """An empty, migrated task manager database of the test's own, apart from the one the test Prefect server uses.""" + # Prefect keeps the first database configuration it builds, so a new connection URL alone is not enough. + with ( + temporary_database_interface(), + temporary_settings( + updates={"server.database.connection_url": f"sqlite+aiosqlite:///{tmp_path / 'prefect.db'}"} + ), + ): + db = provide_database_interface() + await db.create_db() + yield db + + def _branch_created_event() -> BranchCreatedEvent: branch_id = uuid.uuid4() return BranchCreatedEvent( @@ -52,7 +71,7 @@ def _branch_created_event() -> BranchCreatedEvent: @pytest.mark.usefixtures("year_of_activity_log") -async def test_prefect_own_events_expire_before_the_activity_log() -> None: +async def test_prefect_own_events_expire_before_the_activity_log(database: PrefectDBInterface) -> None: """Listed Prefect events go after 7 days with their related items, Infrahub and unlisted events after a year.""" now = datetime.now(UTC) kept = { @@ -66,15 +85,16 @@ async def test_prefect_own_events_expire_before_the_activity_log() -> None: await seed_prefect_event(event_name="prefect.task-run.Running", occurred=now - timedelta(days=8)), } seeded = kept | deleted - db = provide_database_interface() - await vacuum_events_with_retention_overrides(db=db) - await vacuum_old_events(db=db) + await vacuum_events_with_retention_overrides(db=database) + await vacuum_old_events(db=database) - async with db.session_context() as session: - remaining_events = set(await session.scalars(sa.select(db.Event.id).where(db.Event.id.in_(seeded)))) + async with database.session_context() as session: + remaining_events = set(await session.scalars(sa.select(database.Event.id).where(database.Event.id.in_(seeded)))) events_with_resources = set( - await session.scalars(sa.select(db.EventResource.event_id).where(db.EventResource.event_id.in_(seeded))) + await session.scalars( + sa.select(database.EventResource.event_id).where(database.EventResource.event_id.in_(seeded)) + ) ) assert remaining_events == kept assert events_with_resources == kept diff --git a/backend/tests/unit/config/test_task_manager_retention_settings.py b/backend/tests/unit/config/test_task_manager_retention_settings.py index b61551d81cf..eb1c4bad75d 100644 --- a/backend/tests/unit/config/test_task_manager_retention_settings.py +++ b/backend/tests/unit/config/test_task_manager_retention_settings.py @@ -9,11 +9,13 @@ from pydantic import ValidationError from infrahub.config import TaskManagerRetentionSettings, load +from infrahub.prefect_server.retention_settings import load_task_manager_settings if TYPE_CHECKING: from pathlib import Path RETENTION_ENV_VARIABLES = ( + "INFRAHUB_TASK_MANAGER_RETENTION", "INFRAHUB_TASK_MANAGER_RETENTION_TASK_HISTORY", "INFRAHUB_TASK_MANAGER_RETENTION_ACTIVITY_LOG", "INFRAHUB_TASK_MANAGER_RETENTION_PREFECT_OWN_EVENTS", @@ -59,7 +61,10 @@ class AcceptedRetentionTestCase: @pytest.mark.parametrize("test_case", [pytest.param(tc, id=tc.name) for tc in ACCEPTED_RETENTION_TEST_CASES]) @pytest.mark.parametrize("key", ["task_history", "activity_log", "prefect_own_events"]) def test_retention_accepts_days_and_iso_8601_durations(key: str, test_case: AcceptedRetentionTestCase) -> None: - settings = TaskManagerRetentionSettings.model_validate({key: test_case.value}) + # With activity_log at its maximum and prefect_own_events at its minimum, no accepted value makes the second longer. + settings = TaskManagerRetentionSettings.model_validate( + {"activity_log": "36500d", "prefect_own_events": "1d", key: test_case.value} + ) assert getattr(settings, key) == test_case.expected @@ -111,11 +116,44 @@ def test_retention_refusal_names_the_setting(key: str, test_case: RefusedRetenti TaskManagerRetentionSettings.model_validate({key: test_case.value}) -def test_own_events_longer_than_activity_log_is_accepted() -> None: - settings = TaskManagerRetentionSettings(activity_log="7d", prefect_own_events="30d") +@dataclass +class OwnEventsLongerThanActivityLogTestCase: + name: str + retention: dict[str, str] - assert settings.activity_log == timedelta(days=7) - assert settings.prefect_own_events == timedelta(days=30) + +OWN_EVENTS_LONGER_THAN_ACTIVITY_LOG_TEST_CASES: list[OwnEventsLongerThanActivityLogTestCase] = [ + OwnEventsLongerThanActivityLogTestCase( + name="own_events_raised_above_the_activity_log", + retention={"activity_log": "7d", "prefect_own_events": "30d"}, + ), + OwnEventsLongerThanActivityLogTestCase( + name="activity_log_lowered_below_the_own_events_default", retention={"activity_log": "3d"} + ), + OwnEventsLongerThanActivityLogTestCase( + name="longer_by_one_second", retention={"activity_log": "P7D", "prefect_own_events": "P7DT1S"} + ), +] + + +@pytest.mark.parametrize( + "test_case", [pytest.param(tc, id=tc.name) for tc in OWN_EVENTS_LONGER_THAN_ACTIVITY_LOG_TEST_CASES] +) +def test_own_events_longer_than_activity_log_are_refused(test_case: OwnEventsLongerThanActivityLogTestCase) -> None: + with pytest.raises( + ValidationError, + match=re.escape( + "Invalid task manager retention: prefect_own_events must not be longer than activity_log, " + "so set prefect_own_events to at most the activity_log value" + ), + ): + TaskManagerRetentionSettings.model_validate(test_case.retention) + + +def test_own_events_as_long_as_activity_log_are_accepted() -> None: + settings = TaskManagerRetentionSettings(activity_log="30d", prefect_own_events="30d") + + assert (settings.activity_log, settings.prefect_own_events) == (timedelta(days=30), timedelta(days=30)) def test_environment_variables_set_the_retention(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: @@ -153,3 +191,80 @@ def test_invalid_environment_variable_fails_the_configuration_load( ValidationError, match=re.escape("Invalid task manager retention: activity_log must be at least 1 day") ): load(config_file_name=tmp_path / "absent.toml") + + +@dataclass +class TaskManagerSectionTestCase: + name: str + file_content: str | None + environ: dict[str, str] + task_history: timedelta + activity_log: timedelta + prefect_own_events: timedelta + + +TASK_MANAGER_SECTION_TEST_CASES: list[TaskManagerSectionTestCase] = [ + TaskManagerSectionTestCase( + name="no_file", + file_content=None, + environ={"INFRAHUB_TASK_MANAGER_RETENTION_ACTIVITY_LOG": "14d"}, + task_history=timedelta(days=30), + activity_log=timedelta(days=14), + prefect_own_events=timedelta(days=7), + ), + TaskManagerSectionTestCase( + name="file_only", + file_content='[task_manager.retention]\ntask_history = "45d"\n', + environ={}, + task_history=timedelta(days=45), + activity_log=timedelta(days=7), + prefect_own_events=timedelta(days=7), + ), + TaskManagerSectionTestCase( + name="file_and_environment_for_another_field", + file_content='[task_manager.retention]\ntask_history = "45d"\n', + environ={"INFRAHUB_TASK_MANAGER_RETENTION_ACTIVITY_LOG": "14d"}, + task_history=timedelta(days=45), + activity_log=timedelta(days=14), + prefect_own_events=timedelta(days=7), + ), + TaskManagerSectionTestCase( + name="file_over_environment_for_the_same_field", + file_content='[task_manager.retention]\ntask_history = "45d"\n', + environ={"INFRAHUB_TASK_MANAGER_RETENTION_TASK_HISTORY": "60d"}, + task_history=timedelta(days=45), + activity_log=timedelta(days=7), + prefect_own_events=timedelta(days=7), + ), + TaskManagerSectionTestCase( + name="json_environment_variable", + file_content="[main]\n", + environ={ + "INFRAHUB_TASK_MANAGER_RETENTION": '{"task_history": "60d"}', + "INFRAHUB_TASK_MANAGER_RETENTION_ACTIVITY_LOG": "14d", + }, + task_history=timedelta(days=60), + activity_log=timedelta(days=14), + prefect_own_events=timedelta(days=7), + ), +] + + +@pytest.mark.parametrize("test_case", [pytest.param(tc, id=tc.name) for tc in TASK_MANAGER_SECTION_TEST_CASES]) +def test_task_manager_section_reads_like_the_whole_configuration( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, test_case: TaskManagerSectionTestCase +) -> None: + config_file = tmp_path / "infrahub.toml" + if test_case.file_content is not None: + config_file.write_text(test_case.file_content, encoding="utf-8") + for name, value in test_case.environ.items(): + monkeypatch.setenv(name, value) + + retention = load_task_manager_settings(config_file=config_file).retention + + assert (retention.task_history, retention.activity_log, retention.prefect_own_events) == ( + test_case.task_history, + test_case.activity_log, + test_case.prefect_own_events, + ) + assert retention == load(config_file_name=config_file).task_manager.retention diff --git a/backend/tests/unit/prefect_server/test_app_retention.py b/backend/tests/unit/prefect_server/test_app_retention.py index d4ed2f8b399..9aa92e27e21 100644 --- a/backend/tests/unit/prefect_server/test_app_retention.py +++ b/backend/tests/unit/prefect_server/test_app_retention.py @@ -1,7 +1,12 @@ from __future__ import annotations +import json import logging import os +import subprocess # noqa: S404 +import sys +import tomllib +from dataclasses import dataclass from datetime import timedelta from typing import TYPE_CHECKING @@ -32,19 +37,31 @@ EVENT_RETENTION_OVERRIDES, ) +# Prefect skips the profile while any of these is set. +PREFECT_TEST_MODE_VARIABLES = ( + "PREFECT_TEST_MODE", + "PREFECT_UNIT_TEST_MODE", + "PREFECT_TESTING_UNIT_TEST_MODE", + "PREFECT_TESTING_TEST_MODE", +) + @pytest.fixture def task_manager_environment(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> Generator[None, None, None]: - """Start from an unloaded configuration and no retention variable, and put the process state back afterwards.""" + """Start with no configuration loaded and no Prefect setting from any source, then restore the process state.""" monkeypatch.setattr(config.SETTINGS, "settings", None) monkeypatch.setattr(prefect.context, "GLOBAL_SETTINGS_CONTEXT", prefect.context.GLOBAL_SETTINGS_CONTEXT) + monkeypatch.chdir(tmp_path) monkeypatch.setenv("INFRAHUB_CONFIG", str(tmp_path / "absent.toml")) + monkeypatch.setenv("PREFECT_PROFILES_PATH", str(tmp_path / "profiles.toml")) for name in ( "INFRAHUB_TASK_MANAGER_RETENTION_TASK_HISTORY", "INFRAHUB_TASK_MANAGER_RETENTION_ACTIVITY_LOG", "INFRAHUB_TASK_MANAGER_RETENTION_PREFECT_OWN_EVENTS", "PREFECT_API_BLOCKS_REGISTER_ON_START", "PREFECT_API_DATABASE_MIGRATE_ON_START", + "PREFECT__SERVER_WEBSERVER_ONLY", + *PREFECT_TEST_MODE_VARIABLES, ): monkeypatch.delenv(name, raising=False) saved = {name: os.environ.pop(name, None) for name in PREFECT_RETENTION_VARIABLES} @@ -83,36 +100,89 @@ def test_task_manager_refuses_to_start_on_a_retention_under_one_day( @pytest.mark.usefixtures("task_manager_environment") -def test_capped_own_events_are_logged(monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture) -> None: - monkeypatch.setenv("INFRAHUB_TASK_MANAGER_RETENTION_ACTIVITY_LOG", "7d") - monkeypatch.setenv("INFRAHUB_TASK_MANAGER_RETENTION_PREFECT_OWN_EVENTS", "30d") +def test_each_preset_prefect_variable_is_logged_and_kept( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + monkeypatch.setenv(VACUUM_RETENTION_PERIOD, "P2D") + monkeypatch.setenv(EVENTS_RETENTION_PERIOD, "P3D") with caplog.at_level(logging.WARNING, logger=APP_LOGGER): apply_infrahub_settings_to_prefect() assert _app_messages(caplog=caplog, level=logging.WARNING) == [ - "task_manager.retention.prefect_own_events P30D is longer than task_manager.retention.activity_log P7D, " - "Prefect's own events are kept for P7D" + f"{VACUUM_RETENTION_PERIOD} is set and overrides task_manager.retention.task_history", + f"{EVENTS_RETENTION_PERIOD} is set and overrides task_manager.retention.activity_log", ] + prefect_settings = get_current_settings() + assert prefect_settings.server.services.db_vacuum.retention_period == timedelta(days=2) + assert prefect_settings.server.events.retention_period == timedelta(days=3) @pytest.mark.usefixtures("task_manager_environment") -def test_each_preset_prefect_variable_is_logged_and_kept( +def test_empty_vacuum_switch_keeps_task_run_deletion( monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture ) -> None: - monkeypatch.setenv(VACUUM_RETENTION_PERIOD, "P2D") - monkeypatch.setenv(EVENTS_RETENTION_PERIOD, "P3D") + """An empty switch, as a compose file passes for an unset host variable, is replaced with the derived value.""" + monkeypatch.setenv(VACUUM_ENABLED, "") + + with caplog.at_level(logging.WARNING, logger=APP_LOGGER): + apply_infrahub_settings_to_prefect() + + assert _app_messages(caplog=caplog, level=logging.WARNING) == [] + assert os.environ[VACUUM_ENABLED] == "events,flow_runs" + assert get_current_settings().server.services.db_vacuum.enabled_vacuum_types == {"events", "flow_runs", "orphans"} + + +@pytest.mark.usefixtures("task_manager_environment") +def test_empty_vacuum_switch_gives_way_to_a_prefect_settings_file( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + """Prefect reads an empty switch as false, which would replace the file value and turn off task run deletion.""" + (tmp_path / "prefect.toml").write_text('[server.services.db_vacuum]\nenabled = "flow_runs"\n', encoding="utf-8") + monkeypatch.setenv(VACUUM_ENABLED, "") with caplog.at_level(logging.WARNING, logger=APP_LOGGER): apply_infrahub_settings_to_prefect() assert _app_messages(caplog=caplog, level=logging.WARNING) == [ - f"{VACUUM_RETENTION_PERIOD} is set and overrides task_manager.retention.task_history", - f"{EVENTS_RETENTION_PERIOD} is set and overrides task_manager.retention.activity_log", + "PREFECT_SERVER_SERVICES_DB_VACUUM_ENABLED does not turn on events, so old events are not deleted and " + "task_manager.retention.activity_log and task_manager.retention.prefect_own_events are not applied" ] - prefect_settings = get_current_settings() - assert prefect_settings.server.services.db_vacuum.retention_period == timedelta(days=2) - assert prefect_settings.server.events.retention_period == timedelta(days=3) + assert VACUUM_ENABLED not in os.environ + assert get_current_settings().server.services.db_vacuum.enabled_vacuum_types == {"flow_runs", "orphans"} + + +@pytest.mark.usefixtures("task_manager_environment") +def test_preset_vacuum_switch_with_flow_runs_keeps_the_task_history_retention( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + """A preset vacuum switch only turns cleanups on or off, so task_history still sets how long task runs are kept.""" + monkeypatch.setenv(VACUUM_ENABLED, "true") + + with caplog.at_level(logging.WARNING, logger=APP_LOGGER): + apply_infrahub_settings_to_prefect() + + assert _app_messages(caplog=caplog, level=logging.WARNING) == [] + assert os.environ[VACUUM_ENABLED] == "true" + vacuum = get_current_settings().server.services.db_vacuum + assert vacuum.enabled_vacuum_types == {"events", "flow_runs", "orphans"} + assert vacuum.retention_period == timedelta(days=30) + + +@pytest.mark.usefixtures("task_manager_environment") +def test_preset_vacuum_switch_that_turns_off_a_cleanup_is_logged( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + monkeypatch.setenv(VACUUM_ENABLED, "flow_runs") + + with caplog.at_level(logging.WARNING, logger=APP_LOGGER): + apply_infrahub_settings_to_prefect() + + assert _app_messages(caplog=caplog, level=logging.WARNING) == [ + "PREFECT_SERVER_SERVICES_DB_VACUUM_ENABLED does not turn on events, so old events are not deleted and " + "task_manager.retention.activity_log and task_manager.retention.prefect_own_events are not applied" + ] + assert get_current_settings().server.services.db_vacuum.enabled_vacuum_types == {"flow_runs", "orphans"} @pytest.mark.usefixtures("task_manager_environment") @@ -127,14 +197,122 @@ def test_prefect_runs_with_the_configured_retention(monkeypatch: pytest.MonkeyPa assert prefect_settings.server.services.db_vacuum.enabled_vacuum_types == {"events", "flow_runs", "orphans"} assert prefect_settings.server.services.db_vacuum.retention_period == timedelta(days=45) assert prefect_settings.server.events.retention_period == timedelta(days=14) - assert config.SETTINGS.task_manager.retention.task_history == timedelta(days=45) @pytest.mark.usefixtures("task_manager_environment") -def test_startup_logs_the_retention_variables_prefect_reads( +def test_a_setting_outside_the_task_manager_section_does_not_stop_the_task_manager( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + config_file = tmp_path / "infrahub.toml" + config_file.write_text( + '[git]\ndelete_git_branch_after_merge = true\n\n[storage]\ndriver = "nope"\n\n' + '[task_manager.retention]\ntask_history = "45d"\n', + encoding="utf-8", + ) + monkeypatch.setenv("INFRAHUB_CONFIG", str(config_file)) + + apply_infrahub_settings_to_prefect() + + assert get_current_settings().server.services.db_vacuum.retention_period == timedelta(days=45) + assert config.SETTINGS.initialized is False + + +@pytest.mark.usefixtures("task_manager_environment") +def test_an_invalid_retention_in_the_configuration_file_stops_the_task_manager( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + config_file = tmp_path / "infrahub.toml" + config_file.write_text( + '[storage]\ndriver = "nope"\n\n[task_manager.retention]\ntask_history = "PT12H"\n', encoding="utf-8" + ) + monkeypatch.setenv("INFRAHUB_CONFIG", str(config_file)) + + with pytest.raises(SystemExit) as exit_info: + apply_infrahub_settings_to_prefect() + + assert exit_info.value.code == 1 + assert capsys.readouterr().out == ( + "Configuration not valid, found 1 error(s)\n" + " task_manager/retention/task_history | Value error, Invalid task manager retention: task_history must be " + "at least 1 day (value_error)\n" + ) + assert {name: os.environ.get(name) for name in PREFECT_RETENTION_VARIABLES} == dict.fromkeys( + PREFECT_RETENTION_VARIABLES + ) + + +@pytest.mark.usefixtures("task_manager_environment") +def test_a_configuration_file_that_cannot_be_parsed_stops_the_task_manager( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + config_file = tmp_path / "infrahub.toml" + config_file.write_text("[task_manager.retention\n", encoding="utf-8") + monkeypatch.setenv("INFRAHUB_CONFIG", str(config_file)) + + with pytest.raises( + tomllib.TOMLDecodeError, match=r"^Expected '\]' at the end of a table declaration \(at line 1, column 24\)$" + ): + apply_infrahub_settings_to_prefect() + + assert {name: os.environ.get(name) for name in PREFECT_RETENTION_VARIABLES} == dict.fromkeys( + PREFECT_RETENTION_VARIABLES + ) + + +_STANDALONE_TASK_MANAGER_SCRIPT = """ +import sys + +from prefect.settings import get_current_settings + +from infrahub.prefect_server.app import apply_infrahub_settings_to_prefect + +apply_infrahub_settings_to_prefect() +retention_period = get_current_settings().server.services.db_vacuum.retention_period +print("infrahub.config" in sys.modules, "infrahub.log" in sys.modules, retention_period.days) +""" + + +def test_the_standalone_task_manager_ignores_infrahub_settings_it_does_not_use(tmp_path: Path) -> None: + """A fresh process is needed because this one has already imported Infrahub's configuration.""" + environ = { + name: value + for name, value in os.environ.items() + if name not in PREFECT_RETENTION_VARIABLES + and not name.startswith(("INFRAHUB_TASK_MANAGER_", "PREFECT_API_BLOCKS_", "PREFECT_API_DATABASE_")) + } + environ |= { + "INFRAHUB_CONFIG": str(tmp_path / "absent.toml"), + "PREFECT_PROFILES_PATH": str(tmp_path / "profiles.toml"), + "INFRAHUB_LOG_LEVEL": "INFO", + "INFRAHUB_PRODUCTION": "true", + "INFRAHUB_STORAGE_DRIVER": "nope", + "INFRAHUB_TASK_MANAGER_RETENTION_TASK_HISTORY": "45d", + } + + result = subprocess.run( # noqa: S603 + [sys.executable, "-c", _STANDALONE_TASK_MANAGER_SCRIPT], + cwd=tmp_path, + env=environ, + capture_output=True, + text=True, + check=False, + timeout=60, + ) + + assert (result.returncode, result.stdout) == (0, "False True 45\n"), result.stderr + task_history_log_lines = [ + line["event"] + for line in (json.loads(text) for text in result.stderr.splitlines() if text.startswith("{")) + if line.get("logger") == APP_LOGGER and VACUUM_RETENTION_PERIOD in line["event"] + ] + assert task_history_log_lines == [f"Task manager retention: {VACUUM_RETENTION_PERIOD}=P45D"] + + +@pytest.mark.usefixtures("task_manager_environment") +def test_startup_logs_the_retention_prefect_applies( monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture ) -> None: - """The startup log shows the value of each retention variable, whether Infrahub derived it or it was preset.""" + """The startup log shows the value Prefect applies for each retention, derived by Infrahub or preset.""" monkeypatch.setenv("INFRAHUB_TASK_MANAGER_RETENTION_TASK_HISTORY", "45d") monkeypatch.setenv("INFRAHUB_TASK_MANAGER_RETENTION_ACTIVITY_LOG", "P14D") monkeypatch.setenv(EVENT_RETENTION_OVERRIDES, '{"prefect.flow-run.heartbeat": "P1D"}') @@ -151,10 +329,25 @@ def test_startup_logs_the_retention_variables_prefect_reads( @pytest.mark.usefixtures("task_manager_environment") -def test_preset_legacy_variable_is_logged_and_applied_under_its_own_name( +def test_default_retention_writes_no_per_type_override(caplog: pytest.LogCaptureFixture) -> None: + """With equal own-event and activity log retentions, Prefect's general event cleanup alone applies both.""" + with caplog.at_level(logging.INFO, logger=APP_LOGGER): + apply_infrahub_settings_to_prefect() + + assert _app_messages(caplog=caplog, level=logging.INFO) == [ + f"Task manager retention: {VACUUM_ENABLED}=events,flow_runs", + f"Task manager retention: {VACUUM_RETENTION_PERIOD}=P30D", + f"Task manager retention: {EVENTS_RETENTION_PERIOD}=P7D", + f"Task manager retention: {EVENT_RETENTION_OVERRIDES}={{}}", + ] + assert get_current_settings().server.services.db_vacuum.event_retention_overrides == {} + + +@pytest.mark.usefixtures("task_manager_environment") +def test_preset_legacy_variable_is_named_in_the_warning_and_its_value_logged( monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture ) -> None: - """A retention preset under Prefect's legacy variable name is the one Prefect applies, and both logs name it.""" + """A retention preset under Prefect's legacy variable name is the one Prefect applies, and the warning names it.""" monkeypatch.setenv(LEGACY_EVENTS_RETENTION_PERIOD, "P3D") monkeypatch.setenv(EVENT_RETENTION_OVERRIDES, '{"prefect.flow-run.heartbeat": "P1D"}') @@ -168,8 +361,51 @@ def test_preset_legacy_variable_is_logged_and_applied_under_its_own_name( assert _app_messages(caplog=caplog, level=logging.INFO) == [ f"Task manager retention: {VACUUM_ENABLED}=events,flow_runs", f"Task manager retention: {VACUUM_RETENTION_PERIOD}=P30D", - f"Task manager retention: {LEGACY_EVENTS_RETENTION_PERIOD}=P3D", + f"Task manager retention: {EVENTS_RETENTION_PERIOD}=P3D", f"Task manager retention: {EVENT_RETENTION_OVERRIDES}=P1D for 1 event types", ] assert EVENTS_RETENTION_PERIOD not in os.environ assert get_current_settings().server.events.retention_period == timedelta(days=3) + + +@dataclass +class PrefectSettingsFileTestCase: + name: str + file_name: str + content: str + + +PREFECT_SETTINGS_FILE_TEST_CASES: list[PrefectSettingsFileTestCase] = [ + PrefectSettingsFileTestCase( + name="prefect_toml", file_name="prefect.toml", content='[server.events]\nretention_period = "P90D"\n' + ), + PrefectSettingsFileTestCase(name="dotenv", file_name=".env", content=f"{EVENTS_RETENTION_PERIOD}=P90D\n"), + PrefectSettingsFileTestCase( + name="active_profile", + file_name="profiles.toml", + content=f'active = "infrahub"\n\n[profiles.infrahub]\n{EVENTS_RETENTION_PERIOD} = "P90D"\n', + ), +] + + +@pytest.mark.parametrize("test_case", [pytest.param(tc, id=tc.name) for tc in PREFECT_SETTINGS_FILE_TEST_CASES]) +@pytest.mark.usefixtures("task_manager_environment") +def test_retention_set_in_a_prefect_settings_file_is_kept_and_logged( + tmp_path: Path, caplog: pytest.LogCaptureFixture, test_case: PrefectSettingsFileTestCase +) -> None: + """An environment variable would replace a value Prefect reads from its own files, so none is written for it.""" + (tmp_path / test_case.file_name).write_text(test_case.content, encoding="utf-8") + + with caplog.at_level(logging.INFO, logger=APP_LOGGER): + apply_infrahub_settings_to_prefect() + + assert _app_messages(caplog=caplog, level=logging.WARNING) == [ + f"{EVENTS_RETENTION_PERIOD} is set in a Prefect settings file or profile and overrides " + "task_manager.retention.activity_log" + ] + assert EVENTS_RETENTION_PERIOD not in os.environ + assert get_current_settings().server.events.retention_period == timedelta(days=90) + events_retention_log_lines = [ + message for message in _app_messages(caplog=caplog, level=logging.INFO) if EVENTS_RETENTION_PERIOD in message + ] + assert events_retention_log_lines == [f"Task manager retention: {EVENTS_RETENTION_PERIOD}=P90D"] diff --git a/backend/tests/unit/prefect_server/test_prefect_event_types.py b/backend/tests/unit/prefect_server/test_prefect_event_types.py index f572df8e71e..a4496f3047b 100644 --- a/backend/tests/unit/prefect_server/test_prefect_event_types.py +++ b/backend/tests/unit/prefect_server/test_prefect_event_types.py @@ -1,32 +1,186 @@ from __future__ import annotations +import ast import inspect +from dataclasses import dataclass +from functools import cache +from pathlib import Path +from typing import TYPE_CHECKING +import prefect from prefect import states as client_states +from prefect.server.events.schemas.automations import TriggerState from prefect.server.schemas import states as server_states +from prefect.server.schemas.statuses import DeploymentStatus, WorkPoolStatus, WorkQueueStatus from infrahub.prefect_server.retention import PREFECT_EVENT_TYPES +if TYPE_CHECKING: + from collections.abc import Callable + from types import ModuleType + +_STATE_MODULES = (client_states, server_states) + + +@dataclass(frozen=True) +class _PrefectSourceNames: + event_names: frozenset[str] + event_name_patterns: frozenset[str] + """Event names built from f-strings, with each placeholder written as `{}`.""" + state_names: frozenset[str] + lifecycle_kinds: frozenset[str] + """Kinds of record that Prefect sends created, updated and deleted events for.""" + + +def _state_constructors( + module: ModuleType, +) -> list[tuple[str, Callable[[], client_states.State | server_states.State]]]: + return [ + (name, constructor) + for name, constructor in inspect.getmembers(module, inspect.isfunction) + if constructor.__module__ == module.__name__ and name[0].isupper() + ] + + +def _callee_name(call: ast.Call) -> str | None: + if isinstance(call.func, ast.Name): + return call.func.id + if isinstance(call.func, ast.Attribute): + return call.func.attr + return None + + +def _keyword_values(call: ast.Call, *names: str) -> list[ast.expr]: + return [keyword.value for keyword in call.keywords if keyword.arg in names] + + +def _is_state_name_attribute(target: ast.expr) -> bool: + return isinstance(target, ast.Attribute) and target.attr == "name" and "state" in ast.unparse(target.value).lower() + + +def _string_literals(expressions: list[ast.expr]) -> frozenset[str]: + return frozenset( + expression.value + for expression in expressions + if isinstance(expression, ast.Constant) and isinstance(expression.value, str) + ) + + +def _f_string_patterns(expressions: list[ast.expr]) -> frozenset[str]: + return frozenset( + "".join(str(part.value) if isinstance(part, ast.Constant) else "{}" for part in expression.values) + for expression in expressions + if isinstance(expression, ast.JoinedStr) + ) + + +@cache +def _names_in_prefect_source() -> _PrefectSourceNames: + """Read the event names and name patterns, state names and lifecycle event kinds written in Prefect's source.""" + state_constructors = {"State", *(name for module in _STATE_MODULES for name, _ in _state_constructors(module))} + events: list[ast.expr] = [] + state_names: list[ast.expr] = [] + lifecycle_kinds: list[ast.expr] = [] + for path in Path(prefect.__file__).parent.rglob("*.py"): + for node in ast.walk(ast.parse(path.read_text(encoding="utf-8"))): + if isinstance(node, ast.Assign): + state_names.extend(node.value for target in node.targets if _is_state_name_attribute(target)) + continue + if not isinstance(node, ast.Call): + continue + callee = _callee_name(node) + events.extend(_keyword_values(node, "event", "event_type")) + if callee == "emit_event": + events.extend(node.args[:1]) + if callee in state_constructors: + state_names.extend(_keyword_values(node, "name")) + if callee == "rename_state": + state_names.extend(node.args[:1]) + if callee == "_lifecycle_event": + lifecycle_kinds.extend(_keyword_values(node, "kind")) + return _PrefectSourceNames( + event_names=_string_literals(events), + event_name_patterns=_f_string_patterns(events), + state_names=_string_literals(state_names), + lifecycle_kinds=_string_literals(lifecycle_kinds), + ) + + +def test_every_prefect_event_name_in_prefect_source_is_listed() -> None: + """Every `prefect.*` event name that Prefect's source sends as a literal is in the list of Prefect event types.""" + event_names = _names_in_prefect_source().event_names + + # Two names sent from different parts of Prefect, so a reader that finds nothing cannot pass. + assert {"prefect.flow-run.heartbeat", "prefect.worker.started"} <= event_names + assert sorted(name for name in event_names if name.startswith("prefect.") and name not in PREFECT_EVENT_TYPES) == [] + + +def test_prefect_builds_event_names_from_known_patterns_only() -> None: + """Prefect's source builds event names from these patterns only. + + A new pattern fails this test until its names are listed and the pattern is added here. The other tests do not read + the lifecycle actions, concurrency limit phases or asset materialization statuses put into these patterns. + """ + assert _names_in_prefect_source().event_name_patterns == { + "prefect.flow-run.{}", + "prefect.task-run.{}", + "prefect.deployment.{}", + "prefect.work-pool.{}", + "prefect.work-queue.{}", + "prefect.automation.{}", + "prefect.concurrency-limit.{}", + "prefect.concurrency-limit.v1.{}", + "prefect.asset.materialization.{}", + "prefect.{}.{}", + "{}.loaded", + } + + +def test_every_prefect_status_and_trigger_state_has_its_event_listed() -> None: + """Every deployment, work pool and work queue status and every automation trigger state has its event listed.""" + event_types = { + *(f"prefect.deployment.{status.in_kebab_case()}" for status in DeploymentStatus), + *(f"prefect.work-pool.{status.in_kebab_case()}" for status in WorkPoolStatus), + *(f"prefect.work-queue.{status.in_kebab_case()}" for status in WorkQueueStatus), + *(f"prefect.automation.{state.value.lower()}" for state in TriggerState), + } + + assert sorted(event_types - set(PREFECT_EVENT_TYPES)) == [] + + +def test_every_prefect_lifecycle_event_is_listed() -> None: + """Every kind of record that Prefect's source sends lifecycle events for has its lifecycle events listed.""" + kinds = _names_in_prefect_source().lifecycle_kinds + + # Two kinds of record, so a reader that finds nothing cannot pass. + assert {"flow", "automation"} <= kinds + assert ( + sorted( + event_type + for kind in kinds + for event_type in (f"prefect.{kind}.created", f"prefect.{kind}.updated", f"prefect.{kind}.deleted") + if event_type not in PREFECT_EVENT_TYPES + ) + == [] + ) + def _state_names_defined_by_prefect() -> set[str]: names: set[str] = set() - for module in (client_states, server_states): + for module in _STATE_MODULES: states = [module.State(type=state_type) for state_type in module.StateType] - states.extend( - constructor() - for name, constructor in inspect.getmembers(module, inspect.isfunction) - if constructor.__module__ == module.__name__ and name[0].isupper() - ) + states.extend(constructor() for _, constructor in _state_constructors(module)) names.update(state.name for state in states if state.name is not None) return names def test_every_built_in_prefect_state_has_its_run_events_listed() -> None: - """Every state name Prefect defines has its flow-run and task-run event in the list of Prefect event types.""" - state_names = _state_names_defined_by_prefect() + """Every state name Prefect defines or sets in its source has its flow-run and task-run event listed.""" + state_names = _state_names_defined_by_prefect() | _names_in_prefect_source().state_names - # One name from a state type and one from a named constructor, so a reader that finds nothing cannot pass. - assert {"Completed", "AwaitingRetry"} <= state_names + # A name from a state type, one from a named constructor and two that Prefect's source sets on a state, so a + # reader that finds nothing cannot pass. + assert {"Completed", "AwaitingRetry", "Cached", "TimedOut"} <= state_names assert ( sorted( event_type diff --git a/backend/tests/unit/prefect_server/test_retention.py b/backend/tests/unit/prefect_server/test_retention.py index 2ce7d44f874..f581b8276e4 100644 --- a/backend/tests/unit/prefect_server/test_retention.py +++ b/backend/tests/unit/prefect_server/test_retention.py @@ -3,8 +3,10 @@ import json from dataclasses import dataclass from datetime import timedelta +from typing import TYPE_CHECKING import pytest +from prefect.settings import Settings from prefect.settings.models.server.events import ServerEventsSettings from prefect.settings.models.server.services import ServerServicesDBVacuumSettings @@ -12,18 +14,49 @@ from infrahub.prefect_server.retention import ( PREFECT_EVENT_TYPES, apply_prefect_retention_env, + background_services_warning, build_prefect_retention_env, - prefect_retention_env_in_effect, + prefect_retention_already_set, + prefect_retention_in_effect, + remove_blank_vacuum_switch, + turned_off_cleanup_warnings, ) +if TYPE_CHECKING: + from pathlib import Path + VACUUM_ENABLED = "PREFECT_SERVER_SERVICES_DB_VACUUM_ENABLED" VACUUM_RETENTION_PERIOD = "PREFECT_SERVER_SERVICES_DB_VACUUM_RETENTION_PERIOD" EVENTS_RETENTION_PERIOD = "PREFECT_SERVER_EVENTS_RETENTION_PERIOD" EVENT_RETENTION_OVERRIDES = "PREFECT_SERVER_SERVICES_DB_VACUUM_EVENT_RETENTION_OVERRIDES" +WEBSERVER_ONLY = "PREFECT__SERVER_WEBSERVER_ONLY" + +LEGACY_EVENTS_RETENTION_PERIOD = "PREFECT_EVENTS_RETENTION_PERIOD" +EVERY_RETENTION_VARIABLE = frozenset( + {VACUUM_ENABLED, VACUUM_RETENTION_PERIOD, EVENTS_RETENTION_PERIOD, EVENT_RETENTION_OVERRIDES} +) + +# Prefect skips the profile while any of these is set. +PREFECT_TEST_MODE_VARIABLES = ( + "PREFECT_TEST_MODE", + "PREFECT_UNIT_TEST_MODE", + "PREFECT_TESTING_UNIT_TEST_MODE", + "PREFECT_TESTING_TEST_MODE", +) EVENT_TYPES = ("prefect.flow-run.Completed", "prefect.task-run.Running") +@pytest.fixture +def prefect_settings_sources(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> Path: + """Give Prefect an empty working directory, an absent profiles file and no retention variable.""" + monkeypatch.chdir(tmp_path) + monkeypatch.setenv("PREFECT_PROFILES_PATH", str(tmp_path / "profiles.toml")) + for name in (*EVERY_RETENTION_VARIABLE, LEGACY_EVENTS_RETENTION_PERIOD, *PREFECT_TEST_MODE_VARIABLES): + monkeypatch.delenv(name, raising=False) + return tmp_path + + def _with_parsed_overrides(environ: dict[str, str]) -> dict[str, object]: return {**environ, EVENT_RETENTION_OVERRIDES: json.loads(environ[EVENT_RETENTION_OVERRIDES])} @@ -41,17 +74,6 @@ def test_variables_follow_the_retention() -> None: } -def test_own_events_are_capped_to_the_activity_log() -> None: - settings = TaskManagerRetentionSettings(task_history="30d", activity_log="7d", prefect_own_events="30d") - - variables = build_prefect_retention_env(settings=settings, event_types=EVENT_TYPES) - - assert json.loads(variables[EVENT_RETENTION_OVERRIDES]) == { - "prefect.flow-run.Completed": "P7D", - "prefect.task-run.Running": "P7D", - } - - def test_durations_keep_their_part_below_a_day() -> None: settings = TaskManagerRetentionSettings(task_history="P1DT12H", activity_log="P400D", prefect_own_events="P1DT0.5S") @@ -68,9 +90,10 @@ class PrefectReadTestCase: task_history: str activity_log: str prefect_own_events: str + event_types: tuple[str, ...] expected_vacuum_retention: timedelta expected_events_retention: timedelta - expected_own_events_retention: timedelta + expected_overrides: dict[str, timedelta] PREFECT_READ_TEST_CASES: list[PrefectReadTestCase] = [ @@ -79,18 +102,36 @@ class PrefectReadTestCase: task_history="30d", activity_log="365d", prefect_own_events="7d", + event_types=EVENT_TYPES, expected_vacuum_retention=timedelta(days=30), expected_events_retention=timedelta(days=365), - expected_own_events_retention=timedelta(days=7), + expected_overrides={ + "prefect.flow-run.Completed": timedelta(days=7), + "prefect.task-run.Running": timedelta(days=7), + }, ), PrefectReadTestCase( name="part_below_a_day", task_history="P1DT12H", activity_log="P400D", prefect_own_events="P1DT0.5S", + event_types=EVENT_TYPES, expected_vacuum_retention=timedelta(days=1, hours=12), expected_events_retention=timedelta(days=400), - expected_own_events_retention=timedelta(days=1, milliseconds=500), + expected_overrides={ + "prefect.flow-run.Completed": timedelta(days=1, milliseconds=500), + "prefect.task-run.Running": timedelta(days=1, milliseconds=500), + }, + ), + PrefectReadTestCase( + name="no_event_types_replaces_the_heartbeat_default", + task_history="30d", + activity_log="30d", + prefect_own_events="30d", + event_types=(), + expected_vacuum_retention=timedelta(days=30), + expected_events_retention=timedelta(days=30), + expected_overrides={}, ), ] @@ -103,7 +144,7 @@ def test_prefect_reads_the_derived_values(monkeypatch: pytest.MonkeyPatch, test_ activity_log=test_case.activity_log, prefect_own_events=test_case.prefect_own_events, ) - for name, value in build_prefect_retention_env(settings=settings, event_types=EVENT_TYPES).items(): + for name, value in build_prefect_retention_env(settings=settings, event_types=test_case.event_types).items(): monkeypatch.setenv(name, value) vacuum = ServerServicesDBVacuumSettings() @@ -111,18 +152,16 @@ def test_prefect_reads_the_derived_values(monkeypatch: pytest.MonkeyPatch, test_ assert vacuum.enabled_vacuum_types == {"events", "flow_runs", "orphans"} assert vacuum.retention_period == test_case.expected_vacuum_retention - assert vacuum.event_retention_overrides == { - "prefect.flow-run.Completed": test_case.expected_own_events_retention, - "prefect.task-run.Running": test_case.expected_own_events_retention, - } + assert vacuum.event_retention_overrides == test_case.expected_overrides assert events.retention_period == test_case.expected_events_retention def test_apply_sets_every_variable_missing_from_the_environment() -> None: + """Per-type overrides equal to the activity log retention shorten nothing, so they are written empty.""" settings = TaskManagerRetentionSettings(task_history="30d", activity_log="7d", prefect_own_events="7d") environ = {"PREFECT_API_DATABASE_CONNECTION_URL": "sqlite+aiosqlite:///prefect.db"} - warnings = apply_prefect_retention_env(environ=environ, settings=settings) + warnings = apply_prefect_retention_env(environ=environ, settings=settings, already_set=frozenset()) assert warnings == [] assert _with_parsed_overrides(environ) == { @@ -130,20 +169,50 @@ def test_apply_sets_every_variable_missing_from_the_environment() -> None: VACUUM_ENABLED: "events,flow_runs", VACUUM_RETENTION_PERIOD: "P30D", EVENTS_RETENTION_PERIOD: "P7D", - EVENT_RETENTION_OVERRIDES: dict.fromkeys(PREFECT_EVENT_TYPES, "P7D"), + EVENT_RETENTION_OVERRIDES: {}, } -def test_apply_warns_when_own_events_are_capped() -> None: - settings = TaskManagerRetentionSettings(task_history="30d", activity_log="7d", prefect_own_events="30d") +@dataclass +class WrittenOverridesTestCase: + name: str + activity_log: str + prefect_own_events: str + already_set: frozenset[str] + expected_overrides: dict[str, str] + + +WRITTEN_OVERRIDES_TEST_CASES: list[WrittenOverridesTestCase] = [ + WrittenOverridesTestCase( + name="own_events_shorter", + activity_log="7d", + prefect_own_events="3d", + already_set=frozenset(), + expected_overrides=dict.fromkeys(PREFECT_EVENT_TYPES, "P3D"), + ), + WrittenOverridesTestCase( + name="events_retention_preset", + activity_log="7d", + prefect_own_events="7d", + already_set=frozenset({EVENTS_RETENTION_PERIOD}), + expected_overrides=dict.fromkeys(PREFECT_EVENT_TYPES, "P7D"), + ), +] + + +@pytest.mark.parametrize("test_case", [pytest.param(tc, id=tc.name) for tc in WRITTEN_OVERRIDES_TEST_CASES]) +def test_overrides_are_written_only_when_they_can_shorten_the_event_retention( + test_case: WrittenOverridesTestCase, +) -> None: + """A preset event retention may be longer than prefect_own_events, so the overrides are kept for it.""" + settings = TaskManagerRetentionSettings( + task_history="30d", activity_log=test_case.activity_log, prefect_own_events=test_case.prefect_own_events + ) environ: dict[str, str] = {} - warnings = apply_prefect_retention_env(environ=environ, settings=settings) + apply_prefect_retention_env(environ=environ, settings=settings, already_set=test_case.already_set) - assert warnings == [ - "task_manager.retention.prefect_own_events P30D is longer than task_manager.retention.activity_log P7D, " - "Prefect's own events are kept for P7D" - ] + assert json.loads(environ[EVENT_RETENTION_OVERRIDES]) == test_case.expected_overrides @dataclass @@ -152,22 +221,17 @@ class PresetVariableTestCase: preset: str preset_value: str expected_value: object + expected_overrides: dict[str, str] expected_warning: str PRESET_VARIABLE_TEST_CASES: list[PresetVariableTestCase] = [ - PresetVariableTestCase( - name="vacuum_enabled", - preset=VACUUM_ENABLED, - preset_value="[]", - expected_value="[]", - expected_warning=f"{VACUUM_ENABLED} is set and overrides task_manager.retention.task_history", - ), PresetVariableTestCase( name="vacuum_retention_period", preset=VACUUM_RETENTION_PERIOD, preset_value="P90D", expected_value="P90D", + expected_overrides={}, expected_warning=f"{VACUUM_RETENTION_PERIOD} is set and overrides task_manager.retention.task_history", ), PresetVariableTestCase( @@ -175,6 +239,7 @@ class PresetVariableTestCase: preset=EVENTS_RETENTION_PERIOD, preset_value="P14D", expected_value="P14D", + expected_overrides=dict.fromkeys(PREFECT_EVENT_TYPES, "P7D"), expected_warning=f"{EVENTS_RETENTION_PERIOD} is set and overrides task_manager.retention.activity_log", ), PresetVariableTestCase( @@ -182,6 +247,7 @@ class PresetVariableTestCase: preset=EVENT_RETENTION_OVERRIDES, preset_value='{"prefect.flow-run.heartbeat": "P1D"}', expected_value={"prefect.flow-run.heartbeat": "P1D"}, + expected_overrides={"prefect.flow-run.heartbeat": "P1D"}, expected_warning=( f"{EVENT_RETENTION_OVERRIDES} is set and overrides task_manager.retention.prefect_own_events" ), @@ -194,19 +260,53 @@ def test_preset_variable_wins_with_a_warning(test_case: PresetVariableTestCase) settings = TaskManagerRetentionSettings(task_history="30d", activity_log="7d", prefect_own_events="7d") environ = {test_case.preset: test_case.preset_value} - warnings = apply_prefect_retention_env(environ=environ, settings=settings) + warnings = apply_prefect_retention_env( + environ=environ, settings=settings, already_set=frozenset({test_case.preset}) + ) assert warnings == [test_case.expected_warning] assert _with_parsed_overrides(environ) == { VACUUM_ENABLED: "events,flow_runs", VACUUM_RETENTION_PERIOD: "P30D", EVENTS_RETENTION_PERIOD: "P7D", - EVENT_RETENTION_OVERRIDES: dict.fromkeys(PREFECT_EVENT_TYPES, "P7D"), + EVENT_RETENTION_OVERRIDES: test_case.expected_overrides, test_case.preset: test_case.expected_value, } -LEGACY_EVENTS_RETENTION_PERIOD = "PREFECT_EVENTS_RETENTION_PERIOD" +def test_preset_vacuum_switch_is_kept_without_an_override_warning() -> None: + """The switch turns cleanups on or off, so the retentions are still derived from the settings.""" + settings = TaskManagerRetentionSettings(task_history="30d", activity_log="7d", prefect_own_events="7d") + environ = {VACUUM_ENABLED: "[]"} + + warnings = apply_prefect_retention_env(environ=environ, settings=settings, already_set=frozenset({VACUUM_ENABLED})) + + assert warnings == [] + assert _with_parsed_overrides(environ) == { + VACUUM_ENABLED: "[]", + VACUUM_RETENTION_PERIOD: "P30D", + EVENTS_RETENTION_PERIOD: "P7D", + EVENT_RETENTION_OVERRIDES: {}, + } + + +def test_setting_preset_outside_the_environment_is_kept_with_a_warning() -> None: + settings = TaskManagerRetentionSettings(task_history="30d", activity_log="7d", prefect_own_events="7d") + environ: dict[str, str] = {} + + warnings = apply_prefect_retention_env( + environ=environ, settings=settings, already_set=frozenset({EVENTS_RETENTION_PERIOD}) + ) + + assert warnings == [ + f"{EVENTS_RETENTION_PERIOD} is set in a Prefect settings file or profile and overrides " + "task_manager.retention.activity_log" + ] + assert _with_parsed_overrides(environ) == { + VACUUM_ENABLED: "events,flow_runs", + VACUUM_RETENTION_PERIOD: "P30D", + EVENT_RETENTION_OVERRIDES: dict.fromkeys(PREFECT_EVENT_TYPES, "P7D"), + } @dataclass @@ -249,7 +349,9 @@ def test_setting_preset_under_any_name_prefect_accepts_wins( settings = TaskManagerRetentionSettings(task_history="30d", activity_log="7d", prefect_own_events="7d") environ = dict(test_case.preset) - warnings = apply_prefect_retention_env(environ=environ, settings=settings) + warnings = apply_prefect_retention_env( + environ=environ, settings=settings, already_set=frozenset({EVENTS_RETENTION_PERIOD}) + ) assert warnings == [test_case.expected_warning] assert _with_parsed_overrides(environ) == { @@ -265,8 +367,55 @@ def test_setting_preset_under_any_name_prefect_accepts_wins( assert ServerEventsSettings().retention_period == test_case.expected_events_retention +@dataclass +class BlankVacuumSwitchTestCase: + name: str + environ: dict[str, str] + expected_environ: dict[str, str] + + +UNRELATED_VARIABLE = {"PREFECT_API_DATABASE_CONNECTION_URL": "sqlite+aiosqlite:///prefect.db"} + +BLANK_VACUUM_SWITCH_TEST_CASES: list[BlankVacuumSwitchTestCase] = [ + BlankVacuumSwitchTestCase( + name="empty", + environ={**UNRELATED_VARIABLE, VACUUM_ENABLED: ""}, + expected_environ=UNRELATED_VARIABLE, + ), + BlankVacuumSwitchTestCase( + name="blank", + environ={**UNRELATED_VARIABLE, VACUUM_ENABLED: " "}, + expected_environ=UNRELATED_VARIABLE, + ), + BlankVacuumSwitchTestCase( + name="lower_case_name_empty", + environ={**UNRELATED_VARIABLE, "prefect_server_services_db_vacuum_enabled": ""}, + expected_environ=UNRELATED_VARIABLE, + ), + BlankVacuumSwitchTestCase( + name="false_is_kept", + environ={**UNRELATED_VARIABLE, VACUUM_ENABLED: "false"}, + expected_environ={**UNRELATED_VARIABLE, VACUUM_ENABLED: "false"}, + ), + BlankVacuumSwitchTestCase( + name="empty_unrelated_variable_is_kept", + environ={**UNRELATED_VARIABLE, WEBSERVER_ONLY: ""}, + expected_environ={**UNRELATED_VARIABLE, WEBSERVER_ONLY: ""}, + ), +] + + +@pytest.mark.parametrize("test_case", [pytest.param(tc, id=tc.name) for tc in BLANK_VACUUM_SWITCH_TEST_CASES]) +def test_blank_vacuum_switch_is_removed_under_any_name_prefect_accepts(test_case: BlankVacuumSwitchTestCase) -> None: + environ = dict(test_case.environ) + + remove_blank_vacuum_switch(environ=environ) + + assert environ == test_case.expected_environ + + def test_environment_with_every_variable_preset_is_left_unchanged() -> None: - settings = TaskManagerRetentionSettings(task_history="30d", activity_log="7d", prefect_own_events="30d") + settings = TaskManagerRetentionSettings(task_history="30d", activity_log="7d", prefect_own_events="7d") environ = { VACUUM_ENABLED: "[]", VACUUM_RETENTION_PERIOD: "P90D", @@ -274,12 +423,9 @@ def test_environment_with_every_variable_preset_is_left_unchanged() -> None: EVENT_RETENTION_OVERRIDES: '{"prefect.flow-run.heartbeat": "P1D"}', } - warnings = apply_prefect_retention_env(environ=environ, settings=settings) + warnings = apply_prefect_retention_env(environ=environ, settings=settings, already_set=EVERY_RETENTION_VARIABLE) assert warnings == [ - "task_manager.retention.prefect_own_events P30D is longer than task_manager.retention.activity_log P7D, " - "Prefect's own events are kept for P7D", - f"{VACUUM_ENABLED} is set and overrides task_manager.retention.task_history", f"{VACUUM_RETENTION_PERIOD} is set and overrides task_manager.retention.task_history", f"{EVENTS_RETENTION_PERIOD} is set and overrides task_manager.retention.activity_log", f"{EVENT_RETENTION_OVERRIDES} is set and overrides task_manager.retention.prefect_own_events", @@ -292,24 +438,239 @@ def test_environment_with_every_variable_preset_is_left_unchanged() -> None: } -def test_overrides_in_effect_are_counted_per_retention() -> None: - """The per-type overrides are reported as a count per retention instead of the full list.""" - environ = { - EVENT_RETENTION_OVERRIDES: json.dumps( - { - "prefect.flow-run.Completed": "P3D", - "prefect.task-run.Running": "P3D", - "prefect.flow-run.heartbeat": "P1D", - } - ) - } +TASK_RUNS_NOT_DELETED_WARNING = ( + "PREFECT_SERVER_SERVICES_DB_VACUUM_ENABLED does not turn on flow_runs, so old task runs are not deleted and " + "task_manager.retention.task_history is not applied" +) +EVENTS_NOT_DELETED_WARNING = ( + "PREFECT_SERVER_SERVICES_DB_VACUUM_ENABLED does not turn on events, so old events are not deleted and " + "task_manager.retention.activity_log and task_manager.retention.prefect_own_events are not applied" +) - assert prefect_retention_env_in_effect(environ=environ) == { - EVENT_RETENTION_OVERRIDES: "P1D for 1 event types, P3D for 2 event types" - } + +@dataclass +class TurnedOffCleanupTestCase: + name: str + vacuum_enabled: str + expected_warnings: list[str] + + +TURNED_OFF_CLEANUP_TEST_CASES: list[TurnedOffCleanupTestCase] = [ + TurnedOffCleanupTestCase(name="true", vacuum_enabled="true", expected_warnings=[]), + TurnedOffCleanupTestCase(name="events_and_flow_runs", vacuum_enabled="events,flow_runs", expected_warnings=[]), + TurnedOffCleanupTestCase( + name="events_only", vacuum_enabled="events", expected_warnings=[TASK_RUNS_NOT_DELETED_WARNING] + ), + TurnedOffCleanupTestCase(name="false", vacuum_enabled="false", expected_warnings=[TASK_RUNS_NOT_DELETED_WARNING]), + TurnedOffCleanupTestCase( + name="flow_runs_only", vacuum_enabled="flow_runs", expected_warnings=[EVENTS_NOT_DELETED_WARNING] + ), + TurnedOffCleanupTestCase( + name="none", + vacuum_enabled="[]", + expected_warnings=[TASK_RUNS_NOT_DELETED_WARNING, EVENTS_NOT_DELETED_WARNING], + ), + TurnedOffCleanupTestCase( + name="unknown_type", + vacuum_enabled="Flow_Runs", + expected_warnings=[ + "PREFECT_SERVER_SERVICES_DB_VACUUM_ENABLED is not valid: Invalid vacuum type(s): ['Flow_Runs']. " + "Valid values are: ['events', 'flow_runs', 'orphans']" + ], + ), +] + + +@pytest.mark.parametrize("test_case", [pytest.param(tc, id=tc.name) for tc in TURNED_OFF_CLEANUP_TEST_CASES]) +def test_turned_off_cleanups_are_warned(monkeypatch: pytest.MonkeyPatch, test_case: TurnedOffCleanupTestCase) -> None: + """The warnings follow Prefect's own reading of the switch, where false still deletes old events.""" + monkeypatch.setenv(VACUUM_ENABLED, test_case.vacuum_enabled) + + assert turned_off_cleanup_warnings(vacuum=ServerServicesDBVacuumSettings()) == test_case.expected_warnings + + +@dataclass +class AlreadySetTestCase: + name: str + files: dict[str, str] + environ: dict[str, str] + expected: frozenset[str] + + +ALREADY_SET_TEST_CASES: list[AlreadySetTestCase] = [ + AlreadySetTestCase(name="nothing_set", files={}, environ={}, expected=frozenset()), + AlreadySetTestCase( + name="environment", + files={}, + environ={VACUUM_RETENTION_PERIOD: "P60D"}, + expected=frozenset({VACUUM_RETENTION_PERIOD}), + ), + AlreadySetTestCase( + name="legacy_environment_name", + files={}, + environ={LEGACY_EVENTS_RETENTION_PERIOD: "P14D"}, + expected=frozenset({EVENTS_RETENTION_PERIOD}), + ), + AlreadySetTestCase( + name="dotenv_file", + files={".env": f"{VACUUM_ENABLED}=events\n"}, + environ={}, + expected=frozenset({VACUUM_ENABLED}), + ), + AlreadySetTestCase( + name="prefect_toml", + files={"prefect.toml": '[server.events]\nretention_period = "P90D"\n'}, + environ={}, + expected=frozenset({EVENTS_RETENTION_PERIOD}), + ), + AlreadySetTestCase( + name="pyproject_toml", + files={"pyproject.toml": "[tool.prefect.server.services.db_vacuum]\nretention_period = 864000\n"}, + environ={}, + expected=frozenset({VACUUM_RETENTION_PERIOD}), + ), + AlreadySetTestCase( + name="profile", + files={ + "profiles.toml": ( + 'active = "infrahub"\n\n[profiles.infrahub]\n' + f'{EVENT_RETENTION_OVERRIDES} = \'{{"prefect.flow-run.heartbeat": "P1D"}}\'\n' + ) + }, + environ={}, + expected=frozenset({EVENT_RETENTION_OVERRIDES}), + ), +] + + +@pytest.mark.parametrize("test_case", [pytest.param(tc, id=tc.name) for tc in ALREADY_SET_TEST_CASES]) +def test_retention_settings_already_set_by_any_prefect_source( + monkeypatch: pytest.MonkeyPatch, prefect_settings_sources: Path, test_case: AlreadySetTestCase +) -> None: + for file_name, content in test_case.files.items(): + (prefect_settings_sources / file_name).write_text(content, encoding="utf-8") + for name, value in test_case.environ.items(): + monkeypatch.setenv(name, value) + + assert prefect_retention_already_set() == test_case.expected + + +@dataclass +class InEffectTestCase: + name: str + environ: dict[str, str] + expected: dict[str, str] + + +IN_EFFECT_TEST_CASES: list[InEffectTestCase] = [ + InEffectTestCase( + name="derived_values", + environ={ + VACUUM_ENABLED: "events,flow_runs", + VACUUM_RETENTION_PERIOD: "P30D", + EVENTS_RETENTION_PERIOD: "P7D", + EVENT_RETENTION_OVERRIDES: '{"prefect.flow-run.Completed": "P7D", "prefect.task-run.Running": "P7D"}', + }, + expected={ + VACUUM_ENABLED: "events,flow_runs", + VACUUM_RETENTION_PERIOD: "P30D", + EVENTS_RETENTION_PERIOD: "P7D", + EVENT_RETENTION_OVERRIDES: "P7D for 2 event types", + }, + ), + InEffectTestCase( + name="overrides_counted_per_retention", + environ={ + VACUUM_ENABLED: "true", + VACUUM_RETENTION_PERIOD: "P60D", + EVENTS_RETENTION_PERIOD: "P14D", + EVENT_RETENTION_OVERRIDES: ( + '{"prefect.flow-run.Completed": "P3D", "prefect.task-run.Running": "P3D", ' + '"prefect.flow-run.heartbeat": "P1D"}' + ), + }, + expected={ + VACUUM_ENABLED: "true", + VACUUM_RETENTION_PERIOD: "P60D", + EVENTS_RETENTION_PERIOD: "P14D", + EVENT_RETENTION_OVERRIDES: "P1D for 1 event types, P3D for 2 event types", + }, + ), + InEffectTestCase( + name="values_as_prefect_parses_them", + environ={ + VACUUM_ENABLED: "[]", + VACUUM_RETENTION_PERIOD: "7776000", + EVENTS_RETENTION_PERIOD: "PT36H", + EVENT_RETENTION_OVERRIDES: "{}", + }, + expected={ + VACUUM_ENABLED: "[]", + VACUUM_RETENTION_PERIOD: "P90D", + EVENTS_RETENTION_PERIOD: "P1DT43200S", + EVENT_RETENTION_OVERRIDES: "{}", + }, + ), + InEffectTestCase( + name="vacuum_switch_false", + environ={ + VACUUM_ENABLED: "false", + VACUUM_RETENTION_PERIOD: "P30D", + EVENTS_RETENTION_PERIOD: "P7D", + EVENT_RETENTION_OVERRIDES: "{}", + }, + expected={ + VACUUM_ENABLED: "false", + VACUUM_RETENTION_PERIOD: "P30D", + EVENTS_RETENTION_PERIOD: "P7D", + EVENT_RETENTION_OVERRIDES: "{}", + }, + ), +] + + +@pytest.mark.parametrize("test_case", [pytest.param(tc, id=tc.name) for tc in IN_EFFECT_TEST_CASES]) +@pytest.mark.usefixtures("prefect_settings_sources") +def test_retention_in_effect_is_read_from_prefects_parsed_settings( + monkeypatch: pytest.MonkeyPatch, test_case: InEffectTestCase +) -> None: + """The values are the ones Prefect parsed, and the per-type overrides are counted per retention.""" + for name, value in test_case.environ.items(): + monkeypatch.setenv(name, value) + + assert prefect_retention_in_effect(settings=Settings()) == test_case.expected + + +BACKGROUND_SERVICES_WARNING = ( + "Task manager retention: PREFECT__SERVER_WEBSERVER_ONLY is set, so this process does not run Prefect's " + "background services, which delete old task runs and events. The process that runs them, such as " + "`prefect server services start`, does not read task_manager.retention: unless it sets the same PREFECT_* " + "variables, it deletes no task runs and keeps events for Prefect's default of 7 days." +) -def test_overrides_in_effect_that_are_not_a_mapping_are_reported_as_set() -> None: - environ = {EVENT_RETENTION_OVERRIDES: "not json"} +@dataclass +class BackgroundServicesWarningTestCase: + name: str + environ: dict[str, str] + expected_warning: str | None + - assert prefect_retention_env_in_effect(environ=environ) == {EVENT_RETENTION_OVERRIDES: "not json"} +BACKGROUND_SERVICES_WARNING_TEST_CASES: list[BackgroundServicesWarningTestCase] = [ + BackgroundServicesWarningTestCase(name="unset", environ={}, expected_warning=None), + BackgroundServicesWarningTestCase(name="empty", environ={WEBSERVER_ONLY: ""}, expected_warning=None), + BackgroundServicesWarningTestCase( + name="true", environ={WEBSERVER_ONLY: "true"}, expected_warning=BACKGROUND_SERVICES_WARNING + ), + BackgroundServicesWarningTestCase( + name="false_is_still_set", environ={WEBSERVER_ONLY: "false"}, expected_warning=BACKGROUND_SERVICES_WARNING + ), +] + + +@pytest.mark.parametrize("test_case", [pytest.param(tc, id=tc.name) for tc in BACKGROUND_SERVICES_WARNING_TEST_CASES]) +def test_background_services_warning_follows_prefects_reading_of_the_variable( + test_case: BackgroundServicesWarningTestCase, +) -> None: + """Prefect runs only its web server for any non-empty value of the variable, so the warning follows the same rule.""" + assert background_services_warning(environ=test_case.environ) == test_case.expected_warning diff --git a/changelog/+task-manager-retention.added.md b/changelog/+task-manager-retention.added.md index 321e3b3f9af..1c8308cfb30 100644 --- a/changelog/+task-manager-retention.added.md +++ b/changelog/+task-manager-retention.added.md @@ -1,7 +1,9 @@ -Added three settings that control how long the task manager keeps its records: `task_manager.retention.task_history` (30 days by default), `task_manager.retention.activity_log` (7 days) and `task_manager.retention.prefect_own_events` (7 days). Set them in `infrahub.toml` or with the `INFRAHUB_TASK_MANAGER_RETENTION_TASK_HISTORY`, `INFRAHUB_TASK_MANAGER_RETENTION_ACTIVITY_LOG` and `INFRAHUB_TASK_MANAGER_RETENTION_PREFECT_OWN_EVENTS` environment variables, as a number of days (`30d`) or an ISO 8601 duration (`P30D`), from 1 to 36500 days. +Added three settings that control how long the task manager keeps its records: `task_manager.retention.task_history` (30 days by default), `task_manager.retention.activity_log` (7 days) and `task_manager.retention.prefect_own_events` (7 days). Set them in `infrahub.toml` or with the `INFRAHUB_TASK_MANAGER_RETENTION_TASK_HISTORY`, `INFRAHUB_TASK_MANAGER_RETENTION_ACTIVITY_LOG` and `INFRAHUB_TASK_MANAGER_RETENTION_PREFECT_OWN_EVENTS` environment variables, as a number of days (`30d`) or an ISO 8601 duration (`P30D`), from 1 to 36500 days. `prefect_own_events` must not be longer than `activity_log`, so set both when you lower `activity_log` below 7 days, or Infrahub refuses to start. -Until now, finished task runs were not deleted automatically by age. The task manager now deletes the finished task runs older than the task history retention every hour, together with their logs and artifacts, and the deletion cannot be undone, so set a longer `task_history` before upgrading if you need older runs. +Until now, finished task runs were not deleted by age unless `PREFECT_SERVER_SERVICES_DB_VACUUM_ENABLED` turned on Prefect's deletion, which kept them for 90 days. Now they are kept for `task_history` (30 days by default), unless `PREFECT_SERVER_SERVICES_DB_VACUUM_RETENTION_PERIOD` is set. Every hour the task manager deletes the finished task runs older than that, together with their logs and artifacts. The deletion cannot be undone, so set a longer `task_history` before upgrading if you need older runs. The activity log is still kept for 7 days by default. To keep it longer, raise `activity_log`, for example to `365d`, while the events the task manager records for its own use are still deleted after `prefect_own_events`. -A `PREFECT_*` variable already set in the task manager environment, such as `PREFECT_SERVER_EVENTS_RETENTION_PERIOD`, takes precedence over the matching setting, and the task manager logs a warning that names it. +When the task manager already has a value for `PREFECT_SERVER_SERVICES_DB_VACUUM_RETENTION_PERIOD`, `PREFECT_SERVER_EVENTS_RETENTION_PERIOD` or `PREFECT_SERVER_SERVICES_DB_VACUUM_EVENT_RETENTION_OVERRIDES`, whether from a `PREFECT_*` environment variable or from Prefect's `.env` file, `prefect.toml`, `pyproject.toml` or active profile, that value takes precedence over the matching setting, and the task manager logs a warning that names it. When `PREFECT_SERVER_SERVICES_DB_VACUUM_ENABLED` is already set in one of these places, it decides which of Prefect's cleanups run: if its value does not turn on `flow_runs` or `events`, the task manager logs a warning that names the retention it does not apply. An empty `PREFECT_SERVER_SERVICES_DB_VACUUM_ENABLED`, which a compose file passes when the host variable is not set, does not count as set: a value from Prefect's files, or from its active profile, applies when there is one, and otherwise the task manager deletes both old task runs and old events. An empty value of one of the other three variables still stops the task manager at startup with an error from Prefect, as it did before. + +When the task manager runs only its web server (`PREFECT__SERVER_WEBSERVER_ONLY` is set) and its background services run in a process of their own, such as one started with `prefect server services start`, that process does not read these settings, so it deletes no task runs unless its own environment sets the matching `PREFECT_*` variables; the task manager logs a warning at startup. diff --git a/dev/specs/infp-507-task-history-retention/contracts/configuration.md b/dev/specs/infp-507-task-history-retention/contracts/configuration.md index e789641234a..ec36f6b601f 100644 --- a/dev/specs/infp-507-task-history-retention/contracts/configuration.md +++ b/dev/specs/infp-507-task-history-retention/contracts/configuration.md @@ -16,8 +16,18 @@ prefect_own_events = "7d" | `prefect_own_events` | `INFRAHUB_TASK_MANAGER_RETENTION_PREFECT_OWN_EVENTS` | `7d` | - Values accept a number of days (`30d`) or an ISO 8601 duration (`P30D`); the configuration reference shows the day form. -- Every value must be at least 1 day and at most 36,500 days (100 years, a safety bound far below the roughly 739,000 days at which Prefect's cutoff date overflows), or the process refuses to start. A `prefect_own_events` longer than `activity_log` is capped to `activity_log` with a warning, because Prefect deletes every event older than the activity log retention anyway. +- Every value must be at least 1 day and at most 36,500 days (100 years, a safety bound far below the roughly 739,000 days at which Prefect's cutoff date overflows), or the process refuses to start. A `prefect_own_events` longer than `activity_log` also stops the process, because when the task manager sets the event retention, Prefect deletes every event older than `activity_log`, so a longer `prefect_own_events` would have no effect, and lowering `activity_log` below 7 days needs a `prefect_own_events` no longer than it. - Invalid values stop the task manager and the background-services command with: `Invalid task manager retention: `. -- A Prefect variable already set in the environment (`PREFECT_SERVER_SERVICES_DB_VACUUM_ENABLED`, `PREFECT_SERVER_SERVICES_DB_VACUUM_RETENTION_PERIOD`, `PREFECT_SERVER_EVENTS_RETENTION_PERIOD`, `PREFECT_SERVER_SERVICES_DB_VACUUM_EVENT_RETENTION_OVERRIDES`) takes precedence; the task manager logs `PREFECT_ is set and overrides task_manager.retention.` for each one. +- Of Infrahub's configuration, the standalone task manager reads only the `task_manager` section of `infrahub.toml`, the `INFRAHUB_TASK_MANAGER_*` variables and the two logging variables named below, so an invalid value in another section does not stop it. A configuration file it cannot read or parse still stops it. An invalid `INFRAHUB_LOG_LEVEL` (such as `verbose`) or `INFRAHUB_PRODUCTION` (such as `maybe`) also stops it, because the task manager reads these two variables to set up its logging. In distributed mode the task manager also loads the whole configuration, as it did before. +- A Prefect retention setting that the task manager already has (`PREFECT_SERVER_SERVICES_DB_VACUUM_RETENTION_PERIOD`, `PREFECT_SERVER_EVENTS_RETENTION_PERIOD`, `PREFECT_SERVER_SERVICES_DB_VACUUM_EVENT_RETENTION_OVERRIDES`) takes precedence, whether it comes from an environment variable that is not empty or from Prefect's `.env` file, `prefect.toml`, the `[tool.prefect]` table of `pyproject.toml` or active profile. The task manager writes no variable for it, because an environment variable would replace the value from a file, and logs one warning for each: + - Set in the environment: `PREFECT_ is set and overrides task_manager.retention.`, naming the variable found, which can be a legacy or lower-case name that Prefect accepts. + - Set in a file or the profile: `PREFECT_ is set in a Prefect settings file or profile and overrides task_manager.retention.`. +- A `PREFECT_SERVER_SERVICES_DB_VACUUM_ENABLED` already set, in the environment or in one of Prefect's settings files or the profile, is also kept, but with no such warning, because it turns Prefect's cleanups on or off and replaces no retention: `task_history` still sets how long task runs are kept unless `PREFECT_SERVER_SERVICES_DB_VACUUM_RETENTION_PERIOD` is set. After applying the variables, the task manager reads the switch as Prefect parses it and logs a warning for each cleanup it turns off: + - Without `flow_runs` (for example `false`, which Prefect reads as `events` only): `PREFECT_SERVER_SERVICES_DB_VACUUM_ENABLED does not turn on flow_runs, so old task runs are not deleted and task_manager.retention.task_history is not applied`. + - Without `events`: `PREFECT_SERVER_SERVICES_DB_VACUUM_ENABLED does not turn on events, so old events are not deleted and task_manager.retention.activity_log and task_manager.retention.prefect_own_events are not applied`. + - With a vacuum type Prefect does not accept: `PREFECT_SERVER_SERVICES_DB_VACUUM_ENABLED is not valid: `. The task manager does not stop on it; Prefect raises the same error when it schedules its background services. +- An empty or blank `PREFECT_SERVER_SERVICES_DB_VACUUM_ENABLED` environment variable, such as a compose file passes for an unset host variable, does not count as set, because Prefect reads it as `false`, which deletes no task runs. The task manager removes it, in any letter case, before it checks which settings are set. When Prefect's `.env` file, `prefect.toml`, the `[tool.prefect]` table of `pyproject.toml` or the active profile sets the switch, that value applies, with the warnings above for each cleanup it turns off. Otherwise the task manager writes `events,flow_runs` and logs no warning. An empty switch in Prefect's `.env` file is not removed: it counts as set, and Prefect reads it as `false`. +- An empty or blank value of one of the three retention variables, under the name Prefect reads (the main name, or a legacy name when the main name is not set), stops the task manager before it reads `task_manager.retention`, as it did before this feature. Prefect refuses the value when the task manager imports Prefect: the two retention periods fail with a `time_delta_parsing` validation error and the per-type overrides with a `SettingsError`. Every `infrahub` command stops on it in the same way, because the command line also imports Prefect when it starts. So neither the task manager nor a command can remove such a variable before Prefect reads it. +- At startup the task manager logs `Task manager retention: =` for each of the four Prefect settings, with the value Prefect parsed after the refresh, whatever its source. The per-type overrides are logged as a count of event types per retention, such as `P3D for event types`, or as `{}` when there is none. With the default retention there is none, because the task manager writes the overrides only when `prefect_own_events` is shorter than `activity_log` or the event retention is already set. - The values are read by the task manager, the background-services command, `infrahub tasks flush flow-runs` and `infrahub upgrade`. The server and workers do not need them, but setting them everywhere is harmless and is what the Helm chart and compose files do. - Names are final once the configuration reference is regenerated; the design doc marks them "not final". diff --git a/dev/specs/infp-507-task-history-retention/data-model.md b/dev/specs/infp-507-task-history-retention/data-model.md index 87440bcd030..6b58f3fd564 100644 --- a/dev/specs/infp-507-task-history-retention/data-model.md +++ b/dev/specs/infp-507-task-history-retention/data-model.md @@ -8,7 +8,7 @@ No graph schema, GraphQL schema or database schema changes. The feature adds one |---|---|---|---| | `task_history` | duration | 30 days | ≥ 1 day, ≤ 36,500 days | | `activity_log` | duration | 7 days | ≥ 1 day, ≤ 36,500 days | -| `prefect_own_events` | duration | 7 days | ≥ 1 day, ≤ 36,500 days; above `activity_log`, capped to it with a warning | +| `prefect_own_events` | duration | 7 days | ≥ 1 day, ≤ 36,500 days, ≤ `activity_log` | - Owned by the Infrahub configuration, as a section of the main settings with its own environment prefix (names final in the configuration contract). - Same values in every edition, with no edition limit. The 36,500-day (100-year) maximum is a technical safety bound, far below the roughly 739,000 days at which Prefect's cutoff date overflows. @@ -21,14 +21,14 @@ No graph schema, GraphQL schema or database schema changes. The feature adds one | `PREFECT_SERVER_SERVICES_DB_VACUUM_ENABLED` | `events,flow_runs` | | `PREFECT_SERVER_SERVICES_DB_VACUUM_RETENTION_PERIOD` | `task_history` | | `PREFECT_SERVER_EVENTS_RETENTION_PERIOD` | `activity_log` | -| `PREFECT_SERVER_SERVICES_DB_VACUUM_EVENT_RETENTION_OVERRIDES` | `{type: prefect_own_events for type in PREFECT_EVENT_TYPES}` | +| `PREFECT_SERVER_SERVICES_DB_VACUUM_EVENT_RETENTION_OVERRIDES` | `{}` when Infrahub sets the event retention and `prefect_own_events` equals `activity_log`; otherwise `{type: prefect_own_events for type in PREFECT_EVENT_TYPES}` | -A Prefect variable already set in the environment is left as is, with a warning. +A Prefect setting that one of Prefect's sources already sets is left as is: the task manager writes no variable for it, because an environment variable takes precedence over every other source. The sources are an environment variable that is not empty, a `.env` file, `prefect.toml`, the `[tool.prefect]` table of `pyproject.toml` and the active profile. A preset retention setting comes with a warning naming the setting it overrides. A preset `PREFECT_SERVER_SERVICES_DB_VACUUM_ENABLED` replaces no retention, so it gets no such warning; instead the task manager warns for each of `flow_runs` and `events` that the switch leaves off, naming the retention that is then not applied. An empty or blank vacuum switch in the environment is removed first, so it does not count as set; an empty switch in Prefect's `.env` file is not removed and counts as set. An empty or blank retention variable that Prefect reads (the main name, or a legacy name when the main name is not set) stops the task manager before it reads its configuration, because Prefect refuses the value when it is imported. The startup log shows each of the four values as Prefect parsed it after the refresh, whatever its source. ## Prefect event types list (new constant) -- A fixed list of the event types the pinned Prefect release emits for Infrahub's workload (flow-run and task-run state events, heartbeat, worker, automation, deployment, work pool, work queue, block events). -- Changes only with a Prefect upgrade; guarded by a unit test over Prefect's built-in states and an integration-docker test on a real workload. +- A fixed list of the event types the pinned Prefect release sends (flow-run and task-run state events for every state name Prefect defines or sets, heartbeat, worker, automation, deployment, work pool, work queue, block and workspace transfer events). +- Changes only with a Prefect upgrade; guarded by unit tests over the event names, name patterns, statuses, lifecycle record kinds and state names in the pinned Prefect's source, and by an integration-docker test on a real workload. - Infrahub event types (prefix `infrahub.`) are never in it. ## Existing Prefect records touched diff --git a/dev/specs/infp-507-task-history-retention/plan.md b/dev/specs/infp-507-task-history-retention/plan.md index 1bf6a646b8e..3d3ff9413aa 100644 --- a/dev/specs/infp-507-task-history-retention/plan.md +++ b/dev/specs/infp-507-task-history-retention/plan.md @@ -44,7 +44,7 @@ Research and code locations: [research.md](research.md). | I. Schema-Driven Integrity | Pass | No Infrahub schema or graph change. Generated docs regenerated, not edited. | | II. Branch-Safe by Default | Pass | No graph query changes. Branch filters resolve names through the branch registry; deleted branches through their deletion event. Branch-deletion purge of runs unchanged. | | III. Type Safety & Explicit Contracts | Pass | Pydantic settings section and request/response models for the new routes; contracts written before implementation (contracts/). | -| IV. Test Discipline | Pass (with private-test evidence) | Unit tests for settings and filter construction; component tests for the cleanup and filter equivalence on the Prefect harness; integration-docker guard and a unit test for the event-type list; Vitest for paging; an E2E test for Activities "load more" without the count; the private performance tests provide the evidence at production scale that CI cannot (backups with 25 to 100 GB of task history and a year of activity log). | +| IV. Test Discipline | Pass (with private-test evidence) | Unit tests for settings and filter construction; component tests for the cleanup and filter equivalence on the Prefect harness; an integration-docker guard and unit tests that check the event-type list against Prefect's source; Vitest for paging; an E2E test for Activities "load more" without the count; the private performance tests provide the evidence at production scale that CI cannot (backups with 25 to 100 GB of task history and a year of activity log). | | V. Query Performance | Pass | SQL built with SQLAlchemy Core, parameterized. Plans validated with EXPLAIN in the design-doc benchmark; regression covered by private performance tests. | | VI. Security & Input Boundaries | **Deviation (needs maintainer approval)** | The new cleanup route mutates without authentication. Decided by the tech owner on 2026-10-04; the constitution allows a deviation only with maintainer approval, so the PR description asks for it explicitly. The route takes only `rewrite`. See Complexity Tracking. Settings input is validated at start. | | VII. Simplicity | Pass, with one justified addition | The background job with status polling exists so that a dropped session or HTTP timeout during a long upgrade does not stop the cleanup. See Complexity Tracking. | @@ -74,9 +74,10 @@ dev/specs/infp-507-task-history-retention/ ```text backend/infrahub/ -├── config.py # new retention settings section (validation) +├── config.py # imports the retention settings into the configuration ├── prefect_server/ -│ ├── app.py # load config in all modes, apply derived Prefect settings, mount cleanup router +│ ├── app.py # read the task_manager section in all modes, apply derived Prefect settings, mount cleanup router +│ ├── retention_settings.py # new: retention settings (validation) and the reader of the task_manager section │ ├── retention.py # new: settings translation + PREFECT_EVENT_TYPES list │ ├── task_history.py # new: cleanup job (SQL deletes per day, rewrite on Postgres) + routes │ ├── database.py # newest-first time windows, optional count, plan per query @@ -94,7 +95,7 @@ backend/tests/ ├── unit/task_manager/event/ # filter construction ├── unit/prefect_server/ # settings translation and validation ├── component/task_manager/ # cleanup equivalence, filter equivalence, cleanup routes, count -├── unit/prefect_server/test_prefect_event_types.py # every built-in Prefect state in the list +├── unit/prefect_server/test_prefect_event_types.py # the list against the event and state names in Prefect's source └── integration_docker/ # Prefect event-type list guard on the full stack frontend/app/src/entities/events/ # page by `until`, drop `count`, dedupe by id @@ -152,12 +153,13 @@ Each of parts 1 to 3 merges only with its private-test evidence. The final run o | Upgrade step takes hours on large instances (Q1) | Per-day commits, progress output, re-runnable; operators can set a longer retention before upgrading. Release notes wait for the 100 GB figure. | | API clients that select `count` | They still get it, and still wait for it (about 10 minutes for level 0 events on a year of activity log), as the design doc states. | | SQL cleanup drifts from Prefect's rules on a Prefect upgrade | Cleanup equivalence component test in CI. | -| New Prefect event type not in the list | Integration-docker guard on the full stack (real workers, failure paths) plus a unit test over every built-in Prefect state; a missed type only costs disk. | +| New Prefect event type not in the list | Integration-docker guard on the full stack (real workers, failure paths) plus unit tests over the event names, name patterns, statuses, lifecycle record kinds and state names in the pinned Prefect's source; a missed type only costs disk. | | Filter results change on a Prefect upgrade | Filter equivalence component test in CI. | | Behaviour proven only on small CI data | Private tests on restored production-scale backups for each part, on both Postgres versions, attached as evidence. | | PR #33's retention override bypasses the new Infrahub setting | Switch it to `INFRAHUB_TASK_MANAGER_RETENTION_ACTIVITY_LOG` (T037). | | The Helm upgrade hook runs while the instance is serving | The chart passes `--no-task-history-cleanup`, and Helm users run the cleanup in a maintenance step after the rollout. Low likelihood anyway: the rewrite runs only when the deletes freed most of the tables, and Prefect's hourly cleanup has usually removed old runs already. The 404 path covers a chart without the flag on the first release. | -| Operators already set PREFECT_* variables by hand | Explicit values win, with a warning naming the hidden Infrahub setting. | +| Operators already set Prefect's retention settings by hand | Explicit values win, whether set by a `PREFECT_*` variable or by Prefect's `.env` file, `prefect.toml`, `pyproject.toml` or active profile. A preset retention setting comes with a warning naming the hidden Infrahub setting, and the startup log shows the values Prefect applies. A preset vacuum switch hides no setting, so the task manager instead warns for each cleanup it turns off, naming the retention that is not applied. | +| A compose or Helm file passes an unset host variable as an empty `PREFECT_*` variable | The task manager removes an empty or blank `PREFECT_SERVER_SERVICES_DB_VACUUM_ENABLED` before it checks which settings are set, because Prefect reads it as `false`, which deletes no task runs. A value from Prefect's settings files or profile then applies, with the usual warning for each cleanup it turns off, and otherwise the task manager writes `events,flow_runs`. An empty retention period or per-type override is not handled: Prefect refuses it when the task manager imports Prefect, before any Infrahub code runs, so the task manager stops at startup, as it did before this feature. | | Several task-manager replicas run two cleanups | Postgres advisory lock; the CLI retries on "running elsewhere" or an unknown job. | | The cleanup and Prefect's now-enabled vacuum run at the same time | Same order as Prefect (runs, then their children); Prefect skips locked runs and each side deletes children only for its own runs, so they cannot deadlock; the concurrency test (T009) guards it. | | A table rewrite waits forever on a lock, or makes queries queue behind it | 60 s lock timeout per table, up to 3 retries; skipped tables are reported. | diff --git a/dev/specs/infp-507-task-history-retention/quickstart.md b/dev/specs/infp-507-task-history-retention/quickstart.md index a87f7666f26..be13a314b34 100644 --- a/dev/specs/infp-507-task-history-retention/quickstart.md +++ b/dev/specs/infp-507-task-history-retention/quickstart.md @@ -31,8 +31,8 @@ Speed with a year of activity log is validated by the private performance tests ## Part 3: activity log retention 1. Set `INFRAHUB_TASK_MANAGER_RETENTION_ACTIVITY_LOG=365d` and restart. Expect `PREFECT_SERVER_EVENTS_RETENTION_PERIOD` of 365 days and the per-type overrides at 7 days in the startup log. -2. Set `INFRAHUB_TASK_MANAGER_RETENTION_PREFECT_OWN_EVENTS=30d` with the activity log at 7 days. Expect a warning and the per-type overrides at 7 days. -3. Run the event-type list tests: the unit test (every built-in Prefect state listed) and the integration-docker guard (a workload on the full stack stores no Prefect event type missing from the list). +2. Set `INFRAHUB_TASK_MANAGER_RETENTION_PREFECT_OWN_EVENTS=30d` with the activity log at 7 days. Expect the task manager to refuse to start with an error that names `prefect_own_events` and `activity_log`. +3. Run the event-type list tests: the unit tests (every event name, status, lifecycle record kind and state name in Prefect's source listed, and no unknown name pattern) and the integration-docker guard (a workload on the full stack stores no Prefect event type missing from the list). ## Part 4: documentation diff --git a/dev/specs/infp-507-task-history-retention/research.md b/dev/specs/infp-507-task-history-retention/research.md index 86f326caeb5..dddcfaecc2d 100644 --- a/dev/specs/infp-507-task-history-retention/research.md +++ b/dev/specs/infp-507-task-history-retention/research.md @@ -11,23 +11,34 @@ The [Notion design doc](https://app.notion.com/p/opsmill/Task-history-and-Activi ## R2. How Infrahub sets Prefect settings in the task manager (D5) -- **Decision**: In `infrahub.prefect_server.app::create_infrahub_prefect`, load the Infrahub configuration, validate the retention settings, and write the derived `PREFECT_*` variables into `os.environ` before `prefect.server.api.server.create_app` is called. A new `infrahub tasks background-services` command does the same translation and then runs Prefect's services (`prefect server services start` equivalent) in the same process. +- **Decision**: In `infrahub.prefect_server.app::create_infrahub_prefect`, read the `task_manager` section of the Infrahub configuration, validate the retention settings, and write the derived `PREFECT_*` variables into `os.environ` before `prefect.server.api.server.create_app` is called. A new `infrahub tasks background-services` command does the same translation and then runs Prefect's services (`prefect server services start` equivalent) in the same process. - **Confirmed**: `prefect.settings.context::get_current_settings` returns the settings of `prefect.context.GLOBAL_SETTINGS_CONTEXT` whenever no settings context is active, and that context is built from the environment once, when `prefect.context` is imported. Values written to `os.environ` afterwards are picked up only after `prefect.context.refresh_global_settings_context()`; once refreshed, the vacuum services see them, since they read `get_current_settings()` on every run. The background-services command needs the same refresh. -- **Gap found**: `create_infrahub_prefect` loads the Infrahub configuration only in distributed mode (when block registration and migrations on start are off). The translation must load the configuration in both modes. -- **Explicit `PREFECT_*` values win**: if an operator already sets one of the four Prefect variables, Infrahub leaves it as is and logs one warning per variable, naming the Prefect variable and the Infrahub setting it hides, so existing deployments that tuned Prefect directly keep their behaviour. The three Infrahub settings are the documented interface, and the configuration reference states the precedence. +- **Gap found**: `create_infrahub_prefect` loads the Infrahub configuration only in distributed mode (when block registration and migrations on start are off). The translation reads the `task_manager` section of the configuration in both modes; only distributed mode loads and validates the whole configuration, so a setting the task manager does not use cannot stop it. +- **Explicit Prefect settings win**: if an operator already sets one of the four Prefect settings, through a `PREFECT_*` variable that is not empty or through Prefect's `.env` file, `prefect.toml`, the `[tool.prefect]` table of `pyproject.toml` or active profile, Infrahub writes no variable for it, so existing deployments that tuned Prefect directly keep the values they set. Prefect gives an environment variable precedence over all of those files, so a variable written by Infrahub would replace a value from a file without a warning. Infrahub finds the settings already set through `model_fields_set` of Prefect's own settings models (`ServerServicesDBVacuumSettings`, `ServerEventsSettings`), which lists a field only when one of Prefect's sources supplies it; it reads them after removing a blank vacuum switch, which Prefect reads as `false`, and before writing any. Only the switch can be blank and read by Prefect at that point: Prefect refuses an empty retention period or per-type override when it is imported, so the task manager and every `infrahub` command stop before any Infrahub code runs, as they did before. For each of the three retention settings it logs one warning, naming the Prefect variable and the Infrahub setting it hides, and saying when a Prefect settings file or profile sets it. The vacuum switch (`PREFECT_SERVER_SERVICES_DB_VACUUM_ENABLED`) hides no setting: it only turns Prefect's cleanups on or off, and `task_history` still sets the task-run retention unless `PREFECT_SERVER_SERVICES_DB_VACUUM_RETENTION_PERIOD` is set, so a deployment that set only the switch, and kept runs for Prefect's default of 90 days, now keeps them for `task_history`. After the refresh, Infrahub reads the switch as Prefect parses it (`ServerServicesDBVacuumSettings.enabled_vacuum_types`, which Prefect checks only when it is read) and logs a warning for each of `flow_runs` and `events` that it leaves off, naming the retention that is then not applied. The startup log reads the four values from Prefect's parsed settings after the refresh, so it shows what Prefect applies whatever the source. The three Infrahub settings are the documented interface, and the configuration reference states the precedence. - **Alternatives considered**: Prefect's `temporary_settings` context (scoped to a context manager, not the server lifetime); waiting for INFP-703 (rejected in D5). ## R3. Where the background services run (D5, Helm row) - **Confirmed**: Compose and default Helm run Prefect's background services inside the task-manager process. The test compose file `python_testcontainers/infrahub_testcontainers/docker-compose.test.yml` and `docker-compose-cluster.test.yml` define `task-manager-background-svc` with `command: prefect server services start`, and the Helm chart offers the same split as an option. - **Decision**: Replace that command with `infrahub tasks background-services` in both test compose files, and change the Helm chart in the infrahub-helm repository (separate PR). +- **Gap found**: a task manager started with `PREFECT__SERVER_WEBSERVER_ONLY` (set by `prefect server start --no-services` and by the scale-out setup of `python_testcontainers/infrahub_testcontainers/container.py`) runs none of Prefect's cleanup services, so the retention it applies to its own process deletes no task runs. With a shared Redis docket, its docket worker still runs some of the event-cleanup tasks that the services process schedules (`prefect/server/api/background_workers.py::task_functions`), and each task reads the settings of the process that runs it, so both processes need the same values. Until the background-services command exists, the task manager logs a warning at startup when the variable is set to any non-empty value, which is how Prefect reads it (`prefect/server/api/server.py::create_app`). ## R4. Event retention per type (D3) -- **Decision**: Set `PREFECT_SERVER_EVENTS_RETENTION_PERIOD` to the activity log retention and `PREFECT_SERVER_SERVICES_DB_VACUUM_EVENT_RETENTION_OVERRIDES` to `{event_type: own_events_retention}` for every type in Infrahub's list of Prefect event types. -- **Confirmed** (`db_vacuum.py::vacuum_events_with_retention_overrides`): overrides match `Event.event == event_type` exactly and the effective retention is `min(type_retention, global_retention)`, which is why an own-event retention longer than the activity log is capped to it with a warning (D5). Related resources are deleted before the events, so none are left behind. -- **List source**: a module-level constant in the backend, built from the pinned Prefect release (run and task-run state events, heartbeat, worker, automation, deployment, work pool, work queue, block events). The benchmark found 26, 14 and 33 types across three datasets. -- **Guard**: an integration-docker test runs a workload of Infrahub tasks on the full stack (real workers, Postgres), including the failure paths it can reach (failed and cancelled flows, a worker restart), and fails when a stored event type that does not start with `infrahub.` is missing from the list. The functional tier is not enough: it runs Prefect's test harness on SQLite with non-essential services off and no real worker, so worker and work-pool events never appear. A unit test also checks that `prefect.flow-run.` and `prefect.task-run.` are listed for every built-in state of the pinned Prefect, so a Prefect upgrade that adds a state fails CI whatever the workload reaches. +- **Decision**: Set `PREFECT_SERVER_EVENTS_RETENTION_PERIOD` to the activity log retention and `PREFECT_SERVER_SERVICES_DB_VACUUM_EVENT_RETENTION_OVERRIDES` to `{event_type: prefect_own_events}` for every type in Infrahub's list of Prefect event types. When Infrahub sets the event retention and `prefect_own_events` equals `activity_log`, the overrides are set to `{}` instead: Prefect's general event cleanup (`vacuum_old_events`) deletes the same events, and each listed type costs at least two delete statements every hour (one for its related resources, one for the events). The variable is set to `{}` rather than left unset, because Prefect's default is `{"prefect.flow-run.heartbeat": 7 days}`, which would delete heartbeats earlier than both retentions when they are longer than 7 days. When the event retention is preset, Infrahub does not know whether the overrides shorten it, so it writes the full list. +- **Confirmed** (`db_vacuum.py::vacuum_events_with_retention_overrides`): overrides match `Event.event == event_type` exactly and the effective retention is `min(type_retention, global_retention)`, which is why an own-event retention longer than the activity log has no effect and is refused at startup. Because Prefect applies this cap with the event retention it actually uses, preset or derived, the overrides carry `prefect_own_events` as configured. Related resources are deleted before the events, so none are left behind. +- **List source**: a module-level constant in the backend, built from the pinned Prefect release (run and task-run state events, heartbeat, worker, automation, deployment, work pool, work queue, block and workspace transfer events). The benchmark found 26, 14 and 33 types across three datasets. +- **Guard**: an integration-docker test runs a workload of Infrahub tasks on the full stack (real workers, Postgres), including the failure paths it can reach (failed and cancelled flows, a worker restart), and fails when a stored event type that does not start with `infrahub.` is missing from the list. The functional tier is not enough: it runs Prefect's test harness on SQLite with non-essential services off and no real worker, so worker and work-pool events never appear. Unit tests also read the installed Prefect package's source and fail when it sends a `prefect.*` event name that is not listed, builds event names from a pattern other than the known ones, or has a deployment, work pool or work queue status, an automation trigger state or a lifecycle record kind whose events are not listed. They also check that `prefect.flow-run.` and `prefect.task-run.` are listed for every state name Prefect defines or sets in its source. A Prefect upgrade that adds one of these fails CI whatever the workload reaches. Names that exist only at runtime are not in the source, so these tests do not cover them: + - custom state names; + - load events of block types that Infrahub does not load; + - `prefect.*` events that API clients post. + + The tests also do not read some values that Prefect's source puts into the known patterns, so a new value is caught only by the integration-docker guard, and only when the workload reaches it: + - the actions of lifecycle events, which the tests take to be `created`, `updated` and `deleted`; + - the phases in `prefect.concurrency-limit.` and `prefect.concurrency-limit.v1.`; + - the statuses in `prefect.asset.materialization.`; + - an event name passed through a helper whose parameter is not named `event` or `event_type`. +- **Alternatives considered**: deleting every event whose type does not start with `infrahub.` once it is older than `prefect_own_events`, which needs no list. Prefect's settings cannot do this: the overrides match exact names and are capped by the event retention, and the general cleanup deletes every event older than the event retention. It would need a periodic delete that Infrahub runs in the process that runs Prefect's services, which goes against D1 and D3: both leave the deletes to Prefect's own cleanups. ## R5. The command for old runs and its endpoint (D9) @@ -59,7 +70,7 @@ The [Notion design doc](https://app.notion.com/p/opsmill/Task-history-and-Activi ## R8. Tests - **Confirmed**: `backend/tests/component/conftest.py::prefect_test_fixture` starts `create_infrahub_prefect` through `prefect_test_harness` (SQLite). `backend/tests/helpers/constants.py` turns Prefect's vacuum off for tests (`PREFECT_SERVER_SERVICES_DB_VACUUM_ENABLED: "[]"`). Existing unit tests live under `backend/tests/unit/task_manager/event/` and `backend/tests/unit/task_manager/flow_run/` (including `test_retention.py`). -- **Decision**: the cleanup and filter guards are component tests on the existing harness: the cleanup equivalence test seeds runs, copies the database state, runs `vacuum_old_flow_runs` and the new cleanup, and compares remaining run, log and artifact IDs; the filter equivalence test seeds events and compares each new filter with the label filter. The event-type list guard is an integration-docker test plus a unit test over Prefect's built-in states (see R4). A concurrency test runs Prefect's vacuum and the new cleanup at the same time on the same old runs and checks both finish with the same end state. The cleanup equivalence and the rewrite also run against Postgres in the integration-docker tier, which already boots the task manager on Postgres. Query speed stays with the private performance tests listed in the design doc. +- **Decision**: the cleanup and filter guards are component tests on the existing harness: the cleanup equivalence test seeds runs, copies the database state, runs `vacuum_old_flow_runs` and the new cleanup, and compares remaining run, log and artifact IDs; the filter equivalence test seeds events and compares each new filter with the label filter. The event-type list guard is an integration-docker test plus unit tests over the pinned Prefect's source (see R4). A concurrency test runs Prefect's vacuum and the new cleanup at the same time on the same old runs and checks both finish with the same end state. The cleanup equivalence and the rewrite also run against Postgres in the integration-docker tier, which already boots the task manager on Postgres. Query speed stays with the private performance tests listed in the design doc. ## R9. Documentation surfaces diff --git a/dev/specs/infp-507-task-history-retention/spec.md b/dev/specs/infp-507-task-history-retention/spec.md index 82fd37fbf7f..7a530b61b47 100644 --- a/dev/specs/infp-507-task-history-retention/spec.md +++ b/dev/specs/infp-507-task-history-retention/spec.md @@ -93,7 +93,7 @@ An operator sets how long task history, the activity log and the task manager's 1. **Given** the three retention settings, **When** the task manager starts, **Then** its cleanups use them, without the operator setting any task-manager-specific variable. 2. **Given** the task manager's background services in their own deployment (a Helm option), **When** they start, **Then** they use the same retention values, because Infrahub's command starts them. 3. **Given** a retention shorter than 1 day, **When** the task manager starts, **Then** it refuses to start and names the setting. -4. **Given** an own-event retention longer than the activity log retention, **When** the task manager starts, **Then** it logs a warning and uses the activity log retention for Prefect's own events, because every event older than that is deleted anyway. +4. **Given** an own-event retention longer than the activity log retention, **When** the task manager starts, **Then** it refuses to start and names both settings, because when the task manager sets the event retention, every event older than the activity log retention is deleted, so a longer own-event retention would have no effect. --- @@ -127,7 +127,7 @@ An operator sets how long task history, the activity log and the task manager's - **FR-001**: System MUST provide three retention settings: task history (default 30 days), activity log (default 7 days) and the task manager's own events (default 7 days), settable in the Infrahub configuration, Helm values and compose environment. - **FR-002**: System MUST apply the three settings to the task manager's cleanups at start, without the operator setting task-manager-specific variables. - **FR-003**: System MUST apply the same settings when the task manager's background services run in their own deployment, by starting that deployment through an Infrahub command. -- **FR-004**: System MUST refuse to start the task manager, naming the setting, when a retention is shorter than 1 day or longer than 36,500 days; when the own-event retention is longer than the activity log retention, it MUST log a warning and use the activity log retention for Prefect's own events. +- **FR-004**: System MUST refuse to start the task manager, naming the setting, when a retention is shorter than 1 day or longer than 36,500 days; it MUST also refuse to start, naming both settings, when the own-event retention is longer than the activity log retention. - **FR-005**: System MUST accept the same retention settings in every edition, with no edition-specific limit; values above 36,500 days (100 years) are refused, a safety bound far below the point where Prefect's cutoff date overflows. **Task history** @@ -208,7 +208,7 @@ Timings are indicative: they come from local benchmarks, not a production contra - Q: How do retention settings reach a separate background-services deployment? → A: A thin Infrahub command starts it after applying the same settings. - Q: Are the SQL cleanup and the new filters checked on task-manager upgrades? → A: Yes, by CI equivalence checks. - Q: Fallback if product refuses to remove the total count? → A: None; argue for removal. (Later found moot: the page never asked for the count; the design doc dropped D7 and Q2.) -- Q: Values accepted by the settings? → A: Durations of at least 1 day, refuse to start otherwise; an own-event retention longer than the activity log is capped to it with a warning (design doc D5). +- Q: Values accepted by the settings? → A: Durations of at least 1 day, refuse to start otherwise; an own-event retention longer than the activity log is refused at startup, instead of being capped to it with a warning as design doc D5 decided. - Q: Activity log default? → A: Stays 7 days; a year or more is a capability, not the default. - Q: Helm upgrade? → A: The pre-upgrade hook does not run the cleanup; a maintenance step after the rollout does. - Q: Load more by time and time windows? → A: Windows count back from the oldest event shown. diff --git a/dev/specs/infp-507-task-history-retention/tasks.md b/dev/specs/infp-507-task-history-retention/tasks.md index d596aea6718..dbdf1450337 100644 --- a/dev/specs/infp-507-task-history-retention/tasks.md +++ b/dev/specs/infp-507-task-history-retention/tasks.md @@ -28,13 +28,13 @@ **Purpose**: The retention settings and their translation into Prefect settings, used by US1, US3 and US4. -- [X] T003 Add `TaskManagerRetentionSettings` (fields `task_history` default 30 days, `activity_log` default 7 days, `prefect_own_events` default 7 days; values accept `30d` or ISO 8601 durations; validators: each ≥ 1 day, error message `Invalid task manager retention: `; a `prefect_own_events` longer than `activity_log` is not an error) and a `TaskManagerSettings` holding it under `retention`, registered on `Settings` as `task_manager`, env prefix `INFRAHUB_TASK_MANAGER_RETENTION_`, in `backend/infrahub/config.py` (contract: [contracts/configuration.md](contracts/configuration.md)) -- [X] T004 [P] Unit tests for T003: defaults, `30d` and `P30D` parsing, refusal below 1 day, own events longer than activity log accepted, env variable names, in `backend/tests/unit/config/test_task_manager_retention_settings.py` -- [X] T005 Create `backend/infrahub/prefect_server/retention.py` with `PREFECT_EVENT_TYPES: tuple[str, ...] = ()` (filled in US3) and `build_prefect_retention_env(settings: TaskManagerRetentionSettings, event_types: tuple[str, ...]) -> dict[str, str]` returning `PREFECT_SERVER_SERVICES_DB_VACUUM_ENABLED=events,flow_runs`, `PREFECT_SERVER_SERVICES_DB_VACUUM_RETENTION_PERIOD`, `PREFECT_SERVER_EVENTS_RETENTION_PERIOD` (ISO 8601 durations in days, such as `P30D`) and `PREFECT_SERVER_SERVICES_DB_VACUUM_EVENT_RETENTION_OVERRIDES` (JSON, one entry per given type, using `min(prefect_own_events, activity_log)`), and `apply_prefect_retention_env(environ: MutableMapping[str, str], settings) -> list[str]` that builds them from `PREFECT_EVENT_TYPES`, sets only variables not already present and returns a warning when own events are capped plus a warning `PREFECT_ is set and overrides task_manager.retention.` per variable left as is (data-model: Derived Prefect settings) -- [X] T006 [P] Unit tests for T005 (derived values, JSON shape, capping of own events with its warning, precedence and warnings) in `backend/tests/unit/prefect_server/test_retention.py` -- [X] T007 In `backend/infrahub/prefect_server/app.py::create_infrahub_prefect`, load the Infrahub configuration in every mode (today only in distributed mode), then call `apply_prefect_retention_env(os.environ, config.SETTINGS.task_manager.retention)`, log each warning and refresh Prefect's global settings (`prefect.context.refresh_global_settings_context`, since Prefect reads the environment once at import), before `create_app()`; an invalid retention stops the process with the validation error. The steps live in `apply_infrahub_settings_to_prefect()`; the stock Prefect container of `backend/tests/helpers/utils.py` (no `infrahub` package) starts `create_prefect_app_with_infrahub_routes` instead +- [X] T003 Add `TaskManagerRetentionSettings` (fields `task_history` default 30 days, `activity_log` default 7 days, `prefect_own_events` default 7 days; values accept `30d` or ISO 8601 durations; validators: each ≥ 1 day, error message `Invalid task manager retention: `; a model validator refuses a `prefect_own_events` longer than `activity_log`) and a `TaskManagerSettings` holding it under `retention`, registered on `Settings` as `task_manager`, env prefix `INFRAHUB_TASK_MANAGER_RETENTION_`, in `backend/infrahub/prefect_server/retention_settings.py`, which imports nothing else from Infrahub; `backend/infrahub/config.py` imports them from there (contract: [contracts/configuration.md](contracts/configuration.md)) +- [X] T004 [P] Unit tests for T003: defaults, `30d` and `P30D` parsing, refusal below 1 day, own events longer than activity log refused, env variable names, in `backend/tests/unit/config/test_task_manager_retention_settings.py` +- [X] T005 Create `backend/infrahub/prefect_server/retention.py` with `PREFECT_EVENT_TYPES: tuple[str, ...] = ()` (filled in US3) and `build_prefect_retention_env(settings: TaskManagerRetentionSettings, event_types: tuple[str, ...]) -> dict[str, str]` returning `PREFECT_SERVER_SERVICES_DB_VACUUM_ENABLED=events,flow_runs`, `PREFECT_SERVER_SERVICES_DB_VACUUM_RETENTION_PERIOD`, `PREFECT_SERVER_EVENTS_RETENTION_PERIOD` (ISO 8601 durations in days, such as `P30D`) and `PREFECT_SERVER_SERVICES_DB_VACUUM_EVENT_RETENTION_OVERRIDES` (JSON, one entry per given type, using `prefect_own_events`, or `{}` for no type), `prefect_retention_already_set() -> frozenset[str]` that returns the variables whose Prefect setting one of Prefect's sources already sets (the environment, a `.env` file, `prefect.toml`, the `[tool.prefect]` table of `pyproject.toml` or the active profile, read from `model_fields_set` of Prefect's settings models), and `apply_prefect_retention_env(environ: MutableMapping[str, str], settings, already_set: frozenset[str]) -> list[str]` that builds them from `PREFECT_EVENT_TYPES` when `prefect_own_events` is shorter than `activity_log` or the event retention is already set (otherwise from no type, which writes `{}`), sets only the variables not in `already_set` and returns a warning per retention setting left as is: `PREFECT_ is set and overrides task_manager.retention.` when the environment sets it, or `PREFECT_ is set in a Prefect settings file or profile and overrides task_manager.retention.` otherwise (a preset `PREFECT_SERVER_SERVICES_DB_VACUUM_ENABLED` is kept with no warning, since it replaces no retention), `prefect_retention_in_effect(settings: Settings) -> dict[str, str]` that returns the four values from Prefect's parsed settings, with the per-type overrides counted per retention, and `turned_off_cleanup_warnings(vacuum: ServerServicesDBVacuumSettings) -> list[str]` that returns a warning for each of `flow_runs` and `events` that Prefect's parsed switch leaves off, naming the retention not applied, or one warning when Prefect does not accept the switch (data-model: Derived Prefect settings) +- [X] T006 [P] Unit tests for T005 (derived values, JSON shape, precedence and warnings) in `backend/tests/unit/prefect_server/test_retention.py` +- [X] T007 In `backend/infrahub/prefect_server/app.py::create_infrahub_prefect`, read the `task_manager` section of the configuration in every mode (the `[task_manager]` table of the configuration file and the `INFRAHUB_TASK_MANAGER_*` variables, without validating the other sections), then remove the vacuum switch when it is present but empty or blank (`remove_blank_vacuum_switch(os.environ)`), read `prefect_retention_already_set()` before writing any variable, call `apply_prefect_retention_env(os.environ, , )`, log each warning and refresh Prefect's global settings (`prefect.context.refresh_global_settings_context`, since Prefect reads the environment once at import), then log `turned_off_cleanup_warnings(get_current_settings().server.services.db_vacuum)` and one `Task manager retention: =` line per value of `prefect_retention_in_effect(get_current_settings())`, read from Prefect's parsed settings, before `create_app()`; an invalid retention, or a configuration file that cannot be read or parsed, stops the process. Distributed mode still loads the whole configuration first, as before. The steps live in `apply_infrahub_settings_to_prefect()`; the stock Prefect container of `backend/tests/helpers/utils.py` (no `infrahub` package) starts `create_prefect_app_with_infrahub_routes` instead. When `PREFECT__SERVER_WEBSERVER_ONLY` is set, log a warning that Prefect's background services run in another process, which does not read `task_manager.retention` -**Checkpoint**: The task manager starts with the derived Prefect settings; the component harness still turns the vacuum off through `backend/tests/helpers/constants.py`, which now takes precedence by design. +**Checkpoint**: The task manager starts with the derived Prefect settings; the component harness still turns the vacuum off through `backend/tests/helpers/constants.py`, which now takes precedence by design, and the task manager logs a warning for each of the two cleanups it turns off. --- @@ -122,13 +122,13 @@ Run through the `test-dataset` workflow (`test_filter=...`) on restored backups ### Tests for User Story 3 - [ ] T040 [P] [US3] Integration-docker guard test on the full stack (real workers, Postgres): run a workload of Infrahub tasks that also reaches the failure paths it can (a failed flow, a cancelled flow, a worker restart), list stored event names not starting with `infrahub.`, assert none is missing from `PREFECT_EVENT_TYPES`, in `backend/tests/integration_docker/test_prefect_event_types.py` -- [X] T041 [P] [US3] Unit test independent of any workload: for every built-in state name of the pinned Prefect (read from Prefect's state definitions), assert `prefect.flow-run.` and `prefect.task-run.` are in `PREFECT_EVENT_TYPES`, so a Prefect upgrade that adds a state fails CI, in `backend/tests/unit/prefect_server/test_prefect_event_types.py` +- [X] T041 [P] [US3] Unit tests independent of any workload, reading the installed Prefect package's source: every `prefect.*` event name it sends as a literal is in `PREFECT_EVENT_TYPES`; it builds event names only from the known patterns; every deployment, work pool and work queue status, automation trigger state and lifecycle record kind has its events listed; and every state name it defines or sets (read from Prefect's state definitions and from the names its source sets on states) has `prefect.flow-run.` and `prefect.task-run.` listed. A Prefect upgrade that adds one fails CI. In `backend/tests/unit/prefect_server/test_prefect_event_types.py` - [X] T042 [P] [US3] Component test: with activity log 365 days and own events 7 days applied, run Prefect's `vacuum_events_with_retention_overrides` and `vacuum_old_events`; Infrahub events older than 7 days are kept, listed Prefect events older than 7 days are deleted, an unlisted Prefect type is kept, no `event_resources` row is left without its event, in `backend/tests/component/task_manager/test_event_retention.py` ### Implementation for User Story 3 -- [X] T043 [US3] Fill `PREFECT_EVENT_TYPES` from the pinned Prefect 3.8.6 (flow-run and task-run state events for every built-in state name, heartbeat, worker, automation, deployment, work pool, work queue and block events) and the benchmark datasets, with a one-line comment naming the Prefect version, in `backend/infrahub/prefect_server/retention.py` -- [ ] T044 [US3] Use the activity log retention as the widest Activities time window by default (task manager reads `config.SETTINGS.task_manager.retention.activity_log` when the request does not set it) in `backend/infrahub/prefect_server/models.py` and `backend/infrahub/prefect_server/database.py` +- [X] T043 [US3] Fill `PREFECT_EVENT_TYPES` from the pinned Prefect 3.8.6 (flow-run and task-run state events for every state name Prefect defines or sets, heartbeat, worker, automation, deployment, work pool, work queue, block and workspace transfer events) and the benchmark datasets, with a one-line comment naming the Prefect version, in `backend/infrahub/prefect_server/retention.py` +- [ ] T044 [US3] Use the activity log retention as the widest Activities time window by default (task manager reads the `activity_log` retention from the `task_manager` section of the configuration when the request does not set it; the standalone task manager does not load the whole configuration, so `config.SETTINGS` is not available there) in `backend/infrahub/prefect_server/models.py` and `backend/infrahub/prefect_server/database.py` - [ ] T045 [P] [US3] Changelog for the configurable activity log retention: `changelog/+task-manager-retention.added.md` already covers the unchanged 7-day default and how to raise it; add the sizing guidance there only if the documentation does not carry it ### Evidence for User Story 3 (opsmill/infrahub-private-tests) @@ -145,7 +145,7 @@ Run through the `test-dataset` workflow (`test_filter=...`) on restored backups **Independent Test**: Start the task manager and the separate background services with valid and invalid values (quickstart Part 1 steps 1, 2, 8 and Part 3 step 2). -- [X] T047 [P] [US4] Unit test: `create_infrahub_prefect` refuses to start on a retention under 1 day, logs a warning when own events are capped, and logs a warning per pre-set `PREFECT_*` variable, in `backend/tests/unit/prefect_server/test_app_retention.py` +- [X] T047 [P] [US4] Unit test: `create_infrahub_prefect` refuses to start on a retention under 1 day, logs a warning per pre-set `PREFECT_*` retention variable, and logs a warning per cleanup that a pre-set vacuum switch turns off, in `backend/tests/unit/prefect_server/test_app_retention.py` - [ ] T048 [US4] Add `infrahub tasks background-services [CONFIG_FILE]`: load the configuration, apply `apply_prefect_retention_env`, then start Prefect's background services in the foreground as `prefect server services start` does, in `backend/infrahub/cli/tasks.py` - [ ] T049 [P] [US4] Switch `task-manager-background-svc` to `command: infrahub tasks background-services` in `python_testcontainers/infrahub_testcontainers/docker-compose.test.yml` and `python_testcontainers/infrahub_testcontainers/docker-compose-cluster.test.yml` - [ ] T050 [US4] Open the opsmill/infrahub-helm PR for the same release: background-services deployment runs `infrahub tasks background-services`, the upgrade hook passes `--no-task-history-cleanup`, retention values exposed in values.yaml (outside this repository) diff --git a/docs/docs/reference/configuration.mdx b/docs/docs/reference/configuration.mdx index 672d1fd156f..f46367fa1a5 100644 --- a/docs/docs/reference/configuration.mdx +++ b/docs/docs/reference/configuration.mdx @@ -446,9 +446,9 @@ How long the task manager keeps its task history and activity log. | Name | Description | Type | Default | |------|-------------|------|---------| -| `INFRAHUB_TASK_MANAGER_RETENTION_TASK_HISTORY` | 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. | string | 30d | -| `INFRAHUB_TASK_MANAGER_RETENTION_ACTIVITY_LOG` | 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. | string | 7d | -| `INFRAHUB_TASK_MANAGER_RETENTION_PREFECT_OWN_EVENTS` | 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. | string | 7d | +| `INFRAHUB_TASK_MANAGER_RETENTION_TASK_HISTORY` | 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_RETENTION_PERIOD` takes precedence when set. When `PREFECT_SERVER_SERVICES_DB_VACUUM_ENABLED` is set to a value that does not turn on `flow_runs`, old task runs are not deleted. | string | 30d | +| `INFRAHUB_TASK_MANAGER_RETENTION_ACTIVITY_LOG` | 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, and at least `prefect_own_events`. `PREFECT_SERVER_EVENTS_RETENTION_PERIOD`, or its legacy name `PREFECT_EVENTS_RETENTION_PERIOD`, takes precedence when set. When `PREFECT_SERVER_SERVICES_DB_VACUUM_ENABLED` is set to a value that does not turn on `events`, old events are not deleted. | string | 7d | +| `INFRAHUB_TASK_MANAGER_RETENTION_PREFECT_OWN_EVENTS` | 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, and at most `activity_log`. `PREFECT_SERVER_SERVICES_DB_VACUUM_EVENT_RETENTION_OVERRIDES` takes precedence when set. When `PREFECT_SERVER_SERVICES_DB_VACUUM_ENABLED` is set to a value that does not turn on `events`, old events are not deleted. | string | 7d | ## Logging