diff --git a/docs/data-plane/source-flow.md b/docs/data-plane/source-flow.md index cf2be42c..a73b873f 100644 --- a/docs/data-plane/source-flow.md +++ b/docs/data-plane/source-flow.md @@ -102,7 +102,7 @@ Those topics are the engine's for the source's whole life, and nothing depends o A fetcher's composed config carries ONE connector type, because its source does: the composition refuses a second type, naming the source and the types, unless `DFE_ENGINE_SOURCE_ALLOW_MIXED_FETCHER_TYPES` says otherwise (see [source-definition.md](source-definition.md)). -Every compiled block lands in the deploy repo under `values/`, Argo CD applies it on Kubernetes, and a hand edit to any of them is reported as drift and re-synced on the next source write. A block carries only the keys the sources derive, so the deployment's own settings survive it. Compose has no Argo, so there the engine renders each app's own config file from those blocks into a directory the containers mount. A Compose deployment that has the engine render nothing runs its apps on the static config files the stack mounts, so a source written there compiles but never reaches the receiver, and `GET /api/v1/system/deployment` reports `applies_routing: false`. The receiver's destinations are a named set: the loader and the archiver are destinations, and a rule sends a matched record to one by name. The fetcher carries the same shape as `output.routes` over its default destination. +Every compiled block lands in the deploy repo under `values/`, Argo CD applies it on Kubernetes, and a hand edit to any of them is reported as drift and re-synced on the next source write. A block carries only the keys the sources derive, so the deployment's own settings survive it. Compose has no Argo, so there the engine renders each app's own config file from those blocks into a directory the containers mount. There one fetcher polls every fetched source from one merged file (apps.yaml `shared_instance`). A transform needs a container per source, so a published install refuses any transformed source outside `DFE_ENGINE_DEPLOYMENT_HOSTED_INSTANCES`. A Compose deployment that has the engine render nothing runs its apps on the static config files the stack mounts, so a source written there compiles but never reaches the receiver, and `GET /api/v1/system/deployment` reports `applies_routing: false`. The receiver's destinations are a named set: the loader and the archiver are destinations, and a rule sends a matched record to one by name. The fetcher carries the same shape as `output.routes` over its default destination. ## How a change reaches a pod diff --git a/openapi/openapi.json b/openapi/openapi.json index 99106491..2ebe0dc7 100644 --- a/openapi/openapi.json +++ b/openapi/openapi.json @@ -47162,18 +47162,6 @@ "title": "Cache", "description": "Allow cached results", "default": true - }, - "store": { - "anyOf": [ - { - "type": "string" - }, - { - "type": "null" - } - ], - "title": "Store", - "description": "Target store (database/schema/topic) - only if query allows" } }, "type": "object", diff --git a/src/dfe_engine/api/app.py b/src/dfe_engine/api/app.py index ee3b33b3..34460727 100644 --- a/src/dfe_engine/api/app.py +++ b/src/dfe_engine/api/app.py @@ -185,6 +185,10 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None]: logger.warning("Governed Ops gitcrud unavailable", error=str(exc)) gitcrud = None app.state.gitcrud = gitcrud + from dfe_engine.appmgmt import appconfig + + # The deploy repo's own write check judges an overlay by this deployment. + appconfig.judge_writes_by(lambda: app.state.settings) # Forge client for opening review PRs when a production+team write may not # commit straight to main (gitcrud/routing.py). None -> that posture refuses. @@ -320,6 +324,11 @@ def schema_converged() -> None: from dfe_engine.api.deps import get_source_registry_optional source_registry = get_source_registry_optional() + from dfe_engine.orgs.tenancy import Tenancy + from dfe_engine.source.registry import read_tenancy_from + + # A source save checks a fetched source's org against the one shared fetcher. + read_tenancy_from(functools.partial(Tenancy.read, app.state.org_registry)) # Ahead of the reconcile, which compiles the profiles a source binds. if gitcrud is not None: from dfe_engine.ingest.store import IngestStore @@ -693,6 +702,10 @@ def reconcile_service_roles_now() -> ReconcileResult: await asyncio.gather(*autostart_tasks, return_exceptions=True) shutdown_registries() + appconfig.judge_writes_by(None) + from dfe_engine.source.registry import read_tenancy_from + + read_tenancy_from(None) logger.info("DFE Engine API stopped") diff --git a/src/dfe_engine/api/errors.py b/src/dfe_engine/api/errors.py index 0b3ebdfe..4d51c03b 100644 --- a/src/dfe_engine/api/errors.py +++ b/src/dfe_engine/api/errors.py @@ -193,6 +193,7 @@ def install_exception_handlers(app: FastAPI) -> None: from dfe_engine.api.password_floor import count_floor_refusal, is_floor_refusal from dfe_engine.api.task_manager import TaskCapacityError from dfe_engine.api.v1.scim import SCIM_ROOT, scim_error + from dfe_engine.appmgmt.appconfig import SharedInstanceConflictError from dfe_engine.appmgmt.contract import RefusedNullError from dfe_engine.auth.models import AuthenticationError, AuthorizationError from dfe_engine.auth.protected_accounts import ProtectedAccountError @@ -297,6 +298,11 @@ async def invalid_name(_request: Request, exc: InvalidNameError): async def refused_null(_request: Request, exc: RefusedNullError): return _answer(400, exc.code, str(exc), paths=exc.paths) + # Any write path that would give a shared container an instance it cannot run. + @app.exception_handler(SharedInstanceConflictError) + async def shared_instance_conflict(_request: Request, exc: SharedInstanceConflictError): + return _answer(409, exc.code, str(exc)) + @app.exception_handler(ResourceNotFoundError) async def resource_not_found(_request: Request, _exc: ResourceNotFoundError): return _answer(404, ErrorCode.NOT_FOUND, "The resource does not exist") diff --git a/src/dfe_engine/api/v1/apps.py b/src/dfe_engine/api/v1/apps.py index 6ccdcf4a..a86ddd6f 100644 --- a/src/dfe_engine/api/v1/apps.py +++ b/src/dfe_engine/api/v1/apps.py @@ -123,7 +123,7 @@ from dfe_engine.hunts import runner_metrics from dfe_engine.sampling import SampleMode, SampleRequest, SamplerError from dfe_engine.settings import DFESettings -from dfe_engine.source.flow import stage_instance_ceiling +from dfe_engine.source.flow import hosting_refusal, stage_instance_ceiling from dfe_engine.source.registry import SourceNotFoundError router = APIRouter(prefix="/apps", tags=["App Management"], dependencies=[WRITE_TURN]) @@ -1031,7 +1031,12 @@ def create_instance( raise ProblemError( 409, f"{service}/{body.instance} is already deployed", code="already_exists" ) - ceiling = stage_instance_ceiling(app.descriptor, request.app.state.settings) + settings = request.app.state.settings + # The same rule a source save meets: a fixed set of containers runs named instances only. + refusal = hosting_refusal(app.descriptor, app.instance, "run it", settings) + if refusal: + raise ProblemError(409, refusal, code="not_hosted") + ceiling = stage_instance_ceiling(app.descriptor, settings) allowed, reason = instances.additional_instance_allowed(gc, app, ceiling) if not allowed: raise ProblemError(409, reason, code="single_instance_app") diff --git a/src/dfe_engine/api/v1/sources.py b/src/dfe_engine/api/v1/sources.py index 5ed3acb9..4d6cef0f 100644 --- a/src/dfe_engine/api/v1/sources.py +++ b/src/dfe_engine/api/v1/sources.py @@ -103,6 +103,7 @@ instances, ) from dfe_engine.appmgmt import catalogue as app_catalogue +from dfe_engine.appmgmt.appconfig import SharedInstanceConflictError from dfe_engine.auth.audit import audit_resource_change from dfe_engine.auth.models import AuthContext from dfe_engine.auth.rbac_scopes import scopes_dict @@ -517,7 +518,7 @@ def reconcile_derived_apps(request: Request, user: Any, registry: Any) -> AppsSy # report of that change: the render below finds the file current. hints += written.restart_required done.append(change.describe()) - except ProblemError as exc: + except (ProblemError, SharedInstanceConflictError) as exc: logger.warning(f"apps not reconciled with the sources: {exc}") return AppsSync(changes=done, error=str(exc), restart_required=_one_per_app(hints)) except Exception as exc: diff --git a/src/dfe_engine/appmgmt/appconfig.py b/src/dfe_engine/appmgmt/appconfig.py index e68b9cf9..18d566be 100644 --- a/src/dfe_engine/appmgmt/appconfig.py +++ b/src/dfe_engine/appmgmt/appconfig.py @@ -27,6 +27,10 @@ deployer reads it to declare a container per instance, the way the layer2-apps ApplicationSet reads an Application per overlay on Kubernetes. +A per-config app declaring ``shared_instance`` runs ONE container for every +instance instead, so it gets one file at its app directory with each instance's +block at that path merged into it, and no index. + Two facts make it generic. Which file an app reads is the manifest's ``consumes.config``, and how a rendered file set reaches the app is the file set's ``dir_setting`` (one directory) or ``entries_path`` (one entry per file), @@ -43,9 +47,11 @@ operator and an engine holding the docker socket would be a second one. """ +import copy import os import shutil import stat +from collections.abc import Callable from dataclasses import dataclass from pathlib import Path from typing import Any @@ -53,7 +59,7 @@ from scalo.logger import logger from dfe_engine.gitcrud import GitCrud -from dfe_engine.gitcrud.engine import ResourceNotFoundError, get_path, set_path +from dfe_engine.gitcrud.engine import ResourceNotFoundError, del_path, get_path, set_path from dfe_engine.yaml_utils import ( YAMLError, deep_merge, @@ -63,8 +69,15 @@ ) from . import files, instances -from .catalogue import APP_CATALOGUE, AppDescriptor, ConsumedFileSet, Multiplicity, ReloadMode -from .scaling import DeployTarget +from .catalogue import ( + APP_CATALOGUE, + AppDescriptor, + ConsumedFileSet, + Multiplicity, + ReloadMode, + render_source_binding, +) +from .scaling import DeployTarget, runs_fixed_containers CONFIG_ROOT = "config" """The overlay key holding what becomes the app's own config file. @@ -108,10 +121,10 @@ chains. """ -RESTART_HINT = f"restart required: {APPLY_COMMAND}" +RESTART_HINT = "restart required: {command}" """What an operator runs to apply a write the running app cannot take in place.""" -RECREATE_HINT = f"recreate required: {APPLY_COMMAND}" +RECREATE_HINT = "recreate required: {command}" """Compose reads env_file at up time, so a restart keeps the old environment.""" REMOVE_HINT = "removal required: make apply" @@ -122,6 +135,43 @@ class AppConfigError(RuntimeError): """Raised when a rendered config cannot be written.""" +class SharedInstanceConflictError(ValueError): + """Raised when an overlay write would give a shared container an instance it cannot run beside the others.""" + + code = "shared_instance_conflict" + """The refusal code a route answers with.""" + + +_write_settings: Callable[[], Any] | None = None +"""The deployment settings an overlay write is judged by, set by the process that writes.""" + + +def judge_writes_by(settings: Callable[[], Any] | None) -> None: + """Set where an overlay write reads the deployment settings from, or stop judging with None. + + The engine and the local CLI set it once they hold their settings, so the + deploy repo's own write check knows which containers this deployment runs. + """ + global _write_settings + _write_settings = settings + + +def require_shared_fit(gc: GitCrud, name: str, doc: dict) -> None: + """Refuse an instance overlay the one container its app shares here cannot take. + + Raises: + SharedInstanceConflictError: The instance clashes with one the container runs. + """ + if _write_settings is None or not instances.is_deployed_overlay(name): + return + app = instances.parse_overlay_name(name) + if app is None: + return + refusal = shared_refusal(gc, app.descriptor, app.instance, doc, _write_settings()) + if refusal: + raise SharedInstanceConflictError(refusal) + + def container_name(service: str, instance: str) -> str: """The Compose service one render runs as. @@ -153,6 +203,12 @@ class RenderedApp: removed: bool = False """Whether this instance left the deploy repo, so its container has no config left to read.""" + not_run: str = "" + """Why the shared container this instance would run in cannot take it. Empty where it runs.""" + + apply_command: str = APPLY_COMMAND + """The command that applies a render to one service, ``{service}`` where its name goes.""" + @property def container(self) -> str: """The Compose service carrying this render, which is what an operator acts on.""" @@ -161,15 +217,19 @@ def container(self) -> str: @property def restart_hint(self) -> str: """The command that applies this change, or empty when none is needed.""" + if self.not_run: + return f"not run: {self.not_run}" if self.removed: - return REMOVE_HINT + if self.apply_command == APPLY_COMMAND: + return REMOVE_HINT + return f"removal required: stop and remove the {self.container} container" # A changed env file changes the service definition, which `up` recreates # on, and `up` is also what CREATES the container a new instance lacks. if self.custom_env_changed or self.created: - return RECREATE_HINT.format(service=self.container) + return RECREATE_HINT.format(command=self.apply_command.format(service=self.container)) if not self.restart_required: return "" - return RESTART_HINT.format(service=self.container) + return RESTART_HINT.format(command=self.apply_command.format(service=self.container)) def enabled(settings: Any) -> bool: @@ -185,6 +245,16 @@ def enabled(settings: Any) -> bool: return bool(str(settings.deployment.app_config_dir or "").strip()) +def apply_command(settings: Any) -> str: + """The command an operator runs to apply a render to one service here. + + A deployer with no checkout names its own, so the hint never sends an operator + to a Makefile that is not there. + """ + named = str(settings.deployment.apply_command or "").strip() + return named or APPLY_COMMAND + + def _out_root(settings: Any) -> Path: return Path(str(settings.deployment.app_config_dir).strip()) @@ -573,6 +643,173 @@ def _clear_app_level_custom_env(settings: Any, app: AppDescriptor) -> None: ) +def shares_one_container(app: AppDescriptor, settings: Any) -> bool: + """Whether every instance of *app* runs in one container here, from one merged file.""" + return bool(app.shared_instance) and runs_fixed_containers( + DeployTarget(settings.deployment.target) + ) + + +def _merge_disjoint(into: dict[str, Any], block: dict[str, Any], prefix: str) -> str: + """Merge *block* into *into* in place, or name the first key both set differently. + + A mapping both carry is merged key by key, so two instances of a keyed family + meet under one family. Any other value set on both sides must be the same, + because one process reads one. *into* is left part-merged on a clash, so pass + a copy. + """ + for key, value in block.items(): + path = f"{prefix}.{key}" + if key not in into: + into[key] = copy.deepcopy(value) + elif isinstance(into[key], dict) and isinstance(value, dict): + clash = _merge_disjoint(into[key], value, path) + if clash: + return clash + elif into[key] != value: + return path + return "" + + +def _shared_merge( + app: AppDescriptor, docs: list[tuple[str, dict]] +) -> tuple[dict[str, Any], dict[str, str]]: + """Merge instance overlays, in the order given, into the one overlay a shared container runs. + + Each instance's block at ``shared_instance`` is merged in. The rest of its + config and its ``extraEnv`` must match the first instance's, because the one + process reads one value of each. The bindings tie an instance to its source, + so the shared container binds to the app's own name instead. + + Returns: + The merged overlay, empty where no instance merged, and why each + instance left out could not be, by instance name. + """ + merge_at = _inner_path(app.shared_instance) + bindings = [_inner_path(path) for path in app.source_binding] + merged: dict[str, Any] = {} + agreed: dict[str, Any] | None = None + first = "" + refused: dict[str, str] = {} + for name, doc in docs: + config = _config_block(doc) + block = get_path(config, merge_at, default=None) + rest = copy.deepcopy(config) + for path in (merge_at, *bindings): + del_path(rest, path) + held = {CONFIG_ROOT: rest, ENV_ROOT: custom_env(doc)} + candidate = copy.deepcopy(merged) + reason = "" + if agreed is not None and held != agreed: + reason = f"its {', '.join(_differing_keys(held, agreed))} differs from {first}'s" + elif isinstance(block, dict): + clash = _merge_disjoint(candidate, block, app.shared_instance) + if clash: + reason = f"its {clash} is already set by another source" + if reason: + refused[name] = ( + f"source {name!r} cannot share the one {app.service} this deployment runs: {reason}" + ) + continue + merged = candidate + if agreed is None: + agreed, first = held, name + if agreed is None: + return {}, refused + config = copy.deepcopy(agreed[CONFIG_ROOT]) + set_path(config, merge_at, merged) + for path, value in render_source_binding(app, app.service).items(): + set_path(config, _inner_path(path), value) + return {CONFIG_ROOT: config, ENV_ROOT: agreed[ENV_ROOT]}, refused + + +def _contains(outer: Any, inner: Any) -> bool: + """Whether every value *inner* sets is set the same in *outer*.""" + if isinstance(inner, dict): + if not isinstance(outer, dict): + return False + return all(key in outer and _contains(outer[key], value) for key, value in inner.items()) + return outer == inner + + +def _running_first( + gc: GitCrud, app: AppDescriptor, settings: Any, *, leave_out: str = "" +) -> list[tuple[str, dict]]: + """Every instance overlay of *app*, the ones its container already runs first. + + An instance runs when the file rendered last carries its whole block, so a + newcomer that clashes is the one left out and never evicts a running one. + Name order breaks the tie, so a render with no file yet is stable too. + """ + merge_at = _inner_path(app.shared_instance) + target = _out_root(settings) / app.service / app.config_file + running: Any = None + if target.is_file(): + try: + running = get_path(yaml_load_string(target.read_text(encoding="utf-8")) or {}, merge_at) + except YAMLError: # an unreadable last render shows nothing running + running = None + docs: list[tuple[bool, str, dict]] = [] + for instance in instances.list_instances(gc, service=app.service): + if instance.instance == leave_out: + continue + try: + doc = instances.read_overlay(gc, instance) + except ResourceNotFoundError: + continue + block = get_path(_config_block(doc), merge_at, default=None) + runs = running is not None and block is not None and _contains(running, block) + docs.append((not runs, instance.instance, doc)) + return [(name, doc) for _, name, doc in sorted(docs, key=lambda d: (d[0], d[1]))] + + +def shared_refusal(gc: GitCrud, app: AppDescriptor, instance: str, doc: dict, settings: Any) -> str: + """Why the one container this deployment runs for *app* cannot take *instance*, empty where it can. + + Judged against the instances already there, the running ones first, with + *doc* as the overlay the instance would carry, so a save is refused where a + render would leave it out. + """ + if not shares_one_container(app, settings) or not enabled(settings): + return "" + docs = [*_running_first(gc, app, settings, leave_out=instance), (instance, doc)] + _, refused = _shared_merge(app, docs) + return refused.get(instance, "") + + +def _merged_overlay( + gc: GitCrud, app: AppDescriptor, settings: Any +) -> tuple[dict[str, Any], list[RenderedApp]]: + """The one overlay a shared container runs, and a ``not_run`` render per instance left out.""" + merged, refused = _shared_merge(app, _running_first(gc, app, settings)) + app_dir = _out_root(settings) / app.service + not_run = [ + RenderedApp( + service=app.service, + instance=name, + directory=app_dir, + changed=True, + restart_required=False, + not_run=reason, + ) + for name, reason in sorted(refused.items()) + ] + return merged, not_run + + +def _differing_keys(held: dict[str, dict], agreed: dict[str, dict]) -> list[str]: + """Each ``.`` whose value differs between two instances' shared halves.""" + differs: list[str] = [] + for root, block in held.items(): + other = agreed[root] + differs.extend( + f"{root}.{key}" + for key in sorted(set(block) | set(other)) + if block.get(key) != other.get(key) + ) + return differs + + def _render_one( gc: GitCrud, app: AppDescriptor, @@ -580,14 +817,21 @@ def _render_one( settings: Any, out_root: Path, mount_root: str, + *, + doc: dict | None = None, ) -> RenderedApp: - """Write one instance's config file and file sets, and report what taking it costs.""" - doc: dict = {} - if instance is not None: - try: - doc = instances.read_overlay(gc, instance) - except ResourceNotFoundError: - doc = {} + """Write one instance's config file and file sets, and report what taking it costs. + + ``doc`` stands in for the instance's overlay, which is how a shared container's + merged overlay is rendered at the app's own directory. + """ + if doc is None: + doc = {} + if instance is not None: + try: + doc = instances.read_overlay(gc, instance) + except ResourceNotFoundError: + doc = {} # A per-config app's container reads its OWN instance directory, so two # sources of one connector never share a config file. @@ -635,6 +879,7 @@ def _render_one( custom_env_changed=env_changed, per_instance=per_instance, created=per_instance and first_write, + apply_command=apply_command(settings), ) @@ -730,6 +975,43 @@ def _prune_custom_env(settings: Any, service: str, keep: list[str]) -> None: _report_unwritable(target.name, directory, exc) +def _retire_instance_containers( + settings: Any, app: AppDescriptor, out_root: Path +) -> list[RenderedApp]: + """Take away what a per-instance render left for an app now run by one shared container. + + A leftover index would have the deployer declare those containers beside the + shared one, so the same sources would be polled twice. Each one named is + reported for removal, because its container may still be running. + """ + directory = custom_env_dir(settings) + index = directory / f"{app.service}{INSTANCE_INDEX_SUFFIX}" if directory else None + listed: set[str] = set() + if index is not None and index.is_file(): + listed = {line.strip() for line in index.read_text(encoding="utf-8").splitlines()} + dropped = _prune_instance_dirs(out_root / app.service, [fs.name for fs in app.files]) + if index is not None and index.is_file(): + _prune_custom_env(settings, app.service, []) + try: + index.unlink() + except OSError as exc: + _report_unwritable(index.name, index.parent, exc) + names = sorted(name for name in listed | set(dropped) if name) + return [ + RenderedApp( + service=app.service, + instance=name, + directory=out_root / app.service / name, + changed=True, + restart_required=False, + per_instance=True, + removed=True, + apply_command=apply_command(settings), + ) + for name in names + ] + + def render(gc: GitCrud, settings: Any) -> list[RenderedApp]: """Render every app's config from the deploy repo. Returns what was written. @@ -744,6 +1026,12 @@ def render(gc: GitCrud, settings: Any) -> list[RenderedApp]: out_root.mkdir(parents=True, exist_ok=True) written: list[RenderedApp] = [] for app in renderable(settings): + if shares_one_container(app, settings): + merged, not_run = _merged_overlay(gc, app, settings) + written.append(_render_one(gc, app, None, settings, out_root, mount_root, doc=merged)) + written.extend(not_run) + written.extend(_retire_instance_containers(settings, app, out_root)) + continue found = _instances_for(gc, app) written.extend( _render_one(gc, app, instance, settings, out_root, mount_root) for instance in found @@ -764,6 +1052,7 @@ def render(gc: GitCrud, settings: Any) -> list[RenderedApp]: restart_required=False, per_instance=True, removed=True, + apply_command=apply_command(settings), ) for name in dropped ) @@ -794,6 +1083,14 @@ def render_and_report(gc: GitCrud | None, settings: Any) -> list[str]: for entry in rendered: if not entry.changed: continue + if entry.not_run: + logger.warning( + "An instance is not run: the shared container cannot take its config", + app=entry.service, + instance=entry.instance, + reason=entry.not_run, + ) + continue if entry.removed: logger.info( "Removed the rendered config of an instance the deploy repo dropped", diff --git a/src/dfe_engine/appmgmt/apps.yaml b/src/dfe_engine/appmgmt/apps.yaml index f7d84ff2..a42a7204 100644 --- a/src/dfe_engine/appmgmt/apps.yaml +++ b/src/dfe_engine/appmgmt/apps.yaml @@ -90,6 +90,16 @@ # the rest, so one app can declare both halves. Only on a hot_reload app: on # any other every change is a restart already. # +# shared_instance +# How a target that runs a fixed container set (docker) hosts a per_config app: +# ONE container serves every instance from one config file at the app's own +# directory, the instances' blocks at this config. path merged into it. The +# rest of their config must agree, because one process reads one value, so an +# instance that differs or sets a merged key another already set is reported +# and not run. Kubernetes ignores it and still runs one deployment per +# instance. Absent means one process runs one instance, so such a target +# refuses a source that needs one. +# # endpoints # The listeners the app runs, by name. `push` is the scalo Push service a # direct-transport stage sends records to, and declaring it is ALSO how the @@ -463,11 +473,12 @@ apps: profiles: [slim, single, scale, mesh, docker-single] # On the Kubernetes tiers one instance per fetcher source, written by the # engine when the source is written, so no tier seeds one. docker-single - # starts one fetcher idle, and the engine accepts a source there only where - # it renders app config (DFE_ENGINE_DEPLOYMENT_APP_CONFIG_DIR). The instance - # ceiling is 0 without that and unbounded with it: dfe-engine - # appmgmt/scaling.py, instance_ceiling. + # runs one fetcher for every source, and the engine accepts a source there + # only where it renders app config (DFE_ENGINE_DEPLOYMENT_APP_CONFIG_DIR). default_in: [docker-single] + # One process polls many sources (instance_id prefixes each cursor key), so + # docker merges every instance's sources into the one container's file. + shared_instance: config.sources # Anything that can produce a record is work. The ingest listener defaults # off, so an instance that wants it sets config.ingest.enabled true. idle_when: @@ -478,11 +489,9 @@ apps: # rest of the stanza binds a connection, a topic or the concurrency # semaphore, all established once at startup (its config.example.yaml). hot_reload: true - # Whole blocks: a reload re-reads a source's filter and interval but adds no - # source, re-resolves no credential and rebuilds no output (dfe-fetcher#134), - # so a filter-only edit reports a restart it does not need. + # A reload starts, stops and rebuilds sources (dfe-fetcher#286). The output, + # the container extractors and the ingest listener are built once at startup. restart_paths: - - config.sources - config.output - config.extractors - config.ingest diff --git a/src/dfe_engine/appmgmt/catalogue.py b/src/dfe_engine/appmgmt/catalogue.py index 1671bfcf..d32b6d22 100644 --- a/src/dfe_engine/appmgmt/catalogue.py +++ b/src/dfe_engine/appmgmt/catalogue.py @@ -261,6 +261,13 @@ class AppDescriptor: app declares both halves rather than spending ``hot_reload`` on one. """ + shared_instance: str = "" + """The overlay path one container merges across every instance, where a target runs a fixed container set. + + ``config.`` rooted. Empty where one process runs one instance only, so such + a target cannot run a second. + """ + endpoints: dict[str, AppEndpoint] = field(default_factory=dict) """The listeners this app runs, by name. Absent means the app has none.""" @@ -530,6 +537,9 @@ def _descriptor_from(service: str, raw: dict) -> AppDescriptor: service, raw.get("reload_setting"), hot_reload=hot_reload ), restart_paths=_restart_paths_from(service, raw.get("restart_paths"), hot_reload=hot_reload), + shared_instance=_shared_instance_from( + service, raw.get("shared_instance"), multiplicity=multiplicity + ), endpoints=_endpoints_from(service, raw.get("endpoints")), variant_path=str(raw.get("variant_path", "")), mode_path=str(raw.get("mode_path", "")), @@ -670,6 +680,28 @@ def _restart_paths_from(service: str, raw: object, *, hot_reload: bool) -> tuple return paths +def _shared_instance_from(service: str, raw: object, *, multiplicity: Multiplicity) -> str: + """The overlay path a shared container merges across instances, empty when it names none. + + Refused on a single-deployment app, which has one instance to share already, + and anywhere but under ``config.``: it is merged inside the app's own file. + """ + if raw is None: + return "" + path = str(raw) + if multiplicity is not Multiplicity.PER_CONFIG: + raise CatalogueError( + f"{service}: shared_instance {path!r} needs multiplicity per_config, " + "because a single-deployment app runs one instance already" + ) + if not path.startswith("config.") or path == "config.": + raise CatalogueError( + f"{service}: shared_instance {path!r} must be a config. overlay path, " + "because it is merged inside the app's own config file" + ) + return path + + def _routing_paths_from(service: str, raw: object) -> dict[str, str]: """The overlay dot-path of each block this app's compiler produces.""" if raw is None: diff --git a/src/dfe_engine/appmgmt/derived.py b/src/dfe_engine/appmgmt/derived.py index 8a2f998d..b28832a0 100644 --- a/src/dfe_engine/appmgmt/derived.py +++ b/src/dfe_engine/appmgmt/derived.py @@ -36,13 +36,15 @@ from dataclasses import dataclass from typing import Any, Literal +from scalo.logger import logger + from dfe_engine.gitcrud import GitCrud from dfe_engine.gitcrud.commit_policy import CommitContext, build_message from dfe_engine.gitcrud.engine import ResourceNotFoundError from dfe_engine.orgs.tenancy import Tenancy from dfe_engine.source.reader import SourceReader -from . import catalogue, files, instances, routing +from . import appconfig, catalogue, files, instances, routing from .instances import AppInstance DerivedAction = Literal["deploy", "sync", "remove", "forget"] @@ -170,7 +172,12 @@ def reconcile( keep_parked(gc, change, actor) gc.delete(instances.HELMVARS_CLASS, name, actor, message=message) else: - gc.put(instances.HELMVARS_CLASS, name, change.doc or {}, actor, message=message) + try: + gc.put(instances.HELMVARS_CLASS, name, change.doc or {}, actor, message=message) + except appconfig.SharedInstanceConflictError as exc: + # The deploy repo kept the overlay, so the shared container runs as it did. + logger.warning("an app overlay was not written", app=name, reason=str(exc)) + continue if change.unpark: drop_parked(gc, change.app, actor) done.append(change.describe()) diff --git a/src/dfe_engine/appmgmt/scaling.py b/src/dfe_engine/appmgmt/scaling.py index a9722e83..8f34e9fb 100644 --- a/src/dfe_engine/appmgmt/scaling.py +++ b/src/dfe_engine/appmgmt/scaling.py @@ -17,6 +17,7 @@ """ import re +from collections.abc import Collection from dataclasses import dataclass from enum import StrEnum @@ -73,7 +74,12 @@ class ScalingDials: def instance_ceiling( - app: AppDescriptor, target: DeployTarget, *, writes_app_config: bool = False + app: AppDescriptor, + target: DeployTarget, + *, + writes_app_config: bool = False, + reads_instance_index: bool = True, + hosted: Collection[str] = (), ) -> int | None: """How many deployments of *app* this target can RUN, None meaning unbounded. @@ -82,24 +88,38 @@ def instance_ceiling( A single-deployment app is one everywhere. A per-config app is unbounded on Kubernetes, where each overlay renders its own Argo Application, and equally - unbounded on Compose WHERE THE ENGINE RENDERS THE APP CONFIG: each instance - gets its own directory and its name goes to the deployer's instance index, so - the deployer declares a container per instance on the next up. Where nothing - renders it, Compose holds only the services its committed file declares and - the source would be saved and never run, which is a ceiling of zero. A target - nobody named is not assumed to be Compose. + unbounded on Compose WHERE THE ENGINE RENDERS THE APP CONFIG: an app declaring + ``shared_instance`` reads every instance from its one container's merged file, + and any other gets a directory per instance whose name goes to the deployer's + instance index. A deployer that reads no index runs only the instances it + declared, so the ceiling is how many those are. Where nothing renders it, + Compose holds only the services its committed file declares and the source + would be saved and never run, which is a ceiling of zero. A target nobody + named is not assumed to be Compose. Args: app: The app the ceiling is being asked about. target: Where this deployment runs. writes_app_config: Whether the engine renders the apps' config files here, which is ``appconfig.enabled``. + reads_instance_index: Whether a deployer declares a container per name + in the instance index the engine writes. + hosted: The instances of *app* the deployment declared a container for. """ if app.multiplicity is not Multiplicity.PER_CONFIG: return 1 - if target is not DeployTarget.DOCKER: + if not runs_fixed_containers(target): + return None + if not writes_app_config: + return 0 + if app.shared_instance or reads_instance_index: return None - return None if writes_app_config else 0 + return len(hosted) + + +def runs_fixed_containers(target: DeployTarget) -> bool: + """Whether *target* runs only the containers its deployer declared, so per-config instances share one.""" + return target is DeployTarget.DOCKER def support(app: AppDescriptor, target: DeployTarget) -> tuple[bool, str]: diff --git a/src/dfe_engine/auth/deployment_hints.py b/src/dfe_engine/auth/deployment_hints.py index a0ad4580..795f3f61 100644 --- a/src/dfe_engine/auth/deployment_hints.py +++ b/src/dfe_engine/auth/deployment_hints.py @@ -47,8 +47,15 @@ def credential_fetch_command( namespace: str = "", secret_name: str = "", secret_key: str = "", + deployer_command: str = "", ) -> str: - """The one-line command that prints this deployment's admin password.""" + """The one-line command that prints this deployment's admin password. + + ``deployer_command`` is ``deployment.credential_command``, which wins: an + install published as one compose file has no checkout and so no ``make``. + """ + if deployer_command.strip(): + return deployer_command.strip() if kind == DOCKER: return "make creds" if kind == KUBERNETES: diff --git a/src/dfe_engine/cli/auto/local.py b/src/dfe_engine/cli/auto/local.py index b0cf92e3..abee2a18 100644 --- a/src/dfe_engine/cli/auto/local.py +++ b/src/dfe_engine/cli/auto/local.py @@ -112,6 +112,10 @@ def _build_crud(repo: str | None) -> GitCrud: read=gs.read, pack=gs.pack, ) + from dfe_engine.appmgmt import appconfig + + # The deploy repo's own write check judges an overlay by this deployment. + appconfig.judge_writes_by(get_settings) return GitCrud(grepo) diff --git a/src/dfe_engine/deployment_contract.py b/src/dfe_engine/deployment_contract.py index 53d8f944..11b7f55a 100644 --- a/src/dfe_engine/deployment_contract.py +++ b/src/dfe_engine/deployment_contract.py @@ -166,7 +166,7 @@ def engine_deployment_contract() -> DeploymentContract: ), ], depends_on=["clickhouse"], - # KEDA not used for dfe-engine (control plane -- HPA on CPU is sufficient). + # Nothing scales the engine: it runs one pod (singleton below). keda=None, # The YAML SSoT directory DFE_CONFIG_DIR names, on a claim so the config outlives the pod. writable_paths=[WritablePath(name="config", path="/config", persistent=True, size="1Gi")], diff --git a/src/dfe_engine/gitcrud/engine.py b/src/dfe_engine/gitcrud/engine.py index 78b8f556..d515d96c 100644 --- a/src/dfe_engine/gitcrud/engine.py +++ b/src/dfe_engine/gitcrud/engine.py @@ -219,12 +219,13 @@ def check(self, cls_name: str, name: str, doc: dict) -> None: Raises: Exception: Whatever the class's check raises for a doc its reader refuses. """ - refuse = self._cls(cls_name).check - if refuse is not None: - refuse(name, doc) + cls = self._cls(cls_name) + if cls.check is not None: + cls.check(name, doc) + if cls.repo_check is not None: + cls.repo_check(self, name, doc) - @staticmethod - def _yaml(resource_class: ResourceClass, name: str, rel: str, doc: dict) -> str: + def _yaml(self, resource_class: ResourceClass, name: str, rel: str, doc: dict) -> str: """*doc* as YAML that reads back as *doc*, the one way a resource reaches a commit. The class's ``check`` runs first, so no writer commits a doc its reader refuses. @@ -236,6 +237,8 @@ def _yaml(resource_class: ResourceClass, name: str, rel: str, doc: dict) -> str: """ if resource_class.check is not None: resource_class.check(name, doc) + if resource_class.repo_check is not None: + resource_class.repo_check(self, name, doc) return yaml_dump_string(doc, target=f"deploy-repo:{rel}") def reading(self) -> AbstractContextManager[None]: diff --git a/src/dfe_engine/gitcrud/models.py b/src/dfe_engine/gitcrud/models.py index 5fd00e41..837a7ba6 100644 --- a/src/dfe_engine/gitcrud/models.py +++ b/src/dfe_engine/gitcrud/models.py @@ -17,6 +17,7 @@ from collections.abc import Callable from dataclasses import dataclass, field from enum import StrEnum +from typing import Any class Layout(StrEnum): @@ -57,6 +58,8 @@ class (helmvars, actions) is unversioned - the git log is enough. through the same name rule. Only the separator is permitted. check opt-in: called with a resource's name and doc on every write of the class, and raises to refuse a doc its reader would refuse. + repo_check opt-in: as ``check``, with the deploy repo being written first, for + a refusal that reads the resource's siblings. """ name: str @@ -68,6 +71,7 @@ class (helmvars, actions) is unversioned - the git log is enough. manifest: str = "manifest.yaml" nested: bool = False check: Callable[[str, dict], None] | None = field(default=None, compare=False) + repo_check: Callable[[Any, str, dict], None] | None = field(default=None, compare=False) @property def is_bundle(self) -> bool: diff --git a/src/dfe_engine/gitcrud/registry.py b/src/dfe_engine/gitcrud/registry.py index 31d010a3..592bb641 100644 --- a/src/dfe_engine/gitcrud/registry.py +++ b/src/dfe_engine/gitcrud/registry.py @@ -13,6 +13,7 @@ """ from collections.abc import Iterable +from typing import Any from .models import Layout, ResourceClass @@ -32,6 +33,18 @@ def _check_app_overlay(name: str, doc: dict) -> None: check_overlay(doc) +def _check_shared_container(repo: Any, name: str, doc: dict) -> None: + """Refuse an instance overlay the one container its app shares here cannot run. + + At the write itself, so every writer of the class meets it: the API, an action, + the generic key routes, the seeds and the local CLI. + """ + # At call time: appmgmt reads the deploy repo through this package. + from dfe_engine.appmgmt.appconfig import require_shared_fit + + require_shared_fit(repo, name, doc) + + class UnknownResourceClassError(KeyError): """Raised when a resource-class name is not registered.""" @@ -83,7 +96,13 @@ def default_registry() -> ResourceClassRegistry: return ResourceClassRegistry( [ # helmvars class - deployment overlays - ResourceClass("helmvars", "values", rbac_prefix="helmvars", check=_check_app_overlay), + ResourceClass( + "helmvars", + "values", + rbac_prefix="helmvars", + check=_check_app_overlay, + repo_check=_check_shared_container, + ), # The same class for the charts that are NOT app instances - the # backing-service and platform overlays. A separate DIRECTORY because # anything under values/ becomes an Argo application: the appset's diff --git a/src/dfe_engine/governance/actions.py b/src/dfe_engine/governance/actions.py index a52790ab..147b3c5b 100644 --- a/src/dfe_engine/governance/actions.py +++ b/src/dfe_engine/governance/actions.py @@ -31,6 +31,7 @@ from scalo.logger import logger from dfe_engine.appmgmt import contract, instances +from dfe_engine.appmgmt.appconfig import SharedInstanceConflictError from dfe_engine.appmgmt.catalogue import AppDescriptor from dfe_engine.appmgmt.scaling import keda_by_default from dfe_engine.gitcrud import GitCrud, ResourceNotFoundError, del_path, get_path, set_path @@ -529,7 +530,7 @@ def _document_refusal(self, cls: str, name: str, doc: dict) -> str | None: """ try: self._crud.check(cls, name, doc) - except contract.RefusedNullError as exc: + except (contract.RefusedNullError, SharedInstanceConflictError) as exc: return str(exc) try: if cls == HUNT_CLASS: diff --git a/src/dfe_engine/hunts/runner_cli.py b/src/dfe_engine/hunts/runner_cli.py index 4cd59875..89ea5433 100644 --- a/src/dfe_engine/hunts/runner_cli.py +++ b/src/dfe_engine/hunts/runner_cli.py @@ -16,8 +16,8 @@ Two commands, matching the two ways the runner is driven: - materialise - the ONE-SHOT the Argo PostSync hook (or the engine, on a hunt-config - change) runs to write the deterministic ``hunt_schedule`` table KEDA scales on. + materialise - the ONE-SHOT run to write the deterministic ``hunt_schedule`` table + KEDA scales on. It is deploy-time, NOT the hot path - it runs once per config change, not per tick, so it just publishes and exits. @@ -187,9 +187,9 @@ def _match_notifier(settings: DFESettings) -> MatchNotifier: def materialise() -> None: """Publish the deterministic ``hunt_schedule`` table KEDA scales on (one-shot). - Deploy-time, not the hot path: the Argo PostSync hook runs this once per - hunt-config change so KEDA can answer "is any hunt due?" with zero workers - running. Idempotent - re-materialising tombstones removed hunts. + Deploy-time, not the hot path: run once per hunt-config change so KEDA can + answer "is any hunt due?" with zero workers running. Idempotent - + re-materialising tombstones removed hunts. """ init_scalo_logger(runner_metrics.SERVICE_NAME, otel_tracing=False) settings = load_settings() diff --git a/src/dfe_engine/query/client.py b/src/dfe_engine/query/client.py index cd6ea66e..a88f4ce7 100644 --- a/src/dfe_engine/query/client.py +++ b/src/dfe_engine/query/client.py @@ -18,6 +18,8 @@ import uuid from typing import Any, Literal +from scalo.cli.openapi import Credential + from dfe_engine.query.models import ( AuthContext, QueryMetadata, @@ -26,11 +28,34 @@ from dfe_engine.query.result import QueryResult +def _refuse_unsendable(credential: Credential | None) -> None: + """Refuse a credential no header can carry, naming the fault and never the value. + + The HTTP stack quotes a rejected header whole in its error, so a value with a + trailing newline would put the token in a traceback and a log. + + Raises: + ValueError: The value holds a CR, LF or another control character. + """ + value = credential.value if credential is not None else None + if value is None: + return + if any(ord(char) < 0x20 or ord(char) == 0x7F for char in value): + kind = "API key" if credential is not None and credential.is_api_key else "token" + raise ValueError( + f"the {kind} holds a control character (CR, LF or another), so no header can " + "carry it: strip the whitespace it was read with" + ) + + class QueryClient: """Client for DFE Query API. Supports both direct (in-process) and HTTP modes. Clients specify query labels and parameters -- never raw SQL. + + HTTP mode sends ``credential`` the way the ``dfe`` CLI does: an API key in + ``api_key_header``, a token as ``Authorization: Bearer``. """ def __init__( @@ -38,13 +63,19 @@ def __init__( base_url: str | None = None, direct: bool = False, timeout_seconds: int = 30, + *, + credential: Credential | None = None, + api_key_header: str = "X-API-Key", ): if not base_url and not direct: raise ValueError("Must specify base_url or direct=True") + _refuse_unsendable(credential) self.base_url = base_url self.direct = direct self.timeout_seconds = timeout_seconds + self.credential = credential + self.api_key_header = api_key_header self._http_client = None self._view_executor = None @@ -58,6 +89,7 @@ def http_client(self): base_url=self.base_url, timeout=self.timeout_seconds + 10, retries=3, + headers=self._auth_headers(), ) return self._http_client @@ -76,7 +108,6 @@ def query( time_from: str | None = None, time_to: str | None = None, timeout_seconds: int | None = None, - store: str | None = None, cache: bool = True, ) -> QueryResult: """Execute query and return result. @@ -96,12 +127,24 @@ def query( time_from=time_from, time_to=time_to, timeout_seconds=timeout_seconds, - store=store, cache=cache, ) return self._execute(query_label, params, options) + def _auth_headers(self) -> dict[str, str]: + """The headers the engine authenticates this client by, built as the CLI builds them.""" + from scalo.cli.openapi import ScaloHttpTransport + from scalo.cli.openapi.client import Client + + cli_client = Client( + self.base_url or "", + self.credential, + ScaloHttpTransport(), + api_key_header=self.api_key_header, + ) + return cli_client.credentials() + def _execute( self, query_label: str, diff --git a/src/dfe_engine/query/models.py b/src/dfe_engine/query/models.py index 55ad43c1..105f2d7d 100644 --- a/src/dfe_engine/query/models.py +++ b/src/dfe_engine/query/models.py @@ -80,12 +80,6 @@ class QueryOptions(BaseModel): # Caching cache: bool = Field(default=True, description="Allow cached results") - # Store override (only if query definition allows store: '*') - store: str | None = Field( - default=None, - description="Target store (database/schema/topic) - only if query allows", - ) - # ============================================================================= # Query Response Models diff --git a/src/dfe_engine/services/default_configs/archiver-default.yaml b/src/dfe_engine/services/default_configs/archiver-default.yaml index 5175a385..f0e964f5 100644 --- a/src/dfe_engine/services/default_configs/archiver-default.yaml +++ b/src/dfe_engine/services/default_configs/archiver-default.yaml @@ -12,7 +12,7 @@ kafka: security_protocol: PLAINTEXT archive: - destination: "file:///tmp/dfe-archive" + destination: "file:///var/lib/dfe-archiver/archive" path_template: "{year}/{month}/{day}/{hour}" file_extension: jsonl roll_size_bytes: 1073741824 # 1 GB diff --git a/src/dfe_engine/settings.py b/src/dfe_engine/settings.py index a130dad0..ae5f7d8a 100644 --- a/src/dfe_engine/settings.py +++ b/src/dfe_engine/settings.py @@ -17,6 +17,7 @@ import json import os import re +import string from collections.abc import Callable from pathlib import Path from typing import Any, Literal @@ -1140,6 +1141,34 @@ class DeploymentSettings(BaseModel): "app's own chart. DFE_ENGINE_DEPLOYMENT_APP_ENV_DIR." ), ) + hosted_instances: str = Field( + default="", + description=( + "The per-config app instances a fixed-container deployment declared a " + "container for, as =, with apps separated by ';'. " + "Injected by a deployer that runs only the containers its file declares " + "and reads no instance index, so a source needing an instance outside it " + "is refused rather than saved and never run. " + "DFE_ENGINE_DEPLOYMENT_HOSTED_INSTANCES." + ), + ) + apply_command: str = Field( + default="", + description=( + "The command that applies a rendered config change to one Compose service, " + "with {service} where the service name goes, for a deployer with no " + "checkout to run make in. Empty means make apply. " + "DFE_ENGINE_DEPLOYMENT_APPLY_COMMAND." + ), + ) + credential_command: str = Field( + default="", + description=( + "How an operator of this deployment reads the admin password, where the " + "deployer's own way differs from its target's default. Shown on the " + "pre-login page. DFE_ENGINE_DEPLOYMENT_CREDENTIAL_COMMAND." + ), + ) max_replicas_ceiling: int = Field( default=MAX_REPLICAS_CEILING, gt=0, @@ -1159,6 +1188,53 @@ class DeploymentSettings(BaseModel): ), ) + @field_validator("apply_command") + @classmethod + def _apply_command_names_the_service(cls, value: str) -> str: + if not value.strip(): + return value + try: + parts = [part for part in string.Formatter().parse(value) if part[1] is not None] + except ValueError as exc: + raise ValueError(f"apply_command is not a template: {exc}") from exc + if [name for _, name, _, _ in parts] != ["service"]: + raise ValueError( + "apply_command must carry {service} once and no other field, where the " + "service name goes" + ) + if any(spec or conversion for _, _, spec, conversion in parts): + raise ValueError("apply_command's {service} takes no conversion or format spec") + # Formatted once here, so a hint never meets a template it cannot fill. + if "dfe-probe" not in value.format(service="dfe-probe"): + raise ValueError("apply_command does not put the service name into the command") + return value + + @field_validator("hosted_instances") + @classmethod + def _hosted_instances_parse(cls, value: str) -> str: + parse_hosted_instances(value) + return value + + +def parse_hosted_instances(raw: str) -> dict[str, frozenset[str]]: + """Read ``=,;=...`` into the instances per app. + + Raises: + ValueError: An entry names no app, or names one twice. + """ + hosted: dict[str, frozenset[str]] = {} + for entry in (part.strip() for part in raw.split(";")): + if not entry: + continue + app, sep, names = entry.partition("=") + app = app.strip() + if not sep or not app: + raise ValueError(f"hosted instance entry {entry!r} is not =,...") + if app in hosted: + raise ValueError(f"hosted instances name {app} twice") + hosted[app] = frozenset(n.strip() for n in names.split(",") if n.strip()) + return hosted + class TransformValidationSettings(BaseModel): """Syntax validation for authored transform files.""" @@ -2432,6 +2508,9 @@ def _strip_lower(value: str) -> str: "DFE_ENGINE_DEPLOYMENT_APP_CONFIG_BASE_DIR": "deployment.app_config_base_dir", "DFE_ENGINE_DEPLOYMENT_APP_CONFIG_MOUNT": "deployment.app_config_mount", "DFE_ENGINE_DEPLOYMENT_APP_ENV_DIR": "deployment.app_env_dir", + "DFE_ENGINE_DEPLOYMENT_HOSTED_INSTANCES": "deployment.hosted_instances", + "DFE_ENGINE_DEPLOYMENT_CREDENTIAL_COMMAND": "deployment.credential_command", + "DFE_ENGINE_DEPLOYMENT_APPLY_COMMAND": "deployment.apply_command", "DFE_ENGINE_DEPLOYMENT_KAFKA_PROVIDER": "deployment.backing.kafka.provider", "DFE_ENGINE_DEPLOYMENT_KAFKA_TIER": "deployment.backing.kafka.tier", "DFE_ENGINE_DEPLOYMENT_KAFKA_REGION": "deployment.backing.kafka.region", diff --git a/src/dfe_engine/source/flow.py b/src/dfe_engine/source/flow.py index f6b53142..2ed7b0a9 100644 --- a/src/dfe_engine/source/flow.py +++ b/src/dfe_engine/source/flow.py @@ -41,6 +41,11 @@ FETCHER_SERVICE = "dfe-fetcher" LOADER_SERVICE = "dfe-loader" +PER_SOURCE_DOCS = ( + "https://github.com/hyperi-io/dfe-docker/blob/main/docs/profiles.md#per-source-apps" +) +"""Where a Compose operator reads how a checkout runs one container per source.""" + class FlowError(ValueError): """Raised when a source's stages cannot be run as declared.""" @@ -68,6 +73,42 @@ def stage_instance_ceiling(app: AppDescriptor, settings: DFESettings) -> int | N app, DeployTarget(settings.deployment.target), writes_app_config=appconfig.enabled(settings), + reads_instance_index=appconfig.custom_env_dir(settings) is not None, + hosted=_hosted(settings).get(app.service, ()), + ) + + +def _hosted(settings: DFESettings) -> dict[str, frozenset[str]]: + """The instances each per-config app has a container for here, by app.""" + from dfe_engine.settings import parse_hosted_instances + + return parse_hosted_instances(settings.deployment.hosted_instances) + + +def hosting_refusal(app: AppDescriptor, source: str, purpose: str, settings: DFESettings) -> str: + """Why this deployment cannot run *source*'s instance of *app*, empty where it can.""" + from dfe_engine.appmgmt import appconfig + + ceiling = stage_instance_ceiling(app, settings) + hosted = _hosted(settings).get(app.service, frozenset()) + if app.multiplicity is not Multiplicity.PER_CONFIG or ceiling is None: + return "" + if ceiling and source in hosted: + return "" + service = app.service + if app.shared_instance or not appconfig.enabled(settings): + return ( + f"source {source!r} needs {service} to {purpose}, and this deployment renders " + f"no config into its {service} container, so the source would be saved and " + "never run. Set DFE_ENGINE_DEPLOYMENT_APP_CONFIG_DIR to the directory that " + "container mounts" + ) + running = ", ".join(sorted(hosted)) or "none" + return ( + f"source {source!r} needs its own {service} to {purpose}, and this install runs " + f"{service} only for the sources it was published with ({running}). A new one " + f"needs a dfe-docker checkout, where `make apply` starts a {service} per source: " + f"{PER_SOURCE_DOCS}" ) @@ -190,15 +231,9 @@ def _stage_app( f"runs on {settings.transport.direct_protocol}. Sources that need a per-source " f"app (fetcher, transform) need a {settings.transport.bus_provider} deployment" ) - if stage_instance_ceiling(app, settings) == 0: - raise FlowError( - f"source {source!r} needs its own {service} to {purpose}, and this deployment " - f"cannot configure one: {profile or 'compose'} runs a single {service} and it is " - "deployed idle, but nothing renders the engine's config into that container, " - "so the source would be saved and never run. Deploy on Kubernetes, which runs " - f"one {service} per source, or point this deployment's app-config directory at " - "what its containers mount" - ) + refusal = hosting_refusal(app, source, purpose, settings) + if refusal: + raise FlowError(refusal) if not app.carries(transport): raise FlowError( f"source {source!r} is on the {transport} transport, but {service} " diff --git a/src/dfe_engine/source/registry.py b/src/dfe_engine/source/registry.py index 62147369..3543cc85 100644 --- a/src/dfe_engine/source/registry.py +++ b/src/dfe_engine/source/registry.py @@ -1289,6 +1289,59 @@ def _validate_instance_room(self, source: Source) -> None: allowed, reason = instances.additional_instance_allowed(self._crud, app, ceiling) if not allowed: raise SourceValidationError(f"source {source.source!r}: {reason}") + if source.fetcher is not None and source.state == "active": + self._validate_shared_room(source, settings) + + def _validate_shared_room(self, source: Source, settings: Any) -> None: + """A fetched source sharing one container with the others must not clash with them. + + Judged on the overlay a deploy would write, org included, against the + instances already there, so a source the render would leave out is refused + at save rather than stored and never run. + """ + from dfe_engine.appmgmt import appconfig, instances, routing + from dfe_engine.gitcrud.engine import ResourceNotFoundError + from dfe_engine.source.flow import FETCHER_SERVICE + + if self._crud is None: + return + app = instances.instance_of(FETCHER_SERVICE, source.source) + # Only a deployment whose one fetcher reads the engine's render shares it. + if not ( + appconfig.shares_one_container(app.descriptor, settings) and appconfig.enabled(settings) + ): + return + tenancy = _tenancy_reader() if _tenancy_reader is not None else None + org = source.fetcher.org if source.fetcher is not None else None + judged = source + if tenancy is None and org and source.fetcher is not None: + # With no orgs to read, the org is judged by its own name. + vid = source.runtime_version_id() + runtime = source.version(vid) + bare = runtime.model_copy( + update={"fetcher": source.fetcher.model_copy(update={"org": None})} + ) + judged = source.model_copy(update={"versions": {**source.versions, vid: bare}}) + try: + compiled = routing.compile_for( + app.descriptor, + _WithCandidate(self, judged), + settings, + instance=source.source, + tenancy=tenancy, + ) + except routing.RoutingNotApplicableError: + return # an org nobody registered, which the save refuses on its own + if judged is not source: + compiled["org_id"] = org + try: + doc = instances.read_overlay(self._crud, app) + except ResourceNotFoundError: + doc = instances.initial_overlay(app) + routing.apply(app.descriptor, doc, compiled) + refusal = appconfig.shared_refusal(self._crud, app.descriptor, source.source, doc, settings) + if refusal: + raise SourceValidationError(refusal) # ----------------------------------------------------------------- # Change Callbacks @@ -1382,3 +1435,37 @@ def close(self) -> None: """Stop background refresh and cleanup.""" if self._store is not None: self._store.stop() + + +_tenancy_reader: Callable[[], Any] | None = None +"""Reads the orgs a fetched source's org compiles against, set once the engine holds them.""" + + +def read_tenancy_from(reader: Callable[[], Any] | None) -> None: + """Set where a source save reads the orgs from, or stop reading them with None.""" + global _tenancy_reader + _tenancy_reader = reader + + +class _WithCandidate: + """A source reader that answers for one source being saved as it will be stored.""" + + def __init__(self, registry: Any, candidate: Source) -> None: + self._registry = registry + self._candidate = candidate + + def ingest_rules(self) -> Any: + """Return the stored ingest rules.""" + return self._registry.ingest_rules() + + def get_source(self, source_name: str, /) -> Source: + """Return the candidate under its own name, any other as stored.""" + if source_name == self._candidate.source: + return self._candidate + return self._registry.get_source(source_name) + + def get_all_sources( + self, enabled_only: bool = False, *, states: Collection[str] | None = None + ) -> list[Source]: + """Return the stored sources.""" + return self._registry.get_all_sources(enabled_only, states=states) diff --git a/src/dfe_engine/state_machines/setup.py b/src/dfe_engine/state_machines/setup.py index 0a8e5e73..9f134dc6 100644 --- a/src/dfe_engine/state_machines/setup.py +++ b/src/dfe_engine/state_machines/setup.py @@ -192,6 +192,7 @@ def from_app_state(cls, state: Any) -> Self: namespace=getattr(deployment, "namespace", "") or "", secret_name=getattr(local, "admin_secret_name", "") or "", secret_key=getattr(local, "admin_secret_key", "") or "", + deployer_command=getattr(deployment, "credential_command", "") or "", ), default_engine=default_engine, default_ttl_days=default_ttl_days, diff --git a/tests/e2e/flows/shapes.py b/tests/e2e/flows/shapes.py index 73f805c8..7b0467b9 100644 --- a/tests/e2e/flows/shapes.py +++ b/tests/e2e/flows/shapes.py @@ -225,7 +225,7 @@ def refusal( if needed is not None and needed not in offered: return f"does not deploy {needed}" if not applies_routing and self.per_config_apps(per_config): - return "cannot configure one" + return "would be saved and never run" return self.refused.get(transport) diff --git a/tests/unit/test_api/test_source_apps_reconcile.py b/tests/unit/test_api/test_source_apps_reconcile.py index 2395b97b..68c87634 100644 --- a/tests/unit/test_api/test_source_apps_reconcile.py +++ b/tests/unit/test_api/test_source_apps_reconcile.py @@ -14,6 +14,8 @@ from __future__ import annotations +import pytest + from dfe_engine.gitcrud import GitCrud, default_registry from dfe_engine.gitops.repo import GitopsRepo from dfe_engine.governance import PolicyStore @@ -244,6 +246,180 @@ def test_a_viewer_cannot_reconcile(self, client, app, viewer_headers, tmp_path): assert resp.status_code == 403 +class TestOneFetcherRunsEveryFetchedSourceOnCompose: + """A published Compose install runs one fetcher, so a source it cannot take is refused at save.""" + + def _published(self, app, tmp_path, monkeypatch, *, hosted: str = "") -> None: + """Move the deployment onto a published install: Compose, rendered config, no index.""" + from dfe_engine.settings import get_settings + + # The route reads the app's settings and the save gate the process's. + found = (app.state.settings.deployment, get_settings().deployment) + for deployment in {id(d): d for d in found}.values(): + monkeypatch.setattr(deployment, "target", "docker") + monkeypatch.setattr(deployment, "app_config_dir", str(tmp_path / "app-config")) + monkeypatch.setattr(deployment, "app_env_dir", "") + monkeypatch.setattr(deployment, "hosted_instances", hosted) + + def _running(self, client, admin_headers, name: str, **fetcher) -> None: + from dfe_engine.api.deps import _registries + + body = _fetcher_body(name, **{k: v for k, v in fetcher.items() if v is not None}) + created = client.post("/api/v1/sources", json=body, headers=admin_headers) + assert created.status_code == 201, created.text + _registries["source"].set_deployed_version(name, "1.0.0") + assert ( + client.post("/api/v1/sources/reconcile-apps", headers=admin_headers).status_code == 200 + ) + + def _polled(self, tmp_path) -> dict: + from dfe_engine.yaml_utils import yaml_load + + rendered = yaml_load(tmp_path / "app-config" / FETCHER / "fetcher.yaml") + return rendered["sources"] + + def test_a_clashing_source_is_refused_and_the_running_one_keeps_running( + self, client, app, admin_headers, tmp_path, monkeypatch + ): + self._save_gate(app, tmp_path) + self._published(app, tmp_path, monkeypatch) + self._running(client, admin_headers, "crates-b") + + # crates_io is one flat stanza per process, and this name sorts first. + refused = client.post( + "/api/v1/sources", json=_fetcher_body("crates-a"), headers=admin_headers + ) + + assert refused.status_code in (400, 422), refused.text + assert "cannot share the one dfe-fetcher this deployment runs" in refused.text + assert "config.sources.crates_io" in refused.text + assert self._polled(tmp_path)["crates_io"]["topic"] == "crates-b" + + def _save_gate(self, app, tmp_path): + """A deploy repo, and a registry that runs the source save gate with the engine's orgs.""" + from dfe_engine.api.deps import _registries + from dfe_engine.source.registry import SourceRegistry + + gc = _wire(app, tmp_path) + _registries["source"] = SourceRegistry(crud=gc) + return gc + + @pytest.mark.parametrize( + ("running_org", "newcomer_org", "reason"), + [ + # One org on both: the second crates_io stanza is the clash. + ("acme", "acme", "its config.sources.crates_io.topic is already set"), + # One process stamps one org on everything it emits. + (None, "acme", "its config.org_id differs from crates-b's"), + ], + ) + def test_a_source_naming_an_org_is_judged_with_that_org( + self, client, app, admin_headers, tmp_path, monkeypatch, running_org, newcomer_org, reason + ): + self._save_gate(app, tmp_path) + app.state.org_registry.create("acme", display_name="Acme") + self._published(app, tmp_path, monkeypatch) + self._running(client, admin_headers, "crates-b", org=running_org) + + refused = client.post( + "/api/v1/sources", + json=_fetcher_body("crates-a", org=newcomer_org), + headers=admin_headers, + ) + + assert refused.status_code in (400, 422), refused.text + assert reason in refused.text + assert self._polled(tmp_path)["crates_io"]["topic"] == "crates-b" + + def test_with_no_orgs_to_read_a_source_is_judged_by_its_org_name( + self, client, app, admin_headers, tmp_path, monkeypatch + ): + self._save_gate(app, tmp_path) + app.state.org_registry.create("acme", display_name="Acme") + self._published(app, tmp_path, monkeypatch) + self._running(client, admin_headers, "crates-b") + monkeypatch.setattr("dfe_engine.source.registry._tenancy_reader", None) + + refused = client.post( + "/api/v1/sources", json=_fetcher_body("crates-a", org="acme"), headers=admin_headers + ) + + assert refused.status_code in (400, 422), refused.text + assert "its config.org_id differs from crates-b's" in refused.text + assert "holds no orgs" not in refused.text + + @pytest.mark.parametrize("edited", ["crates-b", "pypi-a"]) + def test_an_update_the_one_fetcher_cannot_take_is_refused_and_both_keep_running( + self, client, app, admin_headers, tmp_path, monkeypatch, edited + ): + self._save_gate(app, tmp_path) + self._published(app, tmp_path, monkeypatch) + self._running(client, admin_headers, "crates-b") + self._running( + client, + admin_headers, + "pypi-a", + source_type="pypi", + config={"packages": ["scalo"]}, + ) + + resp = client.put( + f"/api/v1/apps/{FETCHER}/{edited}/config", + json={"changes": {"extraEnv.ONLY_ME": "1"}}, + headers=admin_headers, + ) + + assert resp.status_code == 409, resp.text + assert resp.json()["code"] == "shared_instance_conflict" + assert "extraEnv.ONLY_ME" in resp.json()["detail"] + polled = self._polled(tmp_path) + assert polled["crates_io"]["topic"] == "crates-b" + assert polled["pypi"]["topic"] == "pypi-a" + + def test_a_generic_helm_write_that_clashes_is_refused( + self, client, app, admin_headers, tmp_path, monkeypatch + ): + gc = self._save_gate(app, tmp_path) + self._published(app, tmp_path, monkeypatch) + self._running(client, admin_headers, "crates-b") + + resp = client.put( + f"/api/v1/helm/files/{FETCHER}-crates-a-values/vars/config.sources.crates_io.topic", + json={"value": "crates-a"}, + headers=admin_headers, + ) + + assert resp.status_code == 409, resp.text + assert resp.json()["code"] == "shared_instance_conflict" + assert f"{FETCHER}-crates-a-values" not in gc.list("helmvars") + assert self._polled(tmp_path)["crates_io"]["topic"] == "crates-b" + + def test_a_hand_made_instance_outside_the_hosted_set_is_refused( + self, client, app, admin_headers, tmp_path, monkeypatch + ): + from dfe_engine.api.deps import _registries + + _wire(app, tmp_path) + # Saved before the move, so the route's own check is what answers. + for name in ("edge", "other"): + body = {**_receiver_body(name), "transform": {"engine": "vrl"}} + assert ( + client.post("/api/v1/sources", json=body, headers=admin_headers).status_code == 201 + ) + _registries["source"].set_deployed_version(name, "1.0.0") + self._published(app, tmp_path, monkeypatch, hosted="dfe-transform-vrl=edge") + + other = client.post( + "/api/v1/apps/dfe-transform-vrl/instances", + json={"instance": "other", "values": {}}, + headers=admin_headers, + ) + + assert other.status_code == 409, other.text + assert other.json()["code"] == "not_hosted" + assert "only for the sources it was published with (edge)" in other.json()["detail"] + + class TestTheReconcileReportsTheRestartsItsOwnWritesNeed: """Where the engine renders the app config, the write that rendered it reports. @@ -258,10 +434,11 @@ class TestTheReconcileReportsTheRestartsItsOwnWritesNeed: RECREATE = f"recreate required: make apply SERVICES={TRANSFORM}-elreg" def _compose(self, app, tmp_path): - """A deploy repo plus the app-config directory a Compose stack mounts.""" + """A deploy repo, the app-config directory a Compose stack mounts, and the env directory its make reads the index from.""" gc = _wire(app, tmp_path) app.state.settings.deployment.target = "docker" app.state.settings.deployment.app_config_dir = str(tmp_path / "app-config") + app.state.settings.deployment.app_env_dir = str(tmp_path / "app-env") return gc def _deployed_transform_source(self, client, admin_headers, name: str) -> None: diff --git a/tests/unit/test_appmgmt/test_appconfig.py b/tests/unit/test_appmgmt/test_appconfig.py index 68ad3983..d9d74e19 100644 --- a/tests/unit/test_appmgmt/test_appconfig.py +++ b/tests/unit/test_appmgmt/test_appconfig.py @@ -565,13 +565,13 @@ def test_the_deployer_is_told_which_containers_to_declare(self, crud, tmp_path): assert _index(settings, VRL).read_text() == "crowdstrike-eu\ncrowdstrike-us\n" def test_an_app_with_no_source_declares_no_container(self, crud, tmp_path): - # A deployment that never adds a fetcher source starts with none running, + # A deployment that never adds a transformed source starts with none running, # rather than with one container holding an empty config. settings = _settings(tmp_path) appconfig.render(crud, settings) - assert _index(settings, "dfe-fetcher").read_text() == "" + assert _index(settings, VRL).read_text() == "" def test_a_deleted_source_takes_its_container_and_its_config_with_it(self, crud, tmp_path): settings = _settings(tmp_path) @@ -636,6 +636,247 @@ def test_a_stack_wide_app_keeps_its_one_directory(self, crud, tmp_path): assert not _index(settings, LOADER).exists() +FETCHER = "dfe-fetcher" + + +def _fetch(crud, instance: str, **values): + """One fetcher instance as the compiler writes it, its stanza keyed under its own name.""" + stanza = {f"config__sources__rest__{instance}__topic": instance, **values} + return _deploy(crud, FETCHER, instance, **stanza) + + +class TestOneContainerForEverySource: + """The fetcher declares shared_instance, so Compose runs one for all of its sources.""" + + def test_two_instances_merge_into_the_one_file_and_no_index(self, crud, tmp_path): + settings = _settings(tmp_path) + _fetch(crud, "github-releases") + _fetch(crud, "wikipedia") + + rendered = [r for r in appconfig.render(crud, settings) if r.service == FETCHER] + + config = _rendered(settings, FETCHER) + assert config["sources"]["rest"] == { + "github-releases": {"topic": "github-releases"}, + "wikipedia": {"topic": "wikipedia"}, + } + # Bound to the app rather than to either source, so the cursor keys stay put. + assert config["instance_id"] == FETCHER + assert [r.container for r in rendered] == [FETCHER] + assert not _index(settings, FETCHER).exists() + assert not _app_dir(settings, FETCHER, "wikipedia").exists() + + def test_a_source_added_to_a_running_fetcher_needs_no_restart(self, crud, tmp_path): + # The fetcher's reload starts an added source, so the published install, + # which has no make to run, is handed no command. + settings = _settings(tmp_path) + _fetch(crud, "github-releases") + appconfig.render(crud, settings) + _fetch(crud, "wikipedia") + + rendered = [r for r in appconfig.render(crud, settings) if r.service == FETCHER] + + assert [r.changed for r in rendered] == [True] + assert appconfig.restart_hints(rendered) == [] + + def test_a_fetcher_with_no_source_still_gets_the_deployment_base(self, crud, tmp_path): + settings = _settings(tmp_path) + _base(settings, FETCHER, "scheduler:\n default_interval_secs: 60\n") + + appconfig.render(crud, settings) + + assert _rendered(settings, FETCHER) == {"scheduler": {"default_interval_secs": 60}} + assert not _index(settings, FETCHER).exists() + + @pytest.mark.parametrize( + ("second", "reason"), + [ + # A flat family is one stanza per process, so a second source of it clashes. + ( + {"config__sources__okta__topic": "okta-b"}, + "its config.sources.okta.topic is already set by another source", + ), + # One process stamps one org on everything it emits. + ({"config__org_id": "acme"}, "its config.org_id differs from okta-a's"), + ], + ) + def test_an_instance_the_one_process_cannot_take_is_named_and_not_run( + self, crud, tmp_path, second, reason + ): + settings = _settings(tmp_path) + _deploy(crud, FETCHER, "okta-a", config__sources__okta__topic="okta-a") + _deploy(crud, FETCHER, "okta-b", **second) + + rendered = appconfig.render(crud, settings) + + assert _rendered(settings, FETCHER)["sources"]["okta"] == {"topic": "okta-a"} + hints = [h for h in appconfig.restart_hints(rendered) if h.startswith("not run:")] + assert hints == [ + f"not run: source 'okta-b' cannot share the one {FETCHER} this deployment runs: " + f"{reason}" + ] + + def test_a_newcomer_that_clashes_never_evicts_the_running_source(self, crud, tmp_path): + settings = _settings(tmp_path) + _deploy(crud, FETCHER, "okta-b", config__sources__okta__topic="okta-b") + appconfig.render(crud, settings) + # Sorts first, so name order alone would have put it in place of okta-b. + newcomer = _deploy(crud, FETCHER, "okta-a", config__sources__okta__topic="okta-a") + + refusal = appconfig.shared_refusal( + crud, descriptor(FETCHER), "okta-a", instances.read_overlay(crud, newcomer), settings + ) + rendered = appconfig.render(crud, settings) + + reason = "its config.sources.okta.topic is already set by another source" + assert refusal == ( + f"source 'okta-a' cannot share the one {FETCHER} this deployment runs: {reason}" + ) + assert _rendered(settings, FETCHER)["sources"]["okta"] == {"topic": "okta-b"} + assert [h for h in appconfig.restart_hints(rendered) if "okta-a" in h] == [ + f"not run: {refusal}" + ] + + def test_a_render_after_per_instance_fetchers_retires_their_index_dirs_and_env( + self, crud, tmp_path + ): + settings = _settings(tmp_path) + env_dir = Path(settings.deployment.app_env_dir) + env_dir.mkdir(parents=True) + _index(settings, FETCHER).write_text("old-a\nold-b\n", encoding="utf-8") + (env_dir / f"{FETCHER}-old-a{appconfig.CUSTOM_ENV_SUFFIX}").write_text( + "KEY=1\n", encoding="utf-8" + ) + for name in ("old-a", "old-b"): + (_app_dir(settings, FETCHER, name)).mkdir(parents=True) + _fetch(crud, "wikipedia") + + rendered = appconfig.render(crud, settings) + + assert not _index(settings, FETCHER).exists() + assert not (env_dir / f"{FETCHER}-old-a{appconfig.CUSTOM_ENV_SUFFIX}").exists() + assert not _app_dir(settings, FETCHER, "old-a").exists() + assert not _app_dir(settings, FETCHER, "old-b").exists() + assert _rendered(settings, FETCHER)["sources"]["rest"] == { + "wikipedia": {"topic": "wikipedia"} + } + removed = [r.container for r in rendered if r.service == FETCHER and r.removed] + assert removed == [f"{FETCHER}-old-a", f"{FETCHER}-old-b"] + assert appconfig.REMOVE_HINT in appconfig.restart_hints(rendered) + + def test_a_kubernetes_deployment_renders_nothing_for_it(self, crud, tmp_path): + # The chart renders one deployment per overlay there, unchanged. + settings = _settings(tmp_path, target="kubernetes") + _fetch(crud, "github-releases") + _fetch(crud, "wikipedia") + + assert appconfig.render(crud, settings) == [] + assert not appconfig.shares_one_container(descriptor(FETCHER), settings) + assert not (tmp_path / "app-config").exists() + + def test_retiring_the_fetcher_s_instances_leaves_every_transform_s_alone(self, crud, tmp_path): + # A transform runs one container per source, so its index, directories and + # env files are its own and a fetcher's retirement must not reach them. + settings = _settings(tmp_path) + env_dir = Path(settings.deployment.app_env_dir) + env_dir.mkdir(parents=True) + _index(settings, FETCHER).write_text("old-a\n", encoding="utf-8") + _app_dir(settings, FETCHER, "old-a").mkdir(parents=True) + for name in ("cisco-ios", "filebeat"): + app = _deploy(crud, VRL, name) + doc = instances.read_overlay(crud, app) + files.upsert_file(doc, file_set(VRL, "transforms"), f"100_{name}.vrl", ".x = 1\n") + set_path(doc, "extraEnv.STAMP", name) + _put(crud, app, doc) + _fetch(crud, "wikipedia") + + appconfig.render(crud, settings) + + assert not _index(settings, FETCHER).exists() + assert _index(settings, VRL).read_text(encoding="utf-8") == "cisco-ios\nfilebeat\n" + for name in ("cisco-ios", "filebeat"): + program = _app_dir(settings, VRL, name) / "transforms" / f"100_{name}.vrl" + assert program.read_text(encoding="utf-8") == ".x = 1\n" + assert _rendered(settings, VRL, name)["transforms"]["dir"] == ( + f"{MOUNT}/{VRL}/{name}/transforms" + ) + assert _env_file(settings, f"{VRL}-{name}").read_text(encoding="utf-8") == ( + f"STAMP={name}\n" + ) + + +class TestEveryOverlayWriterMeetsTheSharedContainer: + """The deploy repo's own write check, so no writer reaches a commit around it.""" + + @pytest.fixture + def judged(self, tmp_path): + settings = _settings(tmp_path) + appconfig.judge_writes_by(lambda: settings) + yield settings + appconfig.judge_writes_by(None) + + def _running(self, crud, settings) -> None: + _deploy(crud, FETCHER, "crates-b", config__sources__crates_io__topic="crates-b") + appconfig.render(crud, settings) + + def test_a_key_write_that_clashes_is_refused_before_it_commits(self, crud, judged): + self._running(crud, judged) + newcomer = instances.instance_of(FETCHER, "crates-a").overlay_name + + with pytest.raises(appconfig.SharedInstanceConflictError, match="cannot share"): + crud.set_key( + instances.HELMVARS_CLASS, + newcomer, + "config.sources.crates_io.topic", + "crates-a", + ACTOR, + ) + + assert newcomer not in crud.list(instances.HELMVARS_CLASS) + + def test_an_action_that_would_make_an_overlay_clash_is_refused(self, crud, judged): + # An action may not create a source-bound instance, so it edits a running one. + from dfe_engine.governance import ActionDef, ActionStore, VarChange + from dfe_engine.governance.actions import InvalidDocumentError + + self._running(crud, judged) + other = _deploy(crud, FETCHER, "pypi-a", config__sources__pypi__topic="pypi-a") + appconfig.render(crud, judged) + store = ActionStore(crud) + store.save( + ActionDef( + name="stamp-acme", + description="x", + required_action="action:invoke:stamp-acme", + changes=[ + VarChange( + cls=instances.HELMVARS_CLASS, + name=other.overlay_name, + path="config.org_id", + value="acme", + ) + ], + ), + actor="admin", + ) + + with pytest.raises(InvalidDocumentError, match="cannot share"): + store.invoke("stamp-acme", actor="bob") + + polled = _rendered(judged, FETCHER)["sources"] + assert polled["crates_io"] == {"topic": "crates-b"} + assert polled["pypi"] == {"topic": "pypi-a"} + assert "org_id" not in instances.read_overlay(crud, other).get("config", {}) + + def test_a_delete_of_a_running_source_still_goes_through(self, crud, judged): + self._running(crud, judged) + running = instances.instance_of(FETCHER, "crates-b").overlay_name + + crud.delete(instances.HELMVARS_CLASS, running, ACTOR, message="test: rm") + + assert running not in crud.list(instances.HELMVARS_CLASS) + + class TestCustomEnvironment: def test_the_overlay_block_becomes_one_file_per_compose_service(self, crud, tmp_path): settings = _settings(tmp_path) @@ -999,8 +1240,24 @@ def test_a_loader_the_deployment_pins_is_restarted_for_a_change(self, crud, tmp_ assert _rendered(settings, LOADER)["hot_reload"] == {"enabled": False} assert rendered[LOADER].restart_hint == f"restart required: make apply SERVICES={LOADER}" - def test_a_startup_bound_app_names_the_command_that_applies_it(self, crud, tmp_path): + @pytest.mark.parametrize( + ("apply_command", "expected"), + [ + # A checkout applies it through make. + ("", f"restart required: make apply SERVICES={VRL}-filebeat"), + # A published install has no Makefile, so it names its own command. + ( + "docker compose -f oci://example/dfe:compose up -d --force-recreate {service}", + "restart required: docker compose -f oci://example/dfe:compose up -d " + f"--force-recreate {VRL}-filebeat", + ), + ], + ) + def test_a_startup_bound_app_names_the_command_that_applies_it( + self, crud, tmp_path, apply_command, expected + ): settings = _settings(tmp_path) + settings.deployment.apply_command = apply_command app = _deploy(crud, VRL, "filebeat") appconfig.render(crud, settings) doc = instances.read_overlay(crud, app) @@ -1012,9 +1269,7 @@ def test_a_startup_bound_app_names_the_command_that_applies_it(self, crud, tmp_p assert rendered[VRL].restart_required # The container carrying this instance, not the app: a per-config app has # one per source, so the app's own name would restart the wrong one. - assert rendered[VRL].restart_hint == ( - f"restart required: make apply SERVICES={VRL}-filebeat" - ) + assert rendered[VRL].restart_hint == expected assert appconfig.restart_hints(list(rendered.values())) == [rendered[VRL].restart_hint] def test_the_first_render_of_a_stack_asks_for_no_restart(self, crud, tmp_path): diff --git a/tests/unit/test_appmgmt/test_catalogue.py b/tests/unit/test_appmgmt/test_catalogue.py index 1bba57d9..6fba3ac6 100644 --- a/tests/unit/test_appmgmt/test_catalogue.py +++ b/tests/unit/test_appmgmt/test_catalogue.py @@ -153,19 +153,18 @@ def test_the_receiver_names_the_destinations_it_builds_at_startup(self): assert "config.destinations" in receiver.restart_paths assert "config.routing" not in receiver.restart_paths - def test_every_block_the_engine_writes_into_a_fetcher_reports_a_restart_but_its_org(self): - # A fetcher reload adds no source and rebuilds no output, so a Compose - # operator told nothing would keep running the old source. The org it - # stamps is re-read every fetch cycle, so it needs no restart. + def test_a_fetcher_reloads_its_sources_and_org_and_restarts_for_its_output(self): + # A reload starts, stops and rebuilds sources and re-reads the org each + # fetch cycle. The output's transport connections are built at startup. fetcher = catalogue.descriptor("dfe-fetcher") written = set(fetcher.routing_paths.values()) - stamped_org = {"config.org_id", "config.default_org_id"} + reloaded = {"config.sources", "config.org_id", "config.default_org_id"} assert fetcher.hot_reload is True - assert stamped_org <= written - assert written - stamped_org <= set(fetcher.restart_paths) - assert not stamped_org & set(fetcher.restart_paths) + assert reloaded <= written + assert written - reloaded <= set(fetcher.restart_paths) + assert not reloaded & set(fetcher.restart_paths) def test_vector_reloads_its_transform_files_in_place(self): # It SIGHUPs Vector when only the transform files changed. The enrichment @@ -503,6 +502,28 @@ def test_a_reload_setting_outside_the_config_block_is_refused(self, tmp_path): with pytest.raises(CatalogueError, match=r"must be a config\. overlay path"): load_catalogue(path) + @pytest.mark.parametrize( + ("app", "refusal"), + [ + ({"shared_instance": "config.sources"}, "needs multiplicity per_config"), + ( + {"multiplicity": "per_config", "shared_instance": "sources"}, + r"must be a config\. overlay path", + ), + ], + ) + def test_a_shared_instance_that_cannot_be_merged_is_refused(self, tmp_path, app, refusal): + with pytest.raises(CatalogueError, match=refusal): + load_catalogue(self._manifest(tmp_path, app)) + + def test_only_the_fetcher_ships_a_shared_instance(self): + # A transform runs one source per process, so it declares none. + shared = { + s: a.shared_instance for s, a in catalogue.APP_CATALOGUE.items() if a.shared_instance + } + + assert shared == {"dfe-fetcher": "config.sources"} + def test_an_idle_when_that_is_not_a_list_is_refused(self, tmp_path): with pytest.raises(CatalogueError, match="idle_when must be a list"): load_catalogue(self._manifest(tmp_path, {"idle_when": "config.sources"})) diff --git a/tests/unit/test_appmgmt/test_scaling.py b/tests/unit/test_appmgmt/test_scaling.py index a15187a4..00838234 100644 --- a/tests/unit/test_appmgmt/test_scaling.py +++ b/tests/unit/test_appmgmt/test_scaling.py @@ -36,6 +36,40 @@ def test_a_docker_target_that_renders_its_app_config_is_unbounded(): assert scaling.instance_ceiling(VRL, DeployTarget.DOCKER, writes_app_config=True) is None +FETCHER = catalogue.descriptor("dfe-fetcher") + + +@pytest.mark.parametrize( + ("app", "reads_index", "hosted", "expected"), + [ + # One container serves every fetcher source, so sources are not containers. + (FETCHER, False, (), None), + # A checkout's make reads the index and declares a container per name. + (VRL, True, (), None), + # A published install runs only the instances it declared. + (VRL, False, ("intake", "cisco-asa"), 2), + (VRL, False, (), 0), + ], +) +def test_a_docker_target_that_renders_caps_by_what_runs_the_instances( + app, reads_index, hosted, expected +): + assert ( + scaling.instance_ceiling( + app, + DeployTarget.DOCKER, + writes_app_config=True, + reads_instance_index=reads_index, + hosted=hosted, + ) + == expected + ) + + +def test_a_shared_instance_app_still_needs_its_config_rendered(): + assert scaling.instance_ceiling(FETCHER, DeployTarget.DOCKER, reads_instance_index=False) == 0 + + def test_a_single_deployment_app_is_one_everywhere(): for target in DeployTarget: assert scaling.instance_ceiling(LOADER, target) == 1 diff --git a/tests/unit/test_auth/test_breakglass_first_login.py b/tests/unit/test_auth/test_breakglass_first_login.py index b348d98b..7ad01c54 100644 --- a/tests/unit/test_auth/test_breakglass_first_login.py +++ b/tests/unit/test_auth/test_breakglass_first_login.py @@ -806,6 +806,19 @@ def test_docker_and_local_commands(self): assert credential_fetch_command(DOCKER) == "make creds" assert "DFE_ENGINE_AUTH_LOCAL_ADMIN_PASSWORD" in credential_fetch_command(LOCAL) + @pytest.mark.parametrize( + ("deployer_command", "expected"), + [ + # A published compose install has no checkout, so it names its own way. + ("log in with the first-login password", "log in with the first-login password"), + # A checkout names none, and blank is the same as unset. + ("", "make creds"), + (" ", "make creds"), + ], + ) + def test_a_docker_deployer_s_own_command_wins(self, deployer_command, expected): + assert credential_fetch_command(DOCKER, deployer_command=deployer_command) == expected + def test_rotation_command_is_a_store_operation(self): assert "patch secret" in rotation_store_command( KUBERNETES, namespace="dfe", secret_name="dfe-engine", secret_key="admin-password" diff --git a/tests/unit/test_e2e_harness/test_flow_fixtures.py b/tests/unit/test_e2e_harness/test_flow_fixtures.py index 8cee2e5b..86ac36fa 100644 --- a/tests/unit/test_e2e_harness/test_flow_fixtures.py +++ b/tests/unit/test_e2e_harness/test_flow_fixtures.py @@ -290,7 +290,7 @@ def test_every_shape_needing_one_is_refused(self) -> None: declared = shape.refusal( transport, (transport,), EVERY_APP, PER_CONFIG, applies_routing=False ) - assert declared == "cannot configure one", f"{shape.name}: {declared}" + assert declared == "would be saved and never run", f"{shape.name}: {declared}" def test_a_shape_needing_none_is_untouched(self) -> None: for shape in shapes.load_shapes(): diff --git a/tests/unit/test_query/test_client.py b/tests/unit/test_query/test_client.py index bd6eabbc..cc21f76f 100644 --- a/tests/unit/test_query/test_client.py +++ b/tests/unit/test_query/test_client.py @@ -1,8 +1,14 @@ """Unit tests for QueryClient.""" +import http.server +import json +import threading +from typing import ClassVar from unittest.mock import MagicMock import pytest +from scalo.cli.openapi import Credential +from scalo.logger import logger from dfe_engine.query.client import QueryClient from dfe_engine.query.models import QueryMetadata @@ -137,3 +143,82 @@ def test_direct_mode_requires_view_executor(self, monkeypatch): monkeypatch.setattr(client, "_get_view_executor", lambda: None) with pytest.raises(RuntimeError, match="ViewExecutor not available"): client.query("analytics/user_activity") + + +class _RecordingHandler(http.server.BaseHTTPRequestHandler): + """Answers one view execute and keeps the headers it arrived with.""" + + seen: ClassVar[list[dict[str, str]]] = [] + + def do_POST(self) -> None: + self.rfile.read(int(self.headers.get("Content-Length", 0))) + type(self).seen.append(dict(self.headers.items())) + body = json.dumps({"rows": [], "columns": [], "row_count": 0}).encode() + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, format: str, *args: object) -> None: + """Keep the listener off the test output.""" + + +@pytest.fixture +def engine_url(): + """A real HTTP listener on loopback standing in for the engine's view route.""" + _RecordingHandler.seen = [] + server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), _RecordingHandler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + yield f"http://127.0.0.1:{server.server_address[1]}" + server.shutdown() + server.server_close() + + +class TestQueryClientAuthentication: + @pytest.mark.parametrize( + ("credential", "header", "value"), + [ + (Credential(account="ops", token="tok-1"), "Authorization", "Bearer tok-1"), + (Credential(account="svc", api_key="key-1"), "X-API-Key", "key-1"), + ], + ) + def test_http_mode_sends_the_credential_as_the_cli_does( + self, engine_url, credential, header, value + ): + with QueryClient(base_url=engine_url, credential=credential) as client: + client.query("analytics/user_activity") + + assert _RecordingHandler.seen[0][header] == value + + def test_http_mode_without_a_credential_sends_no_auth_header(self, engine_url): + with QueryClient(base_url=engine_url) as client: + client.query("analytics/user_activity") + + sent = {name.lower() for name in _RecordingHandler.seen[0]} + assert "authorization" not in sent + assert "x-api-key" not in sent + + +class TestQueryClientRefusesAnUnsendableCredential: + @pytest.mark.parametrize( + "credential", + [ + Credential(account="ops", token="tok-secret-1\n"), + Credential(account="ops", token="tok-\rsecret-1"), + Credential(account="svc", api_key="key-secret-1\x00"), + ], + ) + def test_the_error_and_the_log_name_the_fault_and_not_the_value(self, credential): + lines: list[str] = [] + sink = logger.add(lines.append, level="DEBUG") + try: + with pytest.raises(ValueError, match="control character") as refused: + QueryClient(base_url="http://127.0.0.1:9", credential=credential) + finally: + logger.remove(sink) + + assert "secret" not in str(refused.value) + assert refused.value.__cause__ is None + assert not [line for line in lines if "secret" in line] diff --git a/tests/unit/test_query/test_models.py b/tests/unit/test_query/test_models.py index 235ada77..232655e9 100644 --- a/tests/unit/test_query/test_models.py +++ b/tests/unit/test_query/test_models.py @@ -28,7 +28,13 @@ def test_defaults(self): assert options.after_tiebreak is None assert options.timeout_seconds is None assert options.cache is True - assert options.store is None + + def test_a_store_a_caller_still_sends_is_accepted_and_ignored(self): + # Nothing ever read it: a view always runs in the executor's own database. + options = QueryOptions.model_validate({"store": "events_db", "limit": 5}) + + assert options.limit == 5 + assert "store" not in options.model_dump() def test_limit_validation(self): """Test limit bounds validation.""" diff --git a/tests/unit/test_settings.py b/tests/unit/test_settings.py index ad3790ee..00f97a14 100644 --- a/tests/unit/test_settings.py +++ b/tests/unit/test_settings.py @@ -13,6 +13,7 @@ DFESettings, HuntsSettings, load_settings, + parse_hosted_instances, reset_settings, ) from dfe_engine.transport import SourceTransport @@ -820,6 +821,47 @@ def _config_file(tmp_path, path: str, value: list) -> str: return str(target) +class TestFixedContainerDeployment: + """What a deployer with no checkout tells the engine about the containers it runs.""" + + @pytest.mark.parametrize( + ("raw", "expected"), + [ + ("", {}), + ( + "dfe-transform-elastic=demo-syslog, cisco-asa;dfe-transform-vrl=", + { + "dfe-transform-elastic": frozenset({"demo-syslog", "cisco-asa"}), + "dfe-transform-vrl": frozenset(), + }, + ), + ], + ) + def test_hosted_instances_are_read_per_app(self, raw, expected): + assert parse_hosted_instances(raw) == expected + + @pytest.mark.parametrize( + ("deployment", "refusal"), + [ + ({"hosted_instances": "demo-syslog,cisco-asa"}, "is not ="), + ({"hosted_instances": "a=x;a=y"}, "name a twice"), + ({"apply_command": "docker compose restart"}, r"must carry \{service\}"), + # An escaped placeholder formats to a literal {service}. + ({"apply_command": "docker compose up -d {{service}}"}, r"must carry \{service\}"), + ({"apply_command": "up {service} {project}"}, "no other field"), + ({"apply_command": "up {service}" * 2}, "once"), + ({"apply_command": "up {service"}, "is not a template"), + ({"apply_command": "up {service} }"}, "is not a template"), + ({"apply_command": "up {service:{other}}"}, "no conversion or format spec"), + ({"apply_command": "up {service!x}"}, "no conversion or format spec"), + ({"apply_command": "up {service:>9}"}, "no conversion or format spec"), + ], + ) + def test_a_declaration_the_engine_cannot_act_on_stops_startup(self, deployment, refusal): + with pytest.raises(ValidationError, match=refusal): + DFESettings(env="dev", deployment=deployment) + + class TestListsAcrossLayers: """A list is the whole value its layer sets: the layer above replaces it, never extends it.""" diff --git a/tests/unit/test_source/test_flow.py b/tests/unit/test_source/test_flow.py index 9d0ee7a2..1333841f 100644 --- a/tests/unit/test_source/test_flow.py +++ b/tests/unit/test_source/test_flow.py @@ -28,13 +28,25 @@ def _settings( - *, profile: str = "", target: str = "unknown", app_config_dir: str = "", **transport + *, + profile: str = "", + target: str = "unknown", + app_config_dir: str = "", + app_env_dir: str = "", + hosted_instances: str = "", + **transport, ) -> DFESettings: """Settings differing from the defaults only in the transport block and the deployment.""" return DFESettings( env="dev", transport=transport, - deployment={"profile": profile, "target": target, "app_config_dir": app_config_dir}, + deployment={ + "profile": profile, + "target": target, + "app_config_dir": app_config_dir, + "app_env_dir": app_env_dir, + "hosted_instances": hosted_instances, + }, ) @@ -373,16 +385,23 @@ def test_the_rule_is_manifest_data_rather_than_a_branch_per_app(self): class TestATargetThatRunsOneOfEachApp: - """A per-config stage needs its own deployment, and Compose holds one of each. + """A per-config stage needs a container to run in, and Compose creates none at run time. - Compose declares its services in a committed file and creates none at run - time, so the resident container is started idle and the first source to need - it fills it. Where nothing renders the engine's config into that container - the source would be saved and never run, so it is refused at save instead. + The fetcher runs every source in one container, so it needs only its config + rendered. A transform runs one container per source, which a checkout's + ``make apply`` declares from the engine's index, and a published install + runs only the instances it was published with. Anything else would be saved + and never run, so it is refused at save instead. """ DOCKER = {"profile": "docker-single", "target": "docker"} WRITING = {**DOCKER, "app_config_dir": "/app/app-config"} + CHECKOUT = {**WRITING, "app_env_dir": "/app/env"} + PUBLISHED_DEMO = { + **WRITING, + "hosted_instances": "dfe-transform-elastic=demo-syslog,cisco-asa,fortinet", + } + ELASTIC = {"engine": "elastic", "variant": "filebeat.cisco_asa.default"} def test_a_kubernetes_tier_runs_one_per_source(self): for name in ("okta", "cloudflare"): @@ -392,39 +411,69 @@ def test_a_kubernetes_tier_runs_one_per_source(self): ) assert flow.input == f"dfe-fetcher-{name}" - def test_a_fetched_source_is_refused_where_nothing_renders_the_config(self): - with pytest.raises(FlowError, match="cannot configure one"): - resolve_flow(_fetched(), _settings(default="bus", **self.DOCKER)) + @pytest.mark.parametrize("deployment", ["WRITING", "CHECKOUT", "PUBLISHED_DEMO"]) + @pytest.mark.parametrize("name", ["okta", "cloudflare"]) + def test_the_one_fetcher_takes_every_fetched_source_where_its_config_is_rendered( + self, deployment, name + ): + settings = _settings(default="bus", **getattr(self, deployment)) + + flow = resolve_flow(_fetched(name, fetcher={"source_type": name}), settings) - def test_the_refusal_says_it_runs_one_and_names_what_turns_it_on(self): + assert flow.input == f"dfe-fetcher-{name}" + + @pytest.mark.parametrize( + ("source", "service"), + [ + (_fetched(), "dfe-fetcher"), + (_source(transform={"engine": "vrl"}), "dfe-transform-vrl"), + ], + ) + def test_a_source_is_refused_where_nothing_renders_the_config(self, source, service): with pytest.raises(FlowError) as refused: - resolve_flow(_fetched(), _settings(default="bus", **self.DOCKER)) + resolve_flow(source, _settings(default="bus", **self.DOCKER)) said = str(refused.value) - assert "runs a single dfe-fetcher and it is deployed idle" in said - assert "app-config directory" in said + assert f"renders no config into its {service} container" in said + assert "DFE_ENGINE_DEPLOYMENT_APP_CONFIG_DIR" in said - def test_a_transformed_source_is_refused_the_same_way(self): - # The transform is per-config too, so the rule reads the manifest rather - # than naming the fetcher. - with pytest.raises(FlowError, match="cannot configure one"): - resolve_flow( - _source(transform={"engine": "vrl"}), _settings(default="bus", **self.DOCKER) - ) + def test_a_checkout_runs_a_transform_for_any_source(self): + flow = resolve_flow( + _source("anything", transform=self.ELASTIC), _settings(default="bus", **self.CHECKOUT) + ) - def test_a_docker_target_that_renders_its_app_config_takes_the_first_source(self): - # The cap is one, not zero: the resident container now gets a config, so - # the source it is given runs. The SECOND is refused against the deploy - # repo, which the flow resolver never reads. - flow = resolve_flow(_fetched(), _settings(default="bus", **self.WRITING)) - assert flow.input == "dfe-fetcher-okta" + assert flow.transform is not None + assert flow.transform.app == "dfe-transform-elastic" - def test_a_transformed_source_is_taken_the_same_way(self): + @pytest.mark.parametrize("name", ["demo-syslog", "cisco-asa", "fortinet"]) + def test_a_published_install_takes_the_sources_it_was_published_with(self, name): flow = resolve_flow( - _source(transform={"engine": "vrl"}), _settings(default="bus", **self.WRITING) + _source(name, transform=self.ELASTIC), _settings(default="bus", **self.PUBLISHED_DEMO) ) + assert flow.transform is not None - assert flow.transform.app == "dfe-transform-vrl" + assert flow.transform.instance == f"dfe-transform-elastic-{name}" + + @pytest.mark.parametrize( + ("deployment", "running"), + [ + ("PUBLISHED_DEMO", "(cisco-asa, demo-syslog, fortinet)"), + # The stack file declares none. + ("WRITING", "(none)"), + ], + ) + def test_a_new_transform_source_on_a_published_install_is_refused(self, deployment, running): + with pytest.raises(FlowError) as refused: + resolve_flow( + _source("paloalto", transform=self.ELASTIC), + _settings(default="bus", **getattr(self, deployment)), + ) + + said = str(refused.value) + assert f"dfe-transform-elastic only for the sources it was published with {running}" in said + assert "dfe-docker checkout" in said + assert "docs/profiles.md#per-source-apps" in said + assert "Kubernetes" not in said def test_a_source_needing_only_stack_wide_apps_still_saves(self): # The receiver and the loader are one deployment each, so a plain posted