Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion docs/data-plane/source-flow.md
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
12 changes: 0 additions & 12 deletions openapi/openapi.json
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down
13 changes: 13 additions & 0 deletions src/dfe_engine/api/app.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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")


Expand Down
6 changes: 6 additions & 0 deletions src/dfe_engine/api/errors.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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")
Expand Down
9 changes: 7 additions & 2 deletions src/dfe_engine/api/v1/apps.py
Original file line number Diff line number Diff line change
Expand Up @@ -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])
Expand Down Expand Up @@ -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")
Expand Down
3 changes: 2 additions & 1 deletion src/dfe_engine/api/v1/sources.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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:
Expand Down
Loading
Loading