diff --git a/.github/labeler.yml b/.github/labeler.yml index 3e49d1f69..b52635ccf 100644 --- a/.github/labeler.yml +++ b/.github/labeler.yml @@ -35,10 +35,7 @@ - backend/app/services/result_processor/** - backend/app/runtime_registry.py - backend/app/scripts/** - - backend/workers/run_k8s_worker.py - - backend/workers/run_pod_monitor.py - - backend/workers/run_result_processor.py - - backend/workers/run_saga_orchestrator.py + - backend/app/worker.py "area: events": - changed-files: @@ -50,8 +47,6 @@ - backend/app/services/idempotency/** - backend/app/services/event_replay/** - backend/app/services/event_service.py - - backend/workers/run_dlq_processor.py - - backend/workers/run_event_replay.py "area: realtime": - changed-files: diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 11dd31e65..fcac34b3c 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -566,9 +566,7 @@ jobs: run: | mkdir -p logs docker compose logs --timestamps > logs/docker-compose.log 2>&1 - for svc in backend mongo redis kafka \ - k8s-worker pod-monitor result-processor \ - saga-orchestrator event-replay dlq-processor; do + for svc in backend worker mongo redis kafka; do docker compose logs --timestamps "$svc" > "logs/$svc.log" 2>&1 || true done kubectl get events --sort-by='.metadata.creationTimestamp' -A > logs/k8s-events.log 2>&1 || true diff --git a/AGENTS.md b/AGENTS.md index 71661b8aa..41da2c96f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -1,6 +1,6 @@ # Integr8sCode: Agent Instructions -Platform for running scripts online in isolated Kubernetes pods. Users write Python, Node.js, Ruby, Bash or Go in a Svelte 5 frontend, the FastAPI backend publishes the request to Kafka, and workers run it in its own pod. +Platform for running scripts online in isolated Kubernetes pods. Users write Python, Node.js, Ruby, Bash or Go in a Svelte 5 frontend, the FastAPI backend publishes the request to Kafka, and the worker runs it in its own pod. ## Tech Stack @@ -9,7 +9,7 @@ Platform for running scripts online in isolated Kubernetes pods. Users write Pyt | Frontend | Svelte 5 (runes), SvelteKit 3 (adapter-static SPA) on Vite 8, TanStack Query 6, TanStack Table 9, bits-ui 2, superforms 3 with formsnap, mode-watcher, TypeScript 6 strict, TailwindCSS 4, Node 26 | | Backend | FastAPI, Dishka DI, Beanie ODM (MongoDB), Python 3.14 (backend/.python-version), async throughout, httpx2 for HTTP | | Events | Kafka (KRaft, no ZooKeeper) + FastStream (Pydantic JSON) | -| Workers | k8s_worker, pod_monitor, result_processor, saga_orchestrator, dlq_processor, event_replay | +| Worker | one process (`app/worker.py`): k8s worker, pod monitor, result processor, saga orchestrator, DLQ retries | | Infra | MongoDB 9, Redis 8; optional Grafana, Jaeger, Victoria Metrics, Loki, OTel Collector (`--observability`) and Kafdrop (`--debug`) | | Runtimes | Python 3.11 to 3.14, Node 22/24/26, Ruby 3.3/3.4/4.0, Bash 5.3, Go 1.26/1.27 (app/runtime_registry.py) | @@ -30,9 +30,9 @@ backend/ │ ├── services/ # Business services, constructor-injected via Dishka │ ├── scripts/ # entrypoint.sh mounted into executor pods │ ├── main.py # create_app() +│ ├── worker.py # the worker process: Kafka consumers, pod watch, scheduled jobs │ ├── runtime_registry.py # Languages, versions, images, per-language pod settings │ └── settings.py # Settings: TOML plus secrets from env -├── workers/ # run_.py entry points, one per worker ├── scripts/ # check_orphan_modules.py, seed_users.py └── tests/ # unit/, e2e/, contract/, load/ @@ -218,7 +218,7 @@ Rules: - Use `raise SpecificError(...) from original_exc` to preserve exception chains - Never catch `Exception` broadly; catch specific exception types (the two allowed cases follow) -A catch-all is allowed in two cases only. The first records or logs, then re-raises: `UnifiedProducer.produce` in `app/events/core/producer.py` marks the event as failed to publish, `IdempotencyMiddleware` and `DeadLetterMiddleware` in `app/events/middlewares.py` release the idempotency key and save the event to the DLQ, and `ExecutionService` marks the execution as errored before raising `InfrastructureError`. The second is a loop or batch boundary that must survive one bad item: the saga orchestrator's queue loop (`try_schedule_from_queue`), saga step execution and its compensation loop, the pod monitor's watch cycle (`workers/run_pod_monitor.py`) and per-event processing, k8s worker pod creation (publishes a pod-creation-failed event instead), notification delivery and the scheduled-notification batch, DLQ batch retry and per-event replay. A new catch-all needs the same justification and must log or record the error; never swallow it silently. +A catch-all is allowed in two cases only. The first records or logs, then re-raises: `UnifiedProducer.produce` in `app/events/core/producer.py` marks the event as failed to publish, `IdempotencyMiddleware` and `DeadLetterMiddleware` in `app/events/middlewares.py` release the idempotency key and save the event to the DLQ, and `ExecutionService` marks the execution as errored before raising `InfrastructureError`. The second is a loop or batch boundary that must survive one bad item: the saga orchestrator's queue loop (`try_schedule_from_queue`), saga step execution and its compensation loop, the pod monitor's watch cycle (`app/worker.py`) and per-event processing, k8s worker pod creation (publishes a pod-creation-failed event instead), notification delivery and the scheduled-notification batch, DLQ batch retry and per-event replay. A new catch-all needs the same justification and must log or record the error; never swallow it silently. ```python # Repository: returns None for reads; raises only on write constraints @@ -314,10 +314,10 @@ class MyProvider(Provider): ``` - All providers live in one file, `app/core/providers.py` -- Container factories live in `app/core/container.py`, one per process: `create_app_container`, `create_k8s_worker_container`, `create_pod_monitor_container`, `create_result_processor_container`, `create_saga_orchestrator_container`, `create_event_replay_container`, `create_dlq_processor_container` +- Container factories live in `app/core/container.py`, one per process: `create_app_container` and `create_worker_container` - `FromDishka[ServiceType]` injects into route handlers (`dishka.integrations.fastapi`) and Kafka subscribers (`dishka_faststream`) - `@inject` decorator when needed outside routes (e.g. the `admin_user` dependency) -- `setup_dishka(container, broker=broker, auto_inject=True)` from `dishka_faststream` runs **after** subscriber registration and **before** the broker starts, in workers and in the API lifespan +- `setup_dishka(container, broker=broker, auto_inject=True)` from `dishka_faststream` runs **after** subscriber registration and **before** the broker starts, in the worker and in the API lifespan - Never instantiate services directly; always use DI ## Backend: Domain Models (Stdlib Dataclasses) @@ -475,13 +475,13 @@ settings: FromDishka[Settings] Config load order (each overrides the previous): 1. `config.toml`: base settings (committed) -2. `config..toml`: per-worker overrides (optional) +2. `config.worker.toml`: the worker's override (optional) 3. Secrets: `SECRET_KEY`, `MONGO_USER`, `MONGO_PASSWORD`, `REDIS_PASSWORD`. Each one is read from the env var first, then the TOML value, then a dev default, so a missing secret never raises. Production must set all four: the deploy workflow stops when one of their GitHub secrets is empty, since compose would otherwise fall back to the public defaults. `MONGODB_URL` is built from the Mongo values ```python Settings() # config.toml + secrets Settings(config_path="config.test.toml") # tests + secrets -Settings(override_path="config.saga-orchestrator.toml") # worker + secrets +Settings(override_path="config.worker.toml") # worker + secrets ``` `Settings` uses `model_config = ConfigDict(extra="forbid")`, so unknown keys are errors. @@ -533,27 +533,28 @@ def saga_router() -> KafkaRouter: - String literal `group_id` per consumer group - One topic per subscriber, no filtering inside handlers. The one exception is the API's `sse-bridge-pool` subscriber, which takes the 17 execution, result and pod event types in one handler and forwards them to Redis for SSE - Consumers that must not run a handler twice put `idempotent(strategy, ttl_seconds=...)` from `app/events/middlewares.py` on their router -- Consumers whose failures should be retried put `DeadLetterMiddleware` first on their router. It saves the failed event to the DLQ with the consumer group, and the dlq-processor republishes it with headers that `RetryTargetMiddleware` (on every broker) uses to skip it in the other groups. The SSE bridge has none, since a late SSE push is worse than a missed one +- Consumers whose failures should be retried put `DeadLetterMiddleware` first on their router. It saves the failed event to the DLQ with the consumer group, and the worker's DLQ job republishes it with headers that `RetryTargetMiddleware` (on every broker) uses to skip it in the other groups. The SSE bridge has none, since a late SSE push is worse than a missed one - Don't wrap handlers to count failures: FastStream's `KafkaTelemetryMiddleware` already records them under `error_type` on `messaging_process_duration_seconds` -### Worker Pattern -Each `backend/workers/run_.py` follows the same order: +### Worker Process +Everything outside the API runs in one process, `python -m app.worker` (`backend/app/worker.py`), in this order: -1. `Settings(override_path="config..toml")` +1. `Settings(override_path="config.worker.toml")` 2. `init_beanie()` on a tz-aware `AsyncMongoClient`, before the container exists -3. `create__container(settings)` -4. Get `KafkaBroker` from the container and `broker.include_router(_router())` if the worker consumes +3. `create_worker_container(settings)` +4. Get `KafkaBroker` from the container and `broker.include_routers(k8s_worker_router(), result_processor_router(), saga_router())` 5. `setup_dishka(container, broker=broker, auto_inject=True)` 6. `FastStream(broker, on_startup=[...], on_shutdown=[...]).run()`; APScheduler jobs start in `on_startup` -| Worker | Purpose | -|--------|---------| -| k8s_worker | Consumes create/delete pod commands, creates executor pods and ConfigMaps, applies the executor NetworkPolicy | +| Part | Purpose | +|------|---------| +| k8s_worker | Consumes create/delete pod commands, creates executor pods and ConfigMaps, applies the executor NetworkPolicy on startup | | pod_monitor | Watches executor pods (watch cycle every 5s) and publishes pod and execution events; no subscribers | | result_processor | Consumes completed, failed and timeout events, stores results, publishes result_stored or result_failed; moves the stored status to scheduled or running on pod_scheduled and pod_running | | saga_orchestrator | Runs execution sagas, schedules the queue every 10s, checks saga timeouts every 30s | | dlq_processor | APScheduler only (no Kafka consumer), retries due DLQ rows every 10s for the consumer group that failed | -| event_replay | APScheduler only, cleans up replay sessions older than 48h every 6h; replays themselves run in the API | + +Each consumer keeps its own `group_id`, so sharing a process doesn't change what Kafka delivers to it. New background work goes into this process: a router in `app/events/handlers.py` included next to the others, or one more APScheduler job. Run one worker replica: the pod watch and the jobs would run once per replica. Replays run in the API, and a TTL index on `completed_at` drops finished replay sessions after 48 hours. The API process consumes too: `sse-bridge-pool` and `notification-service` are registered in `app/core/dishka_lifespan.py`. @@ -850,7 +851,7 @@ git status --porcelain -- docs/reference/openapi.json frontend/src/lib/api On GitHub every check above is a job of `.github/workflows/ci.yml` that runs only when its files changed (`dorny/paths-filter`), and `All checks pass` (`re-actors/alls-green`) is the one required check on `main`, with no bypass for anyone. A new CI job goes into that job's `needs` and `allowed-skips`. Open a PR with auto-merge on (`gh pr merge --auto --squash`); a stack from `gh stack` lands with `gh stack merge` once every PR in it passes, since auto-merge doesn't cover stacks yet. -CI also runs the backend E2E suite and Playwright against a k3s stack, plus bandit and image scans. The grimp check (`backend/scripts/check_orphan_modules.py`) fails on any module in `app` that no other module imports. It skips `app.main`, the only whitelisted entry point, and `__init__.py` files that hold nothing but a docstring. +CI also runs the backend E2E suite and Playwright against a k3s stack, plus bandit and image scans. The grimp check (`backend/scripts/check_orphan_modules.py`) fails on any module in `app` that no other module imports. It skips the two entry points, `app.main` and `app.worker`, and `__init__.py` files that hold nothing but a docstring. ## Security Constraints diff --git a/README.md b/README.md index 22a771ea5..4737cadbf 100644 --- a/README.md +++ b/README.md @@ -169,7 +169,7 @@ The status stays completed because the pod itself ran fine. The script's own fai system diagram -A Svelte 5 frontend talks to a FastAPI backend, which keeps its state in MongoDB and Redis and publishes every execution request to Kafka. Workers take it from there: the saga orchestrator schedules the run, the k8s worker starts a locked-down pod, the pod monitor watches it, and the result processor stores the output. The frontend follows each run live over SSE. +A Svelte 5 frontend talks to a FastAPI backend, which keeps its state in MongoDB and Redis and publishes every execution request to Kafka. The worker takes it from there: its saga orchestrator schedules the run, the k8s worker starts a locked-down pod, the pod monitor watches it, and the result processor stores the output. The frontend follows each run live over SSE. ## License diff --git a/backend/Dockerfile b/backend/Dockerfile index 5ec6bce02..bb01914df 100644 --- a/backend/Dockerfile +++ b/backend/Dockerfile @@ -3,7 +3,6 @@ FROM base # Copy application files and configuration COPY ./app /app/app -COPY ./workers /app/workers COPY ./scripts /app/scripts COPY config.toml /app/config.toml COPY openssl.cnf /app/openssl.cnf diff --git a/backend/app/core/container.py b/backend/app/core/container.py index 58965a85a..05ee78528 100644 --- a/backend/app/core/container.py +++ b/backend/app/core/container.py @@ -65,8 +65,8 @@ def create_app_container(settings: Settings) -> AsyncContainer: ) -def create_result_processor_container(settings: Settings) -> AsyncContainer: - """Create a minimal DI container for the ResultProcessor worker.""" +def create_worker_container(settings: Settings) -> AsyncContainer: + """Create the DI container for the worker process.""" return make_async_container( SettingsProvider(), LoggingProvider(), @@ -77,97 +77,13 @@ def create_result_processor_container(settings: Settings) -> AsyncContainer: RepositoryProvider(), MessagingProvider(), DLQProvider(), + ExecutionQueueProvider(), RuntimeSettingsProvider(), - ResultProcessorProvider(), - context={Settings: settings}, - ) - - -def create_k8s_worker_container(settings: Settings) -> AsyncContainer: - """Create DI container for the KubernetesWorker.""" - return make_async_container( - SettingsProvider(), - LoggingProvider(), - BrokerProvider(), - RedisProvider(), - CoreServicesProvider(), - MetricsProvider(), - RepositoryProvider(), - MessagingProvider(), - DLQProvider(), - KubernetesProvider(), - K8sWorkerProvider(), - context={Settings: settings}, - ) - - -def create_pod_monitor_container(settings: Settings) -> AsyncContainer: - """Create DI container for the PodMonitor worker.""" - return make_async_container( - SettingsProvider(), - LoggingProvider(), - BrokerProvider(), - RedisProvider(), - CoreServicesProvider(), - MetricsProvider(), - RepositoryProvider(), - MessagingProvider(), - DLQProvider(), KafkaServicesProvider(), KubernetesProvider(), + K8sWorkerProvider(), PodMonitorProvider(), - context={Settings: settings}, - ) - - -def create_saga_orchestrator_container(settings: Settings) -> AsyncContainer: - """Create DI container for the SagaOrchestrator worker.""" - return make_async_container( - SettingsProvider(), - LoggingProvider(), - BrokerProvider(), - RedisProvider(), - CoreServicesProvider(), - MetricsProvider(), - RepositoryProvider(), - MessagingProvider(), - DLQProvider(), - ExecutionQueueProvider(), - RuntimeSettingsProvider(), + ResultProcessorProvider(), SagaOrchestratorProvider(), context={Settings: settings}, ) - - -def create_event_replay_container(settings: Settings) -> AsyncContainer: - """Create DI container for the EventReplay worker.""" - return make_async_container( - SettingsProvider(), - LoggingProvider(), - BrokerProvider(), - RedisProvider(), - CoreServicesProvider(), - MetricsProvider(), - RepositoryProvider(), - MessagingProvider(), - DLQProvider(), - SSEProvider(), - EventReplayProvider(), - context={Settings: settings}, - ) - - -def create_dlq_processor_container(settings: Settings) -> AsyncContainer: - """Create DI container for the DLQ processor worker.""" - return make_async_container( - SettingsProvider(), - LoggingProvider(), - BrokerProvider(), - RedisProvider(), - CoreServicesProvider(), - MetricsProvider(), - RepositoryProvider(), - MessagingProvider(), - DLQProvider(), - context={Settings: settings}, - ) diff --git a/backend/app/db/docs/replay.py b/backend/app/db/docs/replay.py index 0ef4a06ee..e1504cee0 100644 --- a/backend/app/db/docs/replay.py +++ b/backend/app/db/docs/replay.py @@ -81,4 +81,5 @@ class Settings: indexes = [ IndexModel([("status", 1)]), IndexModel([("replay_id", 1)]), + IndexModel([("completed_at", 1)], expireAfterSeconds=48 * 3600), ] diff --git a/backend/app/settings.py b/backend/app/settings.py index 6de0b651b..043962b94 100644 --- a/backend/app/settings.py +++ b/backend/app/settings.py @@ -24,14 +24,14 @@ class Settings(BaseModel): Non-secret config comes from TOML files; secrets come from env vars with dev defaults. Load order (each layer overrides the previous): - 1. config_path — base settings (committed to git) - 2. override_path — per-worker service overrides (TRACING_SERVICE_NAME, etc.) + 1. config_path: base settings (committed to git) + 2. override_path: the worker's overrides (TRACING_SERVICE_NAME) 3. Environment variables for secrets: SECRET_KEY, MONGO_USER, MONGO_PASSWORD, REDIS_PASSWORD Usage: Settings() # config.toml + env vars Settings(config_path="config.test.toml") # test config + env vars - Settings(override_path="config.saga-orchestrator.toml") # base + worker + env vars + Settings(override_path="config.worker.toml") # base + worker + env vars """ model_config = ConfigDict(extra="forbid") diff --git a/backend/workers/run_pod_monitor.py b/backend/app/worker.py similarity index 60% rename from backend/workers/run_pod_monitor.py rename to backend/app/worker.py index b9e133590..cf91b9952 100644 --- a/backend/workers/run_pod_monitor.py +++ b/backend/app/worker.py @@ -9,22 +9,26 @@ from kubernetes_asyncio.client.rest import ApiException from pymongo import AsyncMongoClient -from app.core.container import create_pod_monitor_container +from app.core.container import create_worker_container from app.core.logging import setup_log_exporter, setup_logger from app.core.metrics import KubernetesMetrics from app.db.docs import ALL_DOCUMENTS +from app.dlq.manager import DLQManager +from app.events.handlers import k8s_worker_router, result_processor_router, saga_router +from app.services.k8s_worker import KubernetesWorker from app.services.pod_monitor import ErrorType, PodMonitor +from app.services.saga import SagaOrchestrator from app.settings import Settings def main() -> None: - """Main entry point for pod monitor worker""" - settings = Settings(override_path="config.pod-monitor.toml") + """Run everything outside the API: the Kafka consumers, the pod watch and the scheduled jobs.""" + settings = Settings(override_path="config.worker.toml") logger = setup_logger(settings.LOG_LEVEL) setup_log_exporter(settings, logger) - logger.info("Starting PodMonitor worker...") + logger.info("Starting worker...") async def run() -> None: # Initialize Beanie with tz_aware client (so MongoDB returns aware datetimes) @@ -35,22 +39,24 @@ async def run() -> None: ) logger.info("MongoDB initialized via Beanie") - # Create DI container - container = create_pod_monitor_container(settings) - - # Get broker from DI (PodMonitor publishes events through UnifiedProducer) + container = create_worker_container(settings) broker: KafkaBroker = await container.get(KafkaBroker) - # Set up DI integration (no subscribers for pod monitor - it only publishes) + # Include the subscribers, then set up DI integration + broker.include_routers(k8s_worker_router(), result_processor_router(), saga_router()) setup_dishka(container, broker=broker, auto_inject=True) scheduler = AsyncIOScheduler() - async def init_monitor() -> None: + async def start() -> None: + k8s_worker = await container.get(KubernetesWorker) + await k8s_worker.ensure_namespace_security() + await k8s_worker.ensure_image_pre_puller_daemonset() + monitor = await container.get(PodMonitor) kubernetes_metrics = await container.get(KubernetesMetrics) - async def _watch_cycle() -> None: + async def watch_pods() -> None: error_type: ErrorType | None = None try: await monitor.watch_pod_events() @@ -70,28 +76,33 @@ async def _watch_cycle() -> None: kubernetes_metrics.record_pod_monitor_watch_error(error_type) kubernetes_metrics.increment_pod_monitor_watch_reconnects() - scheduler.add_job( - _watch_cycle, - trigger="interval", - seconds=5, - id="pod_monitor_watch", - max_instances=1, - misfire_grace_time=60, + orchestrator = await container.get(SagaOrchestrator) + dlq_manager = await container.get(DLQManager) + + jobs = ( + ("pod_monitor_watch", watch_pods, 5, 60), + ("saga_try_schedule", orchestrator.try_schedule_from_queue, 10, 30), + ("saga_check_timeouts", orchestrator.check_timeouts, 30, 60), + ("dlq_monitor_retries", dlq_manager.process_monitoring_cycle, 10, 60), ) + for job_id, job, seconds, grace in jobs: + scheduler.add_job( + job, trigger="interval", seconds=seconds, id=job_id, max_instances=1, misfire_grace_time=grace + ) scheduler.start() - logger.info("PodMonitor initialized (APScheduler interval=5s)") + logger.info("Worker started", jobs=[job_id for job_id, *_ in jobs]) async def stop_scheduler() -> None: scheduler.shutdown(wait=False) app = FastStream( broker, - on_startup=[init_monitor], + on_startup=[start], on_shutdown=[stop_scheduler], after_shutdown=[container.close], ) await app.run() - logger.info("PodMonitor shutdown complete") + logger.info("Worker shutdown complete") asyncio.run(run()) diff --git a/backend/config.dlq-processor.toml b/backend/config.dlq-processor.toml deleted file mode 100644 index 5d8447fc8..000000000 --- a/backend/config.dlq-processor.toml +++ /dev/null @@ -1,2 +0,0 @@ -TRACING_SERVICE_NAME = "dlq-processor" -KAFKA_CONSUMER_GROUP_ID = "dlq-processor" diff --git a/backend/config.event-replay.toml b/backend/config.event-replay.toml deleted file mode 100644 index 92ea9e784..000000000 --- a/backend/config.event-replay.toml +++ /dev/null @@ -1,2 +0,0 @@ -TRACING_SERVICE_NAME = "event-replay" -KAFKA_CONSUMER_GROUP_ID = "event-replay" diff --git a/backend/config.k8s-worker.toml b/backend/config.k8s-worker.toml deleted file mode 100644 index 2fb752833..000000000 --- a/backend/config.k8s-worker.toml +++ /dev/null @@ -1,2 +0,0 @@ -TRACING_SERVICE_NAME = "k8s-worker" -KAFKA_CONSUMER_GROUP_ID = "k8s-worker" diff --git a/backend/config.pod-monitor.toml b/backend/config.pod-monitor.toml deleted file mode 100644 index 2f2a45e61..000000000 --- a/backend/config.pod-monitor.toml +++ /dev/null @@ -1,2 +0,0 @@ -TRACING_SERVICE_NAME = "pod-monitor" -KAFKA_CONSUMER_GROUP_ID = "pod-monitor" diff --git a/backend/config.result-processor.toml b/backend/config.result-processor.toml deleted file mode 100644 index ae697d602..000000000 --- a/backend/config.result-processor.toml +++ /dev/null @@ -1,2 +0,0 @@ -TRACING_SERVICE_NAME = "result-processor" -KAFKA_CONSUMER_GROUP_ID = "result-processor" diff --git a/backend/config.saga-orchestrator.toml b/backend/config.saga-orchestrator.toml deleted file mode 100644 index 6d259608f..000000000 --- a/backend/config.saga-orchestrator.toml +++ /dev/null @@ -1,2 +0,0 @@ -TRACING_SERVICE_NAME = "saga-orchestrator" -KAFKA_CONSUMER_GROUP_ID = "saga-orchestrator" diff --git a/backend/config.worker.toml b/backend/config.worker.toml new file mode 100644 index 000000000..7d7b71029 --- /dev/null +++ b/backend/config.worker.toml @@ -0,0 +1 @@ +TRACING_SERVICE_NAME = "integr8scode-worker" diff --git a/backend/pyproject.toml b/backend/pyproject.toml index a1e45d4a1..051941cf3 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -98,7 +98,7 @@ requires = ["hatchling"] build-backend = "hatchling.build" [tool.hatch.build.targets.wheel] -packages = ["app", "workers"] +packages = ["app"] [dependency-groups] test = [ diff --git a/backend/scripts/check_orphan_modules.py b/backend/scripts/check_orphan_modules.py index 93a61bba3..8b82d1439 100644 --- a/backend/scripts/check_orphan_modules.py +++ b/backend/scripts/check_orphan_modules.py @@ -2,8 +2,8 @@ """Detect orphan modules in the app package using grimp. An orphan module is one that no other module within the package imports. -Entry points are whitelisted since they are invoked directly by gunicorn -or worker runner scripts, not imported by sibling modules. +Entry points are whitelisted since gunicorn and `python -m` run them directly, +not through an import from a sibling module. Usage: uv run python -m scripts.check_orphan_modules @@ -18,6 +18,7 @@ ENTRY_POINTS: frozenset[str] = frozenset( { "app.main", + "app.worker", } ) diff --git a/backend/tests/unit/app/test_worker_entrypoints.py b/backend/tests/unit/app/test_worker_entrypoint.py similarity index 64% rename from backend/tests/unit/app/test_worker_entrypoints.py rename to backend/tests/unit/app/test_worker_entrypoint.py index f41052364..e93defe4c 100644 --- a/backend/tests/unit/app/test_worker_entrypoints.py +++ b/backend/tests/unit/app/test_worker_entrypoint.py @@ -1,4 +1,3 @@ -import importlib from collections.abc import Awaitable, Callable, Coroutine, Sequence from types import SimpleNamespace from typing import Any @@ -7,27 +6,17 @@ import pytest import structlog +from app import worker as module from app.settings import Settings pytestmark = pytest.mark.unit _Hook = Callable[[], Awaitable[None]] -WORKERS = { - "run_dlq_processor": ("create_dlq_processor_container", True), - "run_event_replay": ("create_event_replay_container", True), - "run_k8s_worker": ("create_k8s_worker_container", False), - "run_pod_monitor": ("create_pod_monitor_container", True), - "run_result_processor": ("create_result_processor_container", False), - "run_saga_orchestrator": ("create_saga_orchestrator_container", True), -} - -@pytest.mark.parametrize(("worker", "factory", "has_scheduler"), [(w, f, s) for w, (f, s) in WORKERS.items()]) async def test_container_closes_after_the_broker_stops( - monkeypatch: pytest.MonkeyPatch, test_settings: Settings, worker: str, factory: str, has_scheduler: bool + monkeypatch: pytest.MonkeyPatch, test_settings: Settings ) -> None: - module = importlib.import_module(f"workers.{worker}") order: list[str] = [] class FakeFastStream: @@ -66,17 +55,12 @@ async def run(self) -> None: monkeypatch.setattr(module, "setup_log_exporter", MagicMock()) monkeypatch.setattr(module, "AsyncMongoClient", MagicMock()) monkeypatch.setattr(module, "init_beanie", AsyncMock()) - monkeypatch.setattr(module, factory, lambda _settings: container) + monkeypatch.setattr(module, "create_worker_container", lambda _settings: container) monkeypatch.setattr(module, "setup_dishka", MagicMock()) monkeypatch.setattr(module, "FastStream", FakeFastStream) - for name in dir(module): - if name.startswith("register_"): - monkeypatch.setattr(module, name, MagicMock()) - if has_scheduler: - monkeypatch.setattr(module, "AsyncIOScheduler", lambda: scheduler) + monkeypatch.setattr(module, "AsyncIOScheduler", lambda: scheduler) module.main() await runs[0] - expected = ["broker.stop", "container.close"] - assert order == (["scheduler.shutdown", *expected] if has_scheduler else expected) + assert order == ["scheduler.shutdown", "broker.stop", "container.close"] diff --git a/backend/workers/__init__.py b/backend/workers/__init__.py deleted file mode 100644 index ab78daf63..000000000 --- a/backend/workers/__init__.py +++ /dev/null @@ -1 +0,0 @@ -"""Worker services for event-driven architecture""" diff --git a/backend/workers/run_dlq_processor.py b/backend/workers/run_dlq_processor.py deleted file mode 100644 index 785265ca9..000000000 --- a/backend/workers/run_dlq_processor.py +++ /dev/null @@ -1,77 +0,0 @@ -import asyncio -from typing import Any - -from apscheduler.schedulers.asyncio import AsyncIOScheduler -from beanie import init_beanie -from dishka_faststream import setup_dishka -from faststream import FastStream -from faststream.kafka import KafkaBroker -from pymongo import AsyncMongoClient - -from app.core.container import create_dlq_processor_container -from app.core.logging import setup_log_exporter, setup_logger -from app.db.docs import ALL_DOCUMENTS -from app.dlq.manager import DLQManager -from app.settings import Settings - - -def main() -> None: - """Main entry point for DLQ processor worker.""" - settings = Settings(override_path="config.dlq-processor.toml") - - logger = setup_logger(settings.LOG_LEVEL) - setup_log_exporter(settings, logger) - - logger.info("Starting DLQ Processor worker...") - - async def run() -> None: - # Initialize Beanie with tz_aware client (so MongoDB returns aware datetimes) - client: AsyncMongoClient[dict[str, Any]] = AsyncMongoClient(settings.MONGODB_URL, tz_aware=True) - await init_beanie( - database=client.get_default_database(default=settings.MONGO_DB), - document_models=ALL_DOCUMENTS, - ) - logger.info("MongoDB initialized via Beanie") - - # Create DI container - container = create_dlq_processor_container(settings) - - # Get broker from DI - broker: KafkaBroker = await container.get(KafkaBroker) - - # Set up DI integration (no subscribers — DLQ manager uses APScheduler, - # broker is only needed for publishing retry/status events) - setup_dishka(container, broker=broker, auto_inject=True) - - scheduler = AsyncIOScheduler() - - async def init_dlq() -> None: - manager = await container.get(DLQManager) - scheduler.add_job( - manager.process_monitoring_cycle, - trigger="interval", - seconds=10, - id="dlq_monitor_retries", - max_instances=1, - misfire_grace_time=60, - ) - scheduler.start() - logger.info("DLQ Processor initialized (APScheduler interval=10s)") - - async def stop_scheduler() -> None: - scheduler.shutdown(wait=False) - - app = FastStream( - broker, - on_startup=[init_dlq], - on_shutdown=[stop_scheduler], - after_shutdown=[container.close], - ) - await app.run() - logger.info("DLQ Processor shutdown complete") - - asyncio.run(run()) - - -if __name__ == "__main__": - main() diff --git a/backend/workers/run_event_replay.py b/backend/workers/run_event_replay.py deleted file mode 100644 index 1e376b51f..000000000 --- a/backend/workers/run_event_replay.py +++ /dev/null @@ -1,72 +0,0 @@ -import asyncio -from typing import Any - -from apscheduler.schedulers.asyncio import AsyncIOScheduler -from beanie import init_beanie -from dishka_faststream import setup_dishka -from faststream import FastStream -from faststream.kafka import KafkaBroker -from pymongo import AsyncMongoClient - -from app.core.container import create_event_replay_container -from app.core.logging import setup_log_exporter, setup_logger -from app.db.docs import ALL_DOCUMENTS -from app.services.event_replay import EventReplayService -from app.settings import Settings - - -def main() -> None: - """Main entry point for event replay service""" - settings = Settings(override_path="config.event-replay.toml") - - logger = setup_logger(settings.LOG_LEVEL) - setup_log_exporter(settings, logger) - - logger.info("Starting Event Replay Service...") - - async def run() -> None: - # Initialize Beanie with tz_aware client (so MongoDB returns aware datetimes) - client: AsyncMongoClient[dict[str, Any]] = AsyncMongoClient(settings.MONGODB_URL, tz_aware=True) - await init_beanie( - database=client.get_default_database(default=settings.MONGO_DB), - document_models=ALL_DOCUMENTS, - ) - logger.info("MongoDB initialized via Beanie") - - container = create_event_replay_container(settings) - broker: KafkaBroker = await container.get(KafkaBroker) - setup_dishka(container, broker=broker, auto_inject=True) - - scheduler = AsyncIOScheduler() - - async def init_replay() -> None: - service = await container.get(EventReplayService) - scheduler.add_job( - service.cleanup_old_sessions, - trigger="interval", - hours=6, - kwargs={"older_than_hours": 48}, - id="replay_cleanup_old_sessions", - max_instances=1, - misfire_grace_time=300, - ) - scheduler.start() - logger.info("Event replay service initialized (APScheduler interval=6h)") - - async def stop_scheduler() -> None: - scheduler.shutdown(wait=False) - - app = FastStream( - broker, - on_startup=[init_replay], - on_shutdown=[stop_scheduler], - after_shutdown=[container.close], - ) - await app.run() - logger.info("EventReplayService shutdown complete") - - asyncio.run(run()) - - -if __name__ == "__main__": - main() diff --git a/backend/workers/run_k8s_worker.py b/backend/workers/run_k8s_worker.py deleted file mode 100644 index fbb210517..000000000 --- a/backend/workers/run_k8s_worker.py +++ /dev/null @@ -1,60 +0,0 @@ -import asyncio -from typing import Any - -from beanie import init_beanie -from dishka_faststream import setup_dishka -from faststream import FastStream -from faststream.kafka import KafkaBroker -from pymongo import AsyncMongoClient - -from app.core.container import create_k8s_worker_container -from app.core.logging import setup_log_exporter, setup_logger -from app.db.docs import ALL_DOCUMENTS -from app.events.handlers import k8s_worker_router -from app.services.k8s_worker import KubernetesWorker -from app.settings import Settings - - -def main() -> None: - """Main entry point for Kubernetes worker""" - settings = Settings(override_path="config.k8s-worker.toml") - - logger = setup_logger(settings.LOG_LEVEL) - setup_log_exporter(settings, logger) - - logger.info("Starting KubernetesWorker...") - - async def run() -> None: - # Initialize Beanie with tz_aware client (so MongoDB returns aware datetimes) - client: AsyncMongoClient[dict[str, Any]] = AsyncMongoClient(settings.MONGODB_URL, tz_aware=True) - await init_beanie( - database=client.get_default_database(default=settings.MONGO_DB), - document_models=ALL_DOCUMENTS, - ) - logger.info("MongoDB initialized via Beanie") - - # Create DI container - container = create_k8s_worker_container(settings) - - # Get broker from DI - broker: KafkaBroker = await container.get(KafkaBroker) - - # Include the subscribers, then set up DI integration - broker.include_router(k8s_worker_router()) - setup_dishka(container, broker=broker, auto_inject=True) - - async def init_k8s_worker() -> None: - worker = await container.get(KubernetesWorker) - await worker.ensure_namespace_security() - await worker.ensure_image_pre_puller_daemonset() - logger.info("KubernetesWorker initialized with namespace security and pre-puller daemonset") - - app = FastStream(broker, on_startup=[init_k8s_worker], after_shutdown=[container.close]) - await app.run() - logger.info("KubernetesWorker shutdown complete") - - asyncio.run(run()) - - -if __name__ == "__main__": - main() diff --git a/backend/workers/run_result_processor.py b/backend/workers/run_result_processor.py deleted file mode 100644 index a2718311b..000000000 --- a/backend/workers/run_result_processor.py +++ /dev/null @@ -1,53 +0,0 @@ -import asyncio -from typing import Any - -from beanie import init_beanie -from dishka_faststream import setup_dishka -from faststream import FastStream -from faststream.kafka import KafkaBroker -from pymongo import AsyncMongoClient - -from app.core.container import create_result_processor_container -from app.core.logging import setup_log_exporter, setup_logger -from app.db.docs import ALL_DOCUMENTS -from app.events.handlers import result_processor_router -from app.settings import Settings - - -def main() -> None: - """Main entry point for result processor worker""" - settings = Settings(override_path="config.result-processor.toml") - - logger = setup_logger(settings.LOG_LEVEL) - setup_log_exporter(settings, logger) - - logger.info("Starting ResultProcessor worker...") - - async def run() -> None: - # Initialize Beanie with tz_aware client (so MongoDB returns aware datetimes) - client: AsyncMongoClient[dict[str, Any]] = AsyncMongoClient(settings.MONGODB_URL, tz_aware=True) - await init_beanie( - database=client.get_default_database(default=settings.MONGO_DB), - document_models=ALL_DOCUMENTS, - ) - logger.info("MongoDB initialized via Beanie") - - # Create DI container - container = create_result_processor_container(settings) - - # Get broker from DI - broker: KafkaBroker = await container.get(KafkaBroker) - - # Include the subscribers, then set up DI integration - broker.include_router(result_processor_router()) - setup_dishka(container, broker=broker, auto_inject=True) - - app = FastStream(broker, after_shutdown=[container.close]) - await app.run() - logger.info("ResultProcessor shutdown complete") - - asyncio.run(run()) - - -if __name__ == "__main__": - main() diff --git a/backend/workers/run_saga_orchestrator.py b/backend/workers/run_saga_orchestrator.py deleted file mode 100644 index bba74161d..000000000 --- a/backend/workers/run_saga_orchestrator.py +++ /dev/null @@ -1,86 +0,0 @@ -import asyncio -from typing import Any - -from apscheduler.schedulers.asyncio import AsyncIOScheduler -from beanie import init_beanie -from dishka_faststream import setup_dishka -from faststream import FastStream -from faststream.kafka import KafkaBroker -from pymongo import AsyncMongoClient - -from app.core.container import create_saga_orchestrator_container -from app.core.logging import setup_log_exporter, setup_logger -from app.db.docs import ALL_DOCUMENTS -from app.events.handlers import saga_router -from app.services.saga import SagaOrchestrator -from app.settings import Settings - - -def main() -> None: - """Main entry point for saga orchestrator worker""" - settings = Settings(override_path="config.saga-orchestrator.toml") - - logger = setup_logger(settings.LOG_LEVEL) - setup_log_exporter(settings, logger) - - logger.info("Starting Saga Orchestrator worker...") - - async def run() -> None: - # Initialize Beanie with tz_aware client (so MongoDB returns aware datetimes) - client: AsyncMongoClient[dict[str, Any]] = AsyncMongoClient(settings.MONGODB_URL, tz_aware=True) - await init_beanie( - database=client.get_default_database(default=settings.MONGO_DB), - document_models=ALL_DOCUMENTS, - ) - logger.info("MongoDB initialized via Beanie") - - # Create DI container - container = create_saga_orchestrator_container(settings) - - # Get broker from DI - broker: KafkaBroker = await container.get(KafkaBroker) - - # Include the subscribers, then set up DI integration - broker.include_router(saga_router()) - setup_dishka(container, broker=broker, auto_inject=True) - - scheduler = AsyncIOScheduler() - - async def init_saga() -> None: - orchestrator = await container.get(SagaOrchestrator) - scheduler.add_job( - orchestrator.check_timeouts, - trigger="interval", - seconds=30, - id="saga_check_timeouts", - max_instances=1, - misfire_grace_time=60, - ) - scheduler.add_job( - orchestrator.try_schedule_from_queue, - trigger="interval", - seconds=10, - id="saga_try_schedule", - max_instances=1, - misfire_grace_time=30, - ) - scheduler.start() - logger.info("SagaOrchestrator initialized (APScheduler: timeouts=30s, scheduling=10s)") - - async def stop_scheduler() -> None: - scheduler.shutdown(wait=False) - - app = FastStream( - broker, - on_startup=[init_saga], - on_shutdown=[stop_scheduler], - after_shutdown=[container.close], - ) - await app.run() - logger.info("SagaOrchestrator shutdown complete") - - asyncio.run(run()) - - -if __name__ == "__main__": - main() diff --git a/docker-compose.yaml b/docker-compose.yaml index 1fc727f4a..2b7959b4c 100644 --- a/docker-compose.yaml +++ b/docker-compose.yaml @@ -125,7 +125,6 @@ services: condition: service_healthy volumes: - ./backend/app:/app/app - - ./backend/workers:/app/workers - ./backend/scripts:/app/scripts - ./backend/tests:/app/tests:ro - ./backend/certs:/app/certs:ro @@ -350,65 +349,13 @@ services: - app-network restart: "no" # Run once and exit - # Event-driven workers - k8s-worker: + # Kafka consumers, the pod watch and the scheduled jobs + worker: logging: *logging image: ghcr.io/hardmax71/integr8scode/backend:${IMAGE_TAG:-latest} - container_name: k8s-worker - mem_limit: 160m - command: ["python", "workers/run_k8s_worker.py"] - depends_on: - user-seed: - condition: service_completed_successfully - kafka: - condition: service_healthy - mongo: - condition: service_started - environment: - <<: *backend-secrets - volumes: - - ./backend/app:/app/app:ro - - ./backend/workers:/app/workers:ro - - ./backend/kubeconfig.yaml:/app/kubeconfig.yaml:ro - - ./backend/config.toml:/app/config.toml:ro - - ./backend/config.k8s-worker.toml:/app/config.k8s-worker.toml:ro - networks: - - app-network - extra_hosts: - - "host.docker.internal:host-gateway" - restart: unless-stopped - - pod-monitor: - logging: *logging - image: ghcr.io/hardmax71/integr8scode/backend:${IMAGE_TAG:-latest} - container_name: pod-monitor - mem_limit: 160m - command: ["python", "workers/run_pod_monitor.py"] - depends_on: - user-seed: - condition: service_completed_successfully - kafka: - condition: service_healthy - environment: - <<: *backend-secrets - volumes: - - ./backend/app:/app/app:ro - - ./backend/workers:/app/workers:ro - - ./backend/kubeconfig.yaml:/app/kubeconfig.yaml:ro - - ./backend/config.toml:/app/config.toml:ro - - ./backend/config.pod-monitor.toml:/app/config.pod-monitor.toml:ro - networks: - - app-network - extra_hosts: - - "host.docker.internal:host-gateway" - restart: unless-stopped - - result-processor: - logging: *logging - image: ghcr.io/hardmax71/integr8scode/backend:${IMAGE_TAG:-latest} - container_name: result-processor - mem_limit: 160m - command: ["python", "workers/run_result_processor.py"] + container_name: worker + mem_limit: 256m + command: ["python", "-m", "app.worker"] depends_on: user-seed: condition: service_completed_successfully @@ -420,40 +367,15 @@ services: <<: *backend-secrets volumes: - ./backend/app:/app/app:ro - - ./backend/workers:/app/workers:ro - ./backend/kubeconfig.yaml:/app/kubeconfig.yaml:ro - ./backend/config.toml:/app/config.toml:ro - - ./backend/config.result-processor.toml:/app/config.result-processor.toml:ro + - ./backend/config.worker.toml:/app/config.worker.toml:ro networks: - app-network extra_hosts: - "host.docker.internal:host-gateway" restart: unless-stopped - saga-orchestrator: - logging: *logging - image: ghcr.io/hardmax71/integr8scode/backend:${IMAGE_TAG:-latest} - container_name: saga-orchestrator - mem_limit: 160m - command: ["python", "workers/run_saga_orchestrator.py"] - depends_on: - user-seed: - condition: service_completed_successfully - kafka: - condition: service_healthy - mongo: - condition: service_started - environment: - <<: *backend-secrets - volumes: - - ./backend/app:/app/app:ro - - ./backend/workers:/app/workers:ro - - ./backend/config.toml:/app/config.toml:ro - - ./backend/config.saga-orchestrator.toml:/app/config.saga-orchestrator.toml:ro - networks: - - app-network - restart: unless-stopped - # Distributed tracing with Jaeger jaeger: logging: *logging @@ -492,56 +414,6 @@ services: restart: unless-stopped # Distroless image with no shell, so no healthcheck - # Event replay service - event-replay: - logging: *logging - image: ghcr.io/hardmax71/integr8scode/backend:${IMAGE_TAG:-latest} - container_name: event-replay - mem_limit: 160m - command: ["python", "workers/run_event_replay.py"] - depends_on: - user-seed: - condition: service_completed_successfully - kafka: - condition: service_healthy - mongo: - condition: service_started - environment: - <<: *backend-secrets - volumes: - - ./backend/app:/app/app:ro - - ./backend/workers:/app/workers:ro - - ./backend/config.toml:/app/config.toml:ro - - ./backend/config.event-replay.toml:/app/config.event-replay.toml:ro - networks: - - app-network - restart: unless-stopped - - # DLQ Processor Service - dlq-processor: - logging: *logging - image: ghcr.io/hardmax71/integr8scode/backend:${IMAGE_TAG:-latest} - container_name: dlq-processor - mem_limit: 160m - command: ["python", "workers/run_dlq_processor.py"] - depends_on: - user-seed: - condition: service_completed_successfully - kafka: - condition: service_healthy - mongo: - condition: service_started - environment: - <<: *backend-secrets - volumes: - - ./backend/app:/app/app:ro - - ./backend/workers:/app/workers:ro - - ./backend/config.toml:/app/config.toml:ro - - ./backend/config.dlq-processor.toml:/app/config.dlq-processor.toml:ro - networks: - - app-network - restart: unless-stopped - # Monitoring Stack # Victoria Metrics - Time series database victoria-metrics: diff --git a/docs/architecture/domain-exceptions.md b/docs/architecture/domain-exceptions.md index 15ff36191..0f1598b00 100644 --- a/docs/architecture/domain-exceptions.md +++ b/docs/architecture/domain-exceptions.md @@ -100,7 +100,7 @@ return script The route just calls the service and returns the schema. It has no `None` check and no `HTTPException`. -Kafka handlers in the workers raise the same exceptions, for example `ExecutionNotFoundError` in the result processor. No HTTP mapping applies there: the error goes up to FastStream, whose telemetry middleware records it under `error_type` on `messaging_process_duration_seconds`. +Kafka handlers in the worker raise the same exceptions, for example `ExecutionNotFoundError` in the result processor. No HTTP mapping applies there: the error goes up to FastStream, whose telemetry middleware records it under `error_type` on `messaging_process_duration_seconds`. ## Testing diff --git a/docs/architecture/execution-queue.md b/docs/architecture/execution-queue.md index 9627ace05..bd6c880ab 100644 --- a/docs/architecture/execution-queue.md +++ b/docs/architecture/execution-queue.md @@ -57,7 +57,7 @@ These events trigger a pass: | `execution_completed`, `execution_failed`, `execution_timeout` | mark the saga terminal, `release`, `release_user_slot` | | `execution_cancelled` | `remove` from every queue key, mark the saga cancelled, `release`, `release_user_slot` | -`release` is a `ZREM` on the running set plus deleting the event key, and `release_user_slot` looks up the owner and removes the id from that user's set, so both are safe on an id that holds nothing. On top of the events, the worker runs `try_schedule_from_queue` every 10 seconds through APScheduler (`run_saga_orchestrator.py`), which catches up after a restart or a lost event. +`release` is a `ZREM` on the running set plus deleting the event key, and `release_user_slot` looks up the owner and removes the id from that user's set, so both are safe on an id that holds nothing. On top of the events, the worker runs `try_schedule_from_queue` every 10 seconds through APScheduler (`app/worker.py`), which catches up after a restart or a lost event. ## Admin endpoints diff --git a/docs/architecture/kafka-topic-architecture.md b/docs/architecture/kafka-topic-architecture.md index 98282b32f..c4379a409 100644 --- a/docs/architecture/kafka-topic-architecture.md +++ b/docs/architecture/kafka-topic-architecture.md @@ -31,17 +31,17 @@ When the producer first publishes them, execution, pod, command, result and saga ## Consumer groups -Group IDs are string literals in `app/events/handlers.py`. The `KAFKA_CONSUMER_GROUP_ID` setting in the per-worker TOML files isn't read anywhere ([#293](https://github.com/HardMax71/Integr8sCode/issues/293)). +Group IDs are string literals in `app/events/handlers.py`. The `KAFKA_CONSUMER_GROUP_ID` setting in config.toml isn't read anywhere ([#293](https://github.com/HardMax71/Integr8sCode/issues/293)). | Group | Runs in | Topics | |-------|---------|--------| -| `saga-orchestrator` | saga-orchestrator | `execution_requested`, `execution_completed`, `execution_failed`, `execution_timeout`, `execution_cancelled` | -| `k8s-worker` | k8s-worker | `create_pod_command`, `delete_pod_command` | -| `result-processor` | result-processor | `execution_completed`, `execution_failed`, `execution_timeout`, `pod_scheduled`, `pod_running` | +| `saga-orchestrator` | worker | `execution_requested`, `execution_completed`, `execution_failed`, `execution_timeout`, `execution_cancelled` | +| `k8s-worker` | worker | `create_pod_command`, `delete_pod_command` | +| `result-processor` | worker | `execution_completed`, `execution_failed`, `execution_timeout`, `pod_scheduled`, `pod_running` | | `notification-service` | API | `execution_completed`, `execution_failed`, `execution_timeout` | | `sse-bridge-pool` | API | 17 types: every `execution_*` event except `execution_accepted`, all seven `pod_*` events, `result_stored` and `result_failed` | -Kafka delivers each message to every group that subscribes to the topic, so one `execution_completed` reaches the saga orchestrator, the result processor, the notification service and the SSE bridge. The pod monitor, event-replay and dlq-processor containers only publish. +Kafka delivers each message to every group that subscribes to the topic, so one `execution_completed` reaches the saga orchestrator, the result processor, the notification service and the SSE bridge. The worker's three groups each get their own copy even though they share a process. The pod monitor and the DLQ job only publish. Most subscribers use `AckPolicy.ACK`, which commits the offset after the handler returns, even when it raised. The SSE bridge uses `ACK_FIRST` and starts from the latest offset, since a browser only cares about events from now on. The result processor reads one record per poll and starts from the earliest offset. @@ -81,11 +81,11 @@ graph LR `EventReplayService` reads stored events that match a filter from MongoDB in batches and publishes them again through `UnifiedProducer`, so each one goes back to the topic of its own type. The event store ignores the duplicate `event_id`, so a replay doesn't double the stored history. A session can also write events to a file instead, or go nowhere for a dry test, and the admin can pause, resume or cancel it. -The service runs inside the API process, and a session's live state is kept in that process's memory. With more than one Gunicorn worker, a follow-up call for a session can land in a process that doesn't know it and get a 404. The `target_topics` field of a replay request is accepted and stored but never used. See [Event Replay](../components/workers/event_replay.md). +The service runs inside the API process, and a session's live state is kept in that process's memory. With more than one Gunicorn worker, a follow-up call for a session can land in a process that doesn't know it and get a 404. The `target_topics` field of a replay request is accepted and stored but never used. See [Event Replay](../components/event-replay.md). ## Dead letters -The DLQ is a MongoDB collection, `dlq_messages`, not a Kafka topic. `DLQManager.handle_message` stores a failed message there and schedules a retry by the event type's category: execution and command events get 5 retries with exponential backoff from 30 seconds, pod events 3 retries from 60 seconds, result events 3 immediate retries, and everything else 4 retries from 60 seconds. The dlq-processor republishes due messages to their original topic every 10 seconds and publishes `dlq_message_retried`. +The DLQ is a MongoDB collection, `dlq_messages`, not a Kafka topic. `DLQManager.handle_message` stores a failed message there and schedules a retry by the event type's category: execution and command events get 5 retries with exponential backoff from 30 seconds, pod events 3 retries from 60 seconds, result events 3 immediate retries, and everything else 4 retries from 60 seconds. The worker's DLQ job republishes due messages to their original topic every 10 seconds and publishes `dlq_message_retried`. `DeadLetterMiddleware` on the k8s worker, result processor, saga orchestrator and notification routers calls `handle_message` when a handler raises. A retry goes back to the original topic with a header naming the failed consumer group, and every other group skips it. See [Dead Letter Queue](../components/dead-letter-queue.md). diff --git a/docs/architecture/lifecycle.md b/docs/architecture/lifecycle.md index 3df0dab02..e6d6c311f 100644 --- a/docs/architecture/lifecycle.md +++ b/docs/architecture/lifecycle.md @@ -1,6 +1,6 @@ # Service lifecycle -Services don't manage their own lifecycle. They get their dependencies through Dishka constructors and expose plain async methods, with no `start()`, `stop()` or background task of their own. Whatever needs starting and stopping is owned by the process entry point: the FastAPI lifespan in `app/core/dishka_lifespan.py` for the API, and `backend/workers/run_*.py` for each worker. +Services don't manage their own lifecycle. They get their dependencies through Dishka constructors and expose plain async methods, with no `start()`, `stop()` or background task of their own. Whatever needs starting and stopping is owned by the process entry point: the FastAPI lifespan in `app/core/dishka_lifespan.py` for the API, and `app/worker.py` for the worker. That leaves a short list of things with a lifecycle. The Kafka broker needs `start()` and `stop()`. APScheduler instances need `start()` and `shutdown()`. The Redis client and the Kubernetes `ApiClient` are async generator providers in `app/core/providers.py`, so `container.close()` runs their teardown (`aclose()` and `close()`). The one service that breaks the rule is `EventReplayService`, which creates an `AsyncIOScheduler` per replay session in `start_session` and shuts it down when the session finishes or is cancelled. @@ -20,27 +20,27 @@ On shutdown it stops the scheduler, then the broker, then closes the container. gunicorn runs the API with `WEB_CONCURRENCY` workers (2 by default), and each one runs its own lifespan. That means each has its own broker connection, its own notification scheduler and its own in-memory replay sessions. -## Workers +## Worker process -All six workers follow the same shape. Each loads `Settings(override_path="config..toml")`, sets up logging, runs `init_beanie`, builds its own container with `create__container`, registers its subscribers if it has any, calls `setup_dishka`, and then blocks on `FastStream(broker, on_startup=[...], after_shutdown=[container.close]).run()`. +`app/worker.py` loads `Settings(override_path="config.worker.toml")`, sets up logging, runs `init_beanie`, builds its container with `create_worker_container`, registers the k8s worker, result processor and saga routers, calls `setup_dishka`, and then blocks on `FastStream(broker, on_startup=[start], on_shutdown=[stop_scheduler], after_shutdown=[container.close]).run()`. -FastStream 0.7 runs `on_startup` hooks before it starts the broker and `after_startup` hooks after. The same goes for shutdown: `on_shutdown` runs before `broker.stop()` and `after_shutdown` after it. Startup work happens before the broker connects. A worker with a scheduler stops it in `on_shutdown`, and every worker closes its container in `after_shutdown`, so Redis and the Kubernetes client stay open until the broker has drained its handlers. +FastStream 0.7 runs `on_startup` hooks before it starts the broker and `after_startup` hooks after. The same goes for shutdown: `on_shutdown` runs before `broker.stop()` and `after_shutdown` after it. So the startup work happens before the broker connects, the scheduler stops before the broker drains, and the container closes last, which keeps Redis and the Kubernetes client open until the handlers are done. -| Worker | Subscribers | Startup work | Shutdown | -|--------|-------------|--------------|----------| -| k8s-worker | `create_pod_command`, `delete_pod_command` | `ensure_namespace_security()` and `ensure_image_pre_puller_daemonset()` | `container.close` | -| pod-monitor | none | APScheduler runs the watch cycle every 5s | scheduler, then container | -| result-processor | `execution_completed`, `execution_failed`, `execution_timeout` | none | `container.close` | -| saga-orchestrator | `execution_requested`, plus completed, failed, timeout and cancelled | APScheduler runs `check_timeouts` every 30s and `try_schedule_from_queue` every 10s | scheduler, then container | -| dlq-processor | none | APScheduler runs `process_monitoring_cycle` every 10s | scheduler, then container | -| event-replay | none | APScheduler runs `cleanup_old_sessions(older_than_hours=48)` every 6 hours | scheduler, then container | +`start` calls `ensure_namespace_security()` and `ensure_image_pre_puller_daemonset()`, then schedules four APScheduler jobs, each with `max_instances=1`: -The pod monitor's watch cycle calls `watch_pod_events()`. On a 410 from the API server it resets the resource version so the next cycle starts a fresh watch, and every error that ends the cycle is counted in the watch error and reconnect metrics. An error while handling one pod event is caught inside `PodMonitor` and counted as a watch error only, and the watch carries on. +| Job | Interval | Calls | +|-----|----------|-------| +| `pod_monitor_watch` | 5s | the watch cycle around `PodMonitor.watch_pod_events()` | +| `saga_try_schedule` | 10s | `SagaOrchestrator.try_schedule_from_queue` | +| `saga_check_timeouts` | 30s | `SagaOrchestrator.check_timeouts` | +| `dlq_monitor_retries` | 10s | `DLQManager.process_monitoring_cycle` | -Replay sessions are created and driven through the API's replay routes, so they live in the API process. The event-replay worker never holds a session in memory, and its cleanup job only deletes terminal sessions older than 48 hours from MongoDB. +The watch cycle calls `watch_pod_events()`. On a 410 from the API server it resets the resource version so the next cycle starts a fresh watch, and every error that ends the cycle is counted in the watch error and reconnect metrics. An error while handling one pod event is caught inside `PodMonitor` and counted as a watch error only, and the watch carries on. + +Replay sessions are created and driven through the API's replay routes, so they live in the API process. MongoDB deletes finished sessions 48 hours after they complete, through a TTL index on `completed_at`. ## Adding a service or worker A new service is a plain class with constructor dependencies and async methods. If it needs periodic work, add an APScheduler job in the entry point that owns it and keep the service unaware of scheduling. -A new worker copies one of the `run_*.py` files. Give it a `config..toml`, a container factory in `app/core/container.py`, and a router factory in `app/events/handlers.py` if it consumes events. Include the router and call `setup_dishka` before `app.run()`. Stop schedulers in `on_shutdown` and close the container in `after_shutdown`, so nothing the handlers use closes before the broker stops. +New background work goes into the worker rather than a new process. A consumer gets a router factory in `app/events/handlers.py` with its own `group_id`, included next to the others in `app/worker.py`, and its provider joins `create_worker_container`. A periodic job is one more `add_job` in `start`. diff --git a/docs/architecture/overview.md b/docs/architecture/overview.md index 879de6fa6..f36978a05 100644 --- a/docs/architecture/overview.md +++ b/docs/architecture/overview.md @@ -10,7 +10,7 @@ Integr8sCode runs user scripts in Python, Node.js, Ruby, Bash or Go, each in its - :material-cog-outline:{ .lg .middle } **[Services](services-overview.md)** - What lives in `backend/app/services/` and what each worker consumes and publishes + What lives in `backend/app/services/` and what each part of the worker consumes and publishes - :material-transit-connection-variant:{ .lg .middle } **[Kafka Topics](kafka-topic-architecture.md)** @@ -20,9 +20,9 @@ Integr8sCode runs user scripts in Python, Node.js, Ruby, Bash or Go, each in its How Kafka events reach the browser through Redis pub/sub -- :material-robot-industrial-outline:{ .lg .middle } **[Workers](../components/workers/index.md)** +- :material-robot-industrial-outline:{ .lg .middle } **[Worker](../components/workers/index.md)** - The six background processes that run next to the API + The one background process that runs next to the API @@ -32,7 +32,7 @@ Integr8sCode runs user scripts in Python, Node.js, Ruby, Bash or Go, each in its The browser loads the SPA from the frontend container, an nginx on port 5001 that also proxies `/api` to the backend over HTTPS. The backend runs FastAPI under Gunicorn with Uvicorn workers and serves both REST and SSE. Kafka runs in KRaft mode and creates topics on first use, so there is no topic setup step. Events travel as JSON that FastStream serializes from the Pydantic models. -Every worker is a separate container built from the same backend image with a different command. The k8s-worker creates pods through the Kubernetes API, the pod-monitor watches them, the result-processor writes results to MongoDB, and the saga-orchestrator schedules executions with a Redis-backed queue and tracks sagas in MongoDB. The dlq-processor and event-replay containers round it out. +The worker is one more container from the same backend image, running `python -m app.worker`. Inside it, the k8s worker creates pods through the Kubernetes API, the pod monitor watches them, the result processor writes results to MongoDB, and the saga orchestrator schedules executions with a Redis-backed queue and tracks sagas in MongoDB. A DLQ job retries failed events every 10 seconds. The observability profile adds an OpenTelemetry Collector that ships traces to Jaeger, metrics to VictoriaMetrics and logs to Loki, with Grafana on top of all three. Kafdrop is under the debug profile. The cert-generator creates TLS certificates for the frontend and backend from a shared CA, and it also prepares the cluster: it creates the `integr8scode` namespace, installs Kueue, sets up a service account with RBAC and writes the kubeconfig the backend containers use. @@ -149,7 +149,7 @@ Notifications use the same bus on `sse:notif:{user_id}` through `GET /api/v1/eve ## Kafka, sagas and replay -Kafka carries all asynchronous work, with [one topic per event type](kafka-topic-architecture.md). [Sagas](../components/workers/saga_orchestrator.md) run the execution steps and publish compensating commands when a step fails. The [replay system](../components/workers/event_replay.md) lets admins re-publish stored events. A failed handler's event goes to the [dead letter queue](../components/dead-letter-queue.md) and is retried for the consumer group that failed. +Kafka carries all asynchronous work, with [one topic per event type](kafka-topic-architecture.md). [Sagas](../components/workers/saga_orchestrator.md) run the execution steps and publish compensating commands when a step fails. The [replay system](../components/event-replay.md) lets admins re-publish stored events. A failed handler's event goes to the [dead letter queue](../components/dead-letter-queue.md) and is retried for the consumer group that failed. ## Notifications @@ -167,4 +167,4 @@ graph TD ## API surface -The public surface is the REST and SSE endpoints under `/api/v1/`. MongoDB, Redis, Kafka, the Kubernetes API and the workers stay internal. The frontend talks to the backend with fetch and EventSource and [cookie-based auth](authentication.md). The [admin routes](../operations/admin-api.md) under `/api/v1/admin/` and the replay routes under `/api/v1/replay/` require the admin role. The OpenAPI schema and the Swagger UI are served only when `DEVELOPMENT_MODE` is on. See the [API Reference](../reference/api-reference.md) for the full list. +The public surface is the REST and SSE endpoints under `/api/v1/`. MongoDB, Redis, Kafka, the Kubernetes API and the worker stay internal. The frontend talks to the backend with fetch and EventSource and [cookie-based auth](authentication.md). The [admin routes](../operations/admin-api.md) under `/api/v1/admin/` and the replay routes under `/api/v1/replay/` require the admin role. The OpenAPI schema and the Swagger UI are served only when `DEVELOPMENT_MODE` is on. See the [API Reference](../reference/api-reference.md) for the full list. diff --git a/docs/architecture/services-overview.md b/docs/architecture/services-overview.md index 7b653dd98..c36d70288 100644 --- a/docs/architecture/services-overview.md +++ b/docs/architecture/services-overview.md @@ -1,6 +1,6 @@ # Services overview -This page maps `backend/app/services/` and the workers that run outside the API. Read it for a mental model before you open the code. +This page maps `backend/app/services/` and the worker that runs outside the API. Read it for a mental model before you open the code. ## The API's services @@ -22,21 +22,20 @@ Route handlers get these from Dishka. They hold the business logic, talk to Mong | `idempotency/` | `IdempotencyManager` and its Redis repository. See [Idempotency](idempotency.md). | | `runtime_settings.py` | `RuntimeSettingsLoader`, which reads the admin-set system settings from MongoDB, falls back to the TOML values and caches the result for 60 seconds. | | `admin/` | Admin services for users, executions, events and system settings. | -| `k8s_worker/`, `pod_monitor/`, `result_processor/` | The services behind the workers of the same name. | +| `k8s_worker/`, `pod_monitor/`, `result_processor/` | The services behind the worker's parts of the same name. | The API process is a Kafka consumer too. Its lifespan registers two subscriber groups: `sse-bridge-pool` copies 17 execution, pod and result event types into Redis for SSE, and `notification-service` turns `execution_completed`, `execution_failed` and `execution_timeout` into notifications. -## Workers +## Worker -Each worker runs in its own container from the same backend image as the API, with a different command, for example `python workers/run_k8s_worker.py`. The entry points live in `backend/workers/`. Every subscriber is registered in `app/events/handlers.py`. +The worker runs in its own container from the same backend image as the API, with the command `python -m app.worker`. Its entry point is `app/worker.py`, and every subscriber is registered in `app/events/handlers.py`. -| Worker | Consumes | Publishes | +| Part | Consumes | Publishes | |--------|----------|-----------| | Saga orchestrator | `execution_requested`, `execution_completed`, `execution_failed`, `execution_timeout`, `execution_cancelled` | `create_pod_command`, `delete_pod_command`, `saga_started`, `saga_cancelled` | | K8s worker | `create_pod_command`, `delete_pod_command` | `pod_created`, or `execution_failed` when pod creation fails | | Pod monitor | nothing, it watches the Kubernetes API | `pod_scheduled`, `pod_running`, `pod_terminated`, `execution_completed`, `execution_failed`, `execution_timeout` | | Result processor | `execution_completed`, `execution_failed`, `execution_timeout` | `result_stored`, or `result_failed` when storing fails | -| Event replay | nothing | nothing, it only cleans up old sessions | | DLQ processor | nothing | retried events and `dlq_message_retried` | The saga orchestrator puts each request into the execution queue and runs the saga once a slot is free. The saga steps are `validate_execution`, `allocate_resources` and `create_pod`; the last one publishes the pod command. If a step fails, the compensations of the steps that already ran go in reverse order: `delete_pod` publishes `delete_pod_command` and `release_resources` frees the allocation. Terminal execution events mark the saga done and release the queue slot. `execution_cancelled` takes a queued request out of the queue and marks the saga cancelled, but it doesn't delete a pod that is already running. Two APScheduler jobs run next to the consumers: `check_timeouts` every 30 seconds and `try_schedule_from_queue` every 10 seconds. @@ -47,7 +46,7 @@ The pod monitor lists and then watches pods labelled `app=integr8s,component=exe The result processor writes the final status, output and resource usage to MongoDB and records execution metrics. It doesn't touch Kubernetes. -The event-replay container only runs `cleanup_old_sessions` every 6 hours. Replay sessions themselves run inside the API process, which owns the `EventReplayService` the admin routes use. +Replay sessions run inside the API process, which owns the `EventReplayService` the admin routes use. MongoDB drops finished sessions 48 hours after they complete. The DLQ processor runs `process_monitoring_cycle` every 10 seconds, which republishes due messages from the `dlq_messages` collection to their original topic for the consumer group that failed on them. The k8s worker, result processor, saga orchestrator and notification routers save their failed events there. See [Dead Letter Queue](../components/dead-letter-queue.md). diff --git a/docs/assets/images/system_diagram.svg b/docs/assets/images/system_diagram.svg index 32a5368e4..b694284a1 100644 --- a/docs/assets/images/system_diagram.svg +++ b/docs/assets/images/system_diagram.svg @@ -1,4 +1,4 @@ -
HTTPS 443
(SPA + Assets)
HTTPS 443...
Browser SPA
Browser SPA
HTTPS /api/v1/*
(REST/SSE endpoints)
HTTPS /api/v1/*...
frontend
(Nginx + Svelte)
frontend...
SSE /api/v1/events/*
(EventSource)
SSE /api/v1/events/*...
CRUD:
- executions
- settings
- events
CRUD:...
backend
(FastAPI + SSE)
backend...
share CA
bundle
share CA...
shared-ca
shared-ca
TLS cert
TLS cert
cert-generator
cert-generator
Cert Generator
Cert Generator
Observability
Observability
Prom
remote_write
(metrics)
Prom...
otlphtpp
otlphtppotel-collectorJaeger
OTLP gRPC
(traces)
OTLP gRPC...
query
dashboards
(PromQL)
query...
Grafana
Grafana
kafdrop
kafdrop
Kubernetes API
Kuberne...
consume commands
publish pod + exec events
consume commands...
k8s-worker
k8s-worker
publish
execution
events
publish...
pod-monitor
pod-monitor
consume
DLQ queue
consume...
dlq-processor
dlq-processor
event-replay
event-replay
create/delete
pods
create/delete...
watch/list
pods
watch/list...
kubeconfig
cluster bootstrap
kubeconfig...
Saga
states
Saga...
saga-orchestrator
saga-orchestrator
Result_stored
notifications
Result_stored...
result-processor
result-processor
Redis
Redis
Load events
for replay
Load events...
Mongo
Mongo
Rate limiting keys
+ SSE pub/sub
Rate limiting keys...
consume for
API handlers
consume for...
triggers / saga events
+ commands
triggers / saga events...
replay
historical
events
replay...
web UI:
topics/consumers
web UI:...
produce
domain events
produce...
Results
Results
Idempotency
keys/cache
Idempotency...
Project
Scope
Project...
log queries
log queries
Loki
LokiText is not SVG - cannot display \ No newline at end of file +
HTTPS 443
(SPA + Assets)
Browser SPA
HTTPS /api/v1/*
(REST/SSE endpoints)
frontend
(Nginx + Svelte)
SSE /api/v1/events/*
(EventSource)
CRUD:
- executions
- settings
- events
backend
(FastAPI + SSE)
share CA
bundle
shared-ca
TLS cert
cert-generator
Cert Generator
Observability
Prom
remote_write
(metrics)
otlphtpp
otel-collectorJaeger
OTLP gRPC
(traces)
query
dashboards
(PromQL)
Grafana
kafdrop
Kubernetes API
consume commands and events
publish pod, execution and result events
worker
saga orchestrator
k8s worker
pod monitor
result processor
DLQ retries
create/delete,
watch/list pods
kubeconfig
cluster bootstrap
Redis
Mongo
Rate limiting keys
+ SSE pub/sub
consume for
API handlers
web UI:
topics/consumers
produce
domain events
and replays
results, sagas,
DLQ rows
execution queue,
idempotency keys
Project
Scope
log queries
Loki
Text is not SVG - cannot display \ No newline at end of file diff --git a/docs/components/dead-letter-queue.md b/docs/components/dead-letter-queue.md index 8c0262d2c..5afc39b33 100644 --- a/docs/components/dead-letter-queue.md +++ b/docs/components/dead-letter-queue.md @@ -1,6 +1,6 @@ # Dead letter queue -When a Kafka handler raises, its event goes to the dead letter queue, and the dlq-processor later sends it back to the one consumer group that failed. The queue is the `dlq_messages` MongoDB collection, not a Kafka topic. FastStream has no dead letter queue for Kafka, so two small middlewares in `app/events/middlewares.py` connect the subscribers to it. +When a Kafka handler raises, its event goes to the dead letter queue, and the worker's DLQ job later sends it back to the one consumer group that failed. The queue is the `dlq_messages` MongoDB collection, not a Kafka topic. FastStream has no dead letter queue for Kafka, so two small middlewares in `app/events/middlewares.py` connect the subscribers to it. ## What happens to a failed event @@ -51,4 +51,4 @@ The `DLQ_*` keys in config.toml (`DLQ_RETRY_MAX_ATTEMPTS`, `DLQ_RETRY_BASE_DELAY | [`manager.py`](https://github.com/HardMax71/Integr8sCode/blob/main/backend/app/dlq/manager.py) | Intake, retry, discard | | [`models.py`](https://github.com/HardMax71/Integr8sCode/blob/main/backend/app/dlq/models.py) | Statuses, retry policies and the two headers | | [`dlq.py`](https://github.com/HardMax71/Integr8sCode/blob/main/backend/app/db/docs/dlq.py) | `dlq_messages` document and indexes | -| [`run_dlq_processor.py`](https://github.com/HardMax71/Integr8sCode/blob/main/backend/workers/run_dlq_processor.py) | Worker entry point | +| [`worker.py`](https://github.com/HardMax71/Integr8sCode/blob/main/backend/app/worker.py) | Worker entry point, schedules the retry job | diff --git a/docs/components/workers/event_replay.md b/docs/components/event-replay.md similarity index 84% rename from docs/components/workers/event_replay.md rename to docs/components/event-replay.md index 1a7a45938..95345ade5 100644 --- a/docs/components/workers/event_replay.md +++ b/docs/components/event-replay.md @@ -1,6 +1,6 @@ # Event replay -Replays don't run in this worker. They run inside the API process, in `EventReplayService`, and the event replay worker only cleans up after them: every 6 hours it calls `cleanup_old_sessions(older_than_hours=48)`, which deletes finished replay sessions older than 48 hours from the `replay_sessions` collection. +Replays run inside the API process, in `EventReplayService`. MongoDB cleans up after them: a TTL index on `completed_at` drops a finished session from the `replay_sessions` collection 48 hours after it ends, and `POST /api/v1/replay/cleanup` removes finished sessions past a given age on demand. ```mermaid graph LR @@ -8,7 +8,7 @@ graph LR API -->|read| Mongo[(events)] API -->|re-produce| Kafka[(Kafka)] API -->|progress| Redis["sse:replay:{session_id}"] - Worker[event-replay worker] -->|every 6h| Sessions[(replay_sessions)] + API -->|sessions| Sessions[(replay_sessions)] ``` ## Starting a replay @@ -29,7 +29,7 @@ Session state lives in a dict in the API process and is mirrored to `replay_sess | File | Purpose | |------|---------| -| [`run_event_replay.py`](https://github.com/HardMax71/Integr8sCode/blob/main/backend/workers/run_event_replay.py) | Worker: the cleanup job | +| [`db/docs/replay.py`](https://github.com/HardMax71/Integr8sCode/blob/main/backend/app/db/docs/replay.py) | `replay_sessions` document and its TTL index | | [`replay_service.py`](https://github.com/HardMax71/Integr8sCode/blob/main/backend/app/services/event_replay/replay_service.py) | Sessions, dispatch, targets | | [`replay.py`](https://github.com/HardMax71/Integr8sCode/blob/main/backend/app/api/routes/replay.py) | `/api/v1/replay` routes | | [`events.py`](https://github.com/HardMax71/Integr8sCode/blob/main/backend/app/api/routes/admin/events.py) | `/api/v1/admin/events/replay` routes | diff --git a/docs/components/schema-manager.md b/docs/components/schema-manager.md index 346001311..b67419da2 100644 --- a/docs/components/schema-manager.md +++ b/docs/components/schema-manager.md @@ -1,20 +1,21 @@ # MongoDB indexes and TTLs -There's no schema manager or migration step. Each Beanie document declares its own indexes, either as `Indexed(...)` fields or in `Settings.indexes`, and every process creates them at startup: the API in its lifespan and each worker in its entry point call `init_beanie` with `ALL_DOCUMENTS` from `app/db/docs/__init__.py`. Beanie sends every declared index to `create_indexes` on each boot, which MongoDB treats as a no-op for an index that already exists with the same spec. Repositories never create indexes. +There's no schema manager or migration step. Each Beanie document declares its own indexes, either as `Indexed(...)` fields or in `Settings.indexes`, and every process creates them at startup: the API in its lifespan and the worker in its entry point call `init_beanie` with `ALL_DOCUMENTS` from `app/db/docs/__init__.py`. Beanie sends every declared index to `create_indexes` on each boot, which MongoDB treats as a no-op for an index that already exists with the same spec. Repositories never create indexes. Beanie runs with index dropping off, so an index removed from a document stays in the database until someone drops it. Changing the options of an existing index makes MongoDB reject the create, and the process fails at startup. Drop the old index in mongosh first, then restart. ## TTLs -Three collections expire documents on their own: +Four collections expire documents on their own: | Collection | Expires | |------------|---------| | `events` | at `ttl_expires_at`, which `EventDocument` sets to 30 days after the event is created | | `notifications` | 30 days after `created_at` | | `dlq_messages` | 7 days after `created_at` | +| `replay_sessions` | 48 hours after `completed_at`; a session that hasn't finished has none and stays | -The 30 days for events is hardcoded in the document. `EVENT_RETENTION_DAYS` in config.toml is never read ([#293](https://github.com/HardMax71/Integr8sCode/issues/293)). Everything else stays until something deletes it. That includes `executions`, `sagas` and `resource_allocations`. The event replay worker deletes finished `replay_sessions` older than 48 hours. +The 30 days for events is hardcoded in the document. `EVENT_RETENTION_DAYS` in config.toml is never read ([#293](https://github.com/HardMax71/Integr8sCode/issues/293)). Everything else stays until something deletes it. That includes `executions`, `sagas` and `resource_allocations`. ## Unique indexes the code relies on diff --git a/docs/components/workers/dlq_processor.md b/docs/components/workers/dlq_processor.md index b1c3cbdca..ac48b53dd 100644 --- a/docs/components/workers/dlq_processor.md +++ b/docs/components/workers/dlq_processor.md @@ -1,6 +1,6 @@ # DLQ processor -The DLQ processor has no Kafka subscriber. It retries the failed events that the other workers' routers save to the `dlq_messages` collection. [Dead letter queue](../dead-letter-queue.md) covers how they get there. +The DLQ processor has no Kafka subscriber. It retries the failed events that the consumer routers, in the worker and the API, save to the `dlq_messages` collection. [Dead letter queue](../dead-letter-queue.md) covers how they get there. ```mermaid graph LR @@ -18,5 +18,5 @@ Retry policies are hardcoded in `app/dlq/models.py`, and the `DLQ_*` settings ar | File | Purpose | |------|---------| -| [`run_dlq_processor.py`](https://github.com/HardMax71/Integr8sCode/blob/main/backend/workers/run_dlq_processor.py) | Entry point and APScheduler job | +| [`worker.py`](https://github.com/HardMax71/Integr8sCode/blob/main/backend/app/worker.py) | Entry point and APScheduler job | | [`manager.py`](https://github.com/HardMax71/Integr8sCode/blob/main/backend/app/dlq/manager.py) | `process_due_retries`, `retry_message` | diff --git a/docs/components/workers/index.md b/docs/components/workers/index.md index 0d636c2c1..b98d66baf 100644 --- a/docs/components/workers/index.md +++ b/docs/components/workers/index.md @@ -1,8 +1,8 @@ -# Workers +# Worker -Six background processes sit next to the API, each in its own container built from the backend image. Four of them carry an execution from request to stored result. The other two run scheduled jobs: the DLQ processor polls for dead-lettered messages to retry (nothing writes any yet), and the event replay worker prunes old replay sessions. +One background process runs next to the API: `python -m app.worker`, in the `worker` container built from the backend image. Four parts of it carry an execution from request to stored result, and a fifth retries dead-lettered events. They share one broker connection and one Dishka container, but each consumer keeps its own Kafka consumer group, so every part still gets every event it subscribes to. -Workers talk to each other only through Kafka, one topic per event type, and the event names on the arrows below travel as those topics. MongoDB holds the durable state and Redis holds the execution queue and idempotency keys. +The parts talk to each other only through Kafka, one topic per event type, even inside the one process. The event names on the arrows below travel as those topics. MongoDB holds the durable state and Redis holds the execution queue and idempotency keys. ```mermaid graph LR @@ -17,29 +17,32 @@ graph LR Result --> Mongo[(MongoDB)] ``` -| Worker | What it does | Entry point | -|--------|--------------|-------------| -| [Saga orchestrator](saga_orchestrator.md) | Queues executions, runs the execution saga, sends pod commands | `run_saga_orchestrator.py` | -| [K8s worker](k8s_worker.md) | Creates the script ConfigMap and the executor pod | `run_k8s_worker.py` | -| [Pod monitor](pod_monitor.md) | Watches executor pods and publishes pod and execution events | `run_pod_monitor.py` | -| [Result processor](result_processor.md) | Writes the result onto the execution document | `run_result_processor.py` | -| [Event replay](event_replay.md) | Deletes replay sessions older than 48 hours, every 6 hours | `run_event_replay.py` | -| [DLQ processor](dlq_processor.md) | Republishes DLQ rows whose retry time has come | `run_dlq_processor.py` | +| Part | What it does | How it runs | +|------|--------------|-------------| +| [Saga orchestrator](saga_orchestrator.md) | Queues executions, runs the execution saga, sends pod commands | `saga-orchestrator` group; `try_schedule_from_queue` every 10s, `check_timeouts` every 30s | +| [K8s worker](k8s_worker.md) | Creates the script ConfigMap and the executor pod | `k8s-worker` group | +| [Pod monitor](pod_monitor.md) | Watches executor pods and publishes pod and execution events | watch cycle every 5s | +| [Result processor](result_processor.md) | Writes the result onto the execution document | `result-processor` group | +| [DLQ processor](dlq_processor.md) | Republishes DLQ rows whose retry time has come | job every 10s | -All entry points live in [`backend/workers/`](https://github.com/HardMax71/Integr8sCode/tree/main/backend/workers). Each one loads `config.toml` plus its own `config..toml` override, initializes Beanie, builds its Dishka container, registers its Kafka subscribers if it has any, and starts the FastStream app. +The entry point is [`backend/app/worker.py`](https://github.com/HardMax71/Integr8sCode/blob/main/backend/app/worker.py). It loads `config.toml` plus the `config.worker.toml` override, initializes Beanie, builds the container with `create_worker_container`, registers the three routers and starts the FastStream app. On startup it applies the executor NetworkPolicy and the image pre-puller DaemonSet, then starts the scheduler. + +Replays run in the API, not here. See [Event replay](../event-replay.md). ## Running locally -`./deploy.sh dev` starts all six. To debug one from your shell, stop its container and run it with uv from `backend/`: +`./deploy.sh dev` starts it. To debug it from your shell, stop the container and run it with uv from `backend/`: ```bash -docker compose stop saga-orchestrator +docker compose stop worker cd backend -uv run python -m workers.run_saga_orchestrator +uv run python -m app.worker ``` -The worker reads the same `config.toml` as the containers, so it connects to the compose hostnames `mongo`, `redis` and `kafka`. Map those to `127.0.0.1` in `/etc/hosts`, since compose publishes their ports on localhost. The k8s worker and pod monitor also read the kubeconfig from `KUBERNETES_CONFIG_PATH`, which is `/app/kubeconfig.yaml`, a path that exists only inside the containers. To run either one from the shell, set `KUBERNETES_CONFIG_PATH` in its `config..toml` to a kubeconfig the host can use, such as `~/.kube/config`. +The worker reads the same `config.toml` as the container, so it connects to the compose hostnames `mongo`, `redis` and `kafka`. Map those to `127.0.0.1` in `/etc/hosts`, since compose publishes their ports on localhost. It also reads the kubeconfig from `KUBERNETES_CONFIG_PATH`, which is `/app/kubeconfig.yaml`, a path that exists only inside the container. Set it in `config.worker.toml` to a kubeconfig the host can use, such as `~/.kube/config`. ## Scaling -Compose runs one container of each worker, with fixed container names. The pod monitor has to stay at one replica: it keeps the last seen phase of each pod in memory, and a second replica would watch the same pods and publish every event twice. The k8s worker and result processor consume through Kafka consumer groups and guard each handler with a Redis idempotency key, so extra replicas would split the partitions between them. The saga orchestrator keeps saga state in MongoDB and the queue in Redis, so both survive a restart, but nothing resumes a saga that was halfway through its steps when the process died. +Compose runs one worker, and it has to stay at one. The pod monitor keeps the last seen phase of each pod in memory, so a second replica would watch the same pods and publish every event twice, and the scheduled jobs would run once per replica. The consumers alone would scale, since they go through consumer groups and guard each handler with a Redis idempotency key. The saga orchestrator keeps saga state in MongoDB and the queue in Redis, so both survive a restart, but nothing resumes a saga that was halfway through its steps when the process died. + +A crash in any part restarts the whole container (`restart: unless-stopped`), and the consumers pick up from their committed offsets. diff --git a/docs/components/workers/k8s_worker.md b/docs/components/workers/k8s_worker.md index 44422c6cc..0ee8b45bc 100644 --- a/docs/components/workers/k8s_worker.md +++ b/docs/components/workers/k8s_worker.md @@ -50,7 +50,7 @@ Executor pods also carry the label `kueue.x-k8s.io/queue-name: executor-queue`, | File | Purpose | |------|---------| -| [`run_k8s_worker.py`](https://github.com/HardMax71/Integr8sCode/blob/main/backend/workers/run_k8s_worker.py) | Entry point | +| [`worker.py`](https://github.com/HardMax71/Integr8sCode/blob/main/backend/app/worker.py) | Entry point, startup setup | | [`worker.py`](https://github.com/HardMax71/Integr8sCode/blob/main/backend/app/services/k8s_worker/worker.py) | Command handlers, namespace policy, pre-puller | | [`pod_builder.py`](https://github.com/HardMax71/Integr8sCode/blob/main/backend/app/services/k8s_worker/pod_builder.py) | ConfigMap and pod spec | | [`runtime_registry.py`](https://github.com/HardMax71/Integr8sCode/blob/main/backend/app/runtime_registry.py) | Languages, images, per-language pod settings | diff --git a/docs/components/workers/pod_monitor.md b/docs/components/workers/pod_monitor.md index bfed8971c..21a2bc222 100644 --- a/docs/components/workers/pod_monitor.md +++ b/docs/components/workers/pod_monitor.md @@ -41,6 +41,6 @@ After publishing `execution_completed`, `execution_failed` or `execution_timeout | File | Purpose | |------|---------| -| [`run_pod_monitor.py`](https://github.com/HardMax71/Integr8sCode/blob/main/backend/workers/run_pod_monitor.py) | Entry point, watch scheduling, 410 handling | +| [`worker.py`](https://github.com/HardMax71/Integr8sCode/blob/main/backend/app/worker.py) | Entry point, watch scheduling, 410 handling | | [`monitor.py`](https://github.com/HardMax71/Integr8sCode/blob/main/backend/app/services/pod_monitor/monitor.py) | List-then-watch and pod deletion | | [`event_mapper.py`](https://github.com/HardMax71/Integr8sCode/blob/main/backend/app/services/pod_monitor/event_mapper.py) | Phase mapping and log parsing | diff --git a/docs/components/workers/result_processor.md b/docs/components/workers/result_processor.md index 69a11e50d..1dec98984 100644 --- a/docs/components/workers/result_processor.md +++ b/docs/components/workers/result_processor.md @@ -19,6 +19,6 @@ The result processor doesn't touch Kubernetes. The [pod monitor](pod_monitor.md) | File | Purpose | |------|---------| -| [`run_result_processor.py`](https://github.com/HardMax71/Integr8sCode/blob/main/backend/workers/run_result_processor.py) | Entry point | +| [`worker.py`](https://github.com/HardMax71/Integr8sCode/blob/main/backend/app/worker.py) | Entry point | | [`processor.py`](https://github.com/HardMax71/Integr8sCode/blob/main/backend/app/services/result_processor/processor.py) | Event handlers | | [`execution_repository.py`](https://github.com/HardMax71/Integr8sCode/blob/main/backend/app/db/repositories/execution_repository.py) | `write_terminal_result` | diff --git a/docs/components/workers/saga_orchestrator.md b/docs/components/workers/saga_orchestrator.md index 8d55147f9..d6c8e7aa6 100644 --- a/docs/components/workers/saga_orchestrator.md +++ b/docs/components/workers/saga_orchestrator.md @@ -50,7 +50,7 @@ The orchestrator consumes `execution_requested`, `execution_completed`, `executi | File | Purpose | |------|---------| -| [`run_saga_orchestrator.py`](https://github.com/HardMax71/Integr8sCode/blob/main/backend/workers/run_saga_orchestrator.py) | Entry point and APScheduler jobs | +| [`worker.py`](https://github.com/HardMax71/Integr8sCode/blob/main/backend/app/worker.py) | Entry point and APScheduler jobs | | [`saga_orchestrator.py`](https://github.com/HardMax71/Integr8sCode/blob/main/backend/app/services/saga/saga_orchestrator.py) | Queueing, step loop, compensation, timeouts | | [`execution_saga.py`](https://github.com/HardMax71/Integr8sCode/blob/main/backend/app/services/saga/execution_saga.py) | Steps and compensations | | [`execution_queue.py`](https://github.com/HardMax71/Integr8sCode/blob/main/backend/app/services/execution_queue.py) | Redis priority queue | diff --git a/docs/getting-started.md b/docs/getting-started.md index b299c8212..f221a1fe5 100644 --- a/docs/getting-started.md +++ b/docs/getting-started.md @@ -12,7 +12,7 @@ curl -sfL https://get.k3s.io | INSTALL_K3S_EXEC="--disable=traefik --bind-addres CI also passes `--disable-network-policy` to dodge the iptables race under [Troubleshooting](#troubleshooting). Leave it out here: without the policy controller, k3s doesn't enforce the NetworkPolicy that cuts executor pods off the network. -On startup the `cert-generator` container looks for the cluster through `~/.kube` and `/etc/rancher/k3s`. When it finds one, it creates the `integr8scode` namespace, installs Kueue, sets up a service account and writes `backend/kubeconfig.yaml` for the workers. When it can't connect, it logs "Cannot connect to Kubernetes cluster" and skips all of that. The rest of the stack still starts and you can log in, but the k8s worker and pod monitor have no kubeconfig, so scripts stay queued forever. +On startup the `cert-generator` container looks for the cluster through `~/.kube` and `/etc/rancher/k3s`. When it finds one, it creates the `integr8scode` namespace, installs Kueue, sets up a service account and writes `backend/kubeconfig.yaml` for the workers. When it can't connect, it logs "Cannot connect to Kubernetes cluster" and skips all of that. The rest of the stack still starts and you can log in, but the worker has no kubeconfig and keeps restarting, so scripts stay queued forever. ## Start the stack @@ -66,7 +66,7 @@ Kafdrop, Grafana and Jaeger aren't part of the default stack. `./deploy.sh dev - ## Troubleshooting -If scripts stay queued, check `docker compose logs cert-generator` for the connection error above, then `docker compose logs k8s-worker pod-monitor`. A k3s that stopped after a VPN or IP change is covered in [Deployment](operations/deployment.md). +If scripts stay queued, check `docker compose logs cert-generator` for the connection error above, then `docker compose logs worker`. A k3s that stopped after a VPN or IP change is covered in [Deployment](operations/deployment.md). If every service times out talking to mongo or kafka right after k3s started or restarted, k3s's policy sync probably wiped Docker's iptables rules ([kube-router#1248](https://github.com/cloudnativelabs/kube-router/issues/1248)). `docker compose down` followed by `./deploy.sh dev` recreates the network and its rules. diff --git a/docs/index.md b/docs/index.md index c4bc2af25..20a2d3385 100644 --- a/docs/index.md +++ b/docs/index.md @@ -27,7 +27,7 @@ Then open [https://localhost:5001](https://localhost:5001) and log in as `user` Every script runs in its own pod: non-root, read-only root filesystem, no capabilities, no service account token, and a NetworkPolicy that blocks all traffic in and out wherever the cluster's network plugin enforces it. CPU, memory and timeout limits are system-wide admin settings, and a language can raise them for its own pods. Status updates and the result reach the browser over Server-Sent Events. -The API and six workers talk through Kafka, one topic per event type. Every event is also stored in MongoDB for 30 days, where admins can browse and replay it. +The API and the worker talk through Kafka, one topic per event type. Every event is also stored in MongoDB for 30 days, where admins can browse and replay it. ## Documentation diff --git a/docs/operations/deployment.md b/docs/operations/deployment.md index ffc0ab7e2..6e5211c2b 100644 --- a/docs/operations/deployment.md +++ b/docs/operations/deployment.md @@ -1,6 +1,6 @@ # Deployment -Everything runs from one docker-compose.yaml: the backend, the frontend, six workers, MongoDB, Redis, Kafka and the optional observability stack. The workers reuse the backend image with a different `command:`, so there is one application image to build. Kubernetes only runs the executor pods; the backend and workers reach it through a kubeconfig that the cert-generator container writes. +Everything runs from one docker-compose.yaml: the backend, the frontend, the worker, MongoDB, Redis, Kafka and the optional observability stack. The worker reuses the backend image with a different `command:`, so there is one application image to build. Kubernetes only runs the executor pods; the backend and the worker reach it through a kubeconfig that the cert-generator container writes. `backend/Dockerfile.base` is a two-stage build on `python:3.14-slim`. The builder stage runs `uv sync --locked --no-dev --no-install-project --no-build`, and the final stage copies only the virtualenv plus a few runtime libraries. `backend/Dockerfile` adds the application code on top and starts gunicorn with uvicorn workers on port 443. The frontend image builds the Svelte app with `node:26-alpine` and serves it from `nginx:1.31-alpine`, see [Nginx configuration](nginx-configuration.md). @@ -22,7 +22,7 @@ Everything runs from one docker-compose.yaml: the backend, the frontend, six wor ./deploy.sh dev --debug # all of the above plus Kafdrop ``` -The core stack is MongoDB, Redis, Kafka (KRaft, no ZooKeeper), the backend, the frontend and the six workers. A few one-shot containers run first and exit: `shared-ca` prepares the shared CA volume, `cert-generator` writes TLS certs and `backend/kubeconfig.yaml`, and `user-seed` creates the default accounts. Kafka creates topics on first use (`KAFKA_AUTO_CREATE_TOPICS_ENABLE`). Topic names are the event types themselves, such as `execution_requested`, with no prefix. +The core stack is MongoDB, Redis, Kafka (KRaft, no ZooKeeper), the backend, the frontend and the worker. A few one-shot containers run first and exit: `shared-ca` prepares the shared CA volume, `cert-generator` writes TLS certs and `backend/kubeconfig.yaml`, and `user-seed` creates the default accounts. Kafka creates topics on first use (`KAFKA_AUTO_CREATE_TOPICS_ENABLE`). Topic names are the event types themselves, such as `execution_requested`, with no prefix. | Service | URL | Needs | |--------------------|-------------------------|-------------------------------| @@ -45,7 +45,7 @@ The seed job creates `user` / `user123` and `admin` / `admin123`, and every run DEFAULT_USER_PASSWORD=mypass ADMIN_USER_PASSWORD=myadmin ./deploy.sh dev ``` -There's no hot reload. Gunicorn runs without `--reload` and the frontend is an nginx production build. The backend and workers bind-mount `backend/app` and `backend/workers`, so a Python change needs `docker compose restart backend` (and the affected workers). A frontend change needs `docker compose up -d --build frontend`. +There's no hot reload. Gunicorn runs without `--reload` and the frontend is an nginx production build. The backend and the worker bind-mount `backend/app`, so a Python change needs `docker compose restart backend worker`. A frontend change needs `docker compose up -d --build frontend`. Data lives in named volumes: @@ -67,7 +67,7 @@ Data lives in named volumes: | Grafana | `curl localhost:3000/api/health` | | Victoria Metrics | `wget http://127.0.0.1:8428/-/healthy` | -Loki has none because its image is distroless and has no shell, so Grafana and the collector wait for `service_started`. Jaeger, the OTel Collector, Kafdrop and the workers have no healthcheck either. +Loki has none because its image is distroless and has no shell, so Grafana and the collector wait for `service_started`. Jaeger, the OTel Collector, Kafdrop and the worker have no healthcheck either. ## Memory limits @@ -80,7 +80,7 @@ Every long-running service except Kafdrop has a `mem_limit`. | Kafka | 1280m | `-Xms256m -Xmx1g` | | Backend | 768m | `WEB_CONCURRENCY` gunicorn workers (default 2) | | Frontend | 128m | | -| Each of six workers | 160m | | +| Worker | 256m | idles near 150 MiB | | Grafana | 512m | Grafana 13 idles near 300 MiB | | Jaeger | 256m | in-memory storage | | Loki | 256m | | @@ -125,7 +125,7 @@ docker compose exec mongo sh -c 'mongosh -u "$MONGO_INITDB_ROOT_USERNAME" -p "$M |-----------------------|------------------------------------|-------------------------------------------------------------| | Unknown topic errors | Kafka not ready yet | `docker compose logs kafka` | | MongoDB auth errors | Volume created with other credentials | `MONGO_USER`/`MONGO_PASSWORD` must match what the volume was initialized with | -| Workers can't reach k8s | Stale kubeconfig after the cluster was reinstalled or its API address changed | `docker compose restart cert-generator`, then restart the workers | +| Worker can't reach k8s | Stale kubeconfig after the cluster was reinstalled or its API address changed | `docker compose restart cert-generator`, then restart the worker | List topics from inside the broker: @@ -145,16 +145,16 @@ sudo systemctl stop k3s sudo rm -rf /var/lib/rancher/k3s/server/db/ /var/lib/rancher/k3s/server/tls/ sudo systemctl start k3s -# regenerate the kubeconfig and reconnect the workers +# regenerate the kubeconfig and reconnect the worker docker compose restart cert-generator -docker compose restart k8s-worker pod-monitor +docker compose restart worker ``` The cluster comes back empty, so Kueue and the `integr8scode` namespace are recreated by cert-generator on that restart. With NordVPN, `nordvpn set lan-discovery enabled` keeps the host reachable from containers. See [k3s#277](https://github.com/k3s-io/k3s/issues/277) and the [k3s certificate docs](https://docs.k3s.io/cli/certificate). ### k3s certificates expired -k3s issues its certificates for a year and renews them only when it starts. On a server that runs longer than that without a restart, kubectl and the workers fail with `x509: certificate has expired`, and `sudo systemctl restart k3s` fixes it. If k3s then hangs in `activating` and its journal repeats `dynamiclistener [::]:6443: failed to update cert with connection local address: EOF`, the cached serving certificate in /var/lib/rancher/k3s/server/tls/dynamic-cert.json is empty, which happens when the disk fills up. Move that file aside and start k3s again; it writes a new one. +k3s issues its certificates for a year and renews them only when it starts. On a server that runs longer than that without a restart, kubectl and the worker fail with `x509: certificate has expired`, and `sudo systemctl restart k3s` fixes it. If k3s then hangs in `activating` and its journal repeats `dynamiclistener [::]:6443: failed to update cert with connection local address: EOF`, the cached serving certificate in /var/lib/rancher/k3s/server/tls/dynamic-cert.json is empty, which happens when the disk fills up. Move that file aside and start k3s again; it writes a new one. ## Pre-built images diff --git a/docs/operations/logging.md b/docs/operations/logging.md index ff3790673..20a52149d 100644 --- a/docs/operations/logging.md +++ b/docs/operations/logging.md @@ -13,7 +13,7 @@ flowchart LR ## Setup -[`setup_logger`](https://github.com/HardMax71/Integr8sCode/blob/main/backend/app/core/logging.py) configures structlog and the root logger. The backend calls it in `create_app`, the DI container calls it for the logger it injects, and each worker calls it at the top of `main()`. +[`setup_logger`](https://github.com/HardMax71/Integr8sCode/blob/main/backend/app/core/logging.py) configures structlog and the root logger. The backend calls it in `create_app`, the DI container calls it for the logger it injects, and the worker calls it at the top of `main()`. ```python --8<-- "backend/app/core/logging.py:setup_logger" @@ -70,7 +70,7 @@ That keeps fields queryable, and it stops log injection. If user input goes into ## OTLP export -`setup_log_exporter` adds an OTel `LoggingHandler` to the root logger, with a `BatchLogRecordProcessor` and a gRPC `OTLPLogExporter` pointed at `OTEL_EXPORTER_OTLP_ENDPOINT`. It's a no-op when that setting is empty. The backend calls it in `create_app` and each worker right after `setup_logger`. See [Loki](loki.md) for the rest of the path. +`setup_log_exporter` adds an OTel `LoggingHandler` to the root logger, with a `BatchLogRecordProcessor` and a gRPC `OTLPLogExporter` pointed at `OTEL_EXPORTER_OTLP_ENDPOINT`. It's a no-op when that setting is empty. The backend calls it in `create_app` and the worker right after `setup_logger`. See [Loki](loki.md) for the rest of the path. ## Frontend diff --git a/docs/operations/loki.md b/docs/operations/loki.md index 24ddacb4b..731f88231 100644 --- a/docs/operations/loki.md +++ b/docs/operations/loki.md @@ -1,17 +1,17 @@ # Log aggregation (Loki) -Loki stores the logs of the backend and all six workers so you can query them in Grafana across processes and time. It runs only with the `observability` compose profile. Without it, logs stay on container stderr for `docker compose logs`. +Loki stores the logs of the backend and the worker so you can query them in Grafana across processes and time. It runs only with the `observability` compose profile. Without it, logs stay on container stderr for `docker compose logs`. ```mermaid flowchart LR - Proc["backend and workers"] -->|"OTLP gRPC"| Collector["OTel Collector :4317"] + Proc["backend and worker"] -->|"OTLP gRPC"| Collector["OTel Collector :4317"] Collector -->|"otlp_http/loki"| Loki["Loki :3100/otlp"] Loki --> Grafana ``` ## Export from Python -`setup_log_exporter` in [`app/core/logging.py`](https://github.com/HardMax71/Integr8sCode/blob/main/backend/app/core/logging.py) builds an OTel `LoggerProvider` with a `BatchLogRecordProcessor` and a gRPC `OTLPLogExporter`, then adds a `LoggingHandler` to the root logger. The handler carries the same `RedactingFilter` as the stderr handler, so only redacted text leaves the process. The backend calls it in `create_app`, and each worker calls it right after `setup_logger` in its `main()`. +`setup_log_exporter` in [`app/core/logging.py`](https://github.com/HardMax71/Integr8sCode/blob/main/backend/app/core/logging.py) builds an OTel `LoggerProvider` with a `BatchLogRecordProcessor` and a gRPC `OTLPLogExporter`, then adds a `LoggingHandler` to the root logger. The handler carries the same `RedactingFilter` as the stderr handler, so only redacted text leaves the process. The backend calls it in `create_app`, and the worker calls it right after `setup_logger` in its `main()`. The only switch is `OTEL_EXPORTER_OTLP_ENDPOINT`: empty means no export. config.toml sets it to `http://otel-collector:4317`; config.test.toml leaves it unset. @@ -37,7 +37,7 @@ The container has a 256m memory limit, keeps data in the `loki_data` volume and Two things don't work as intended today ([#294](https://github.com/HardMax71/Integr8sCode/issues/294)). -Every process logs as `service_name="integr8scode-backend"`. The per-worker `config..toml` files only override `TRACING_SERVICE_NAME`, not `SERVICE_NAME`, and the collector's `resource` processor copies from a `service_name` attribute that doesn't exist. So the API and all six workers share one `service_name` label (Loki tells their streams apart by `service_instance_id`), and a selector such as `{service_name="integr8scode-saga-orchestrator"}` returns nothing. +Every process logs as `service_name="integr8scode-backend"`. `config.worker.toml` only overrides `TRACING_SERVICE_NAME`, not `SERVICE_NAME`, and the collector's `resource` processor copies from a `service_name` attribute that doesn't exist. So the API and the worker share one `service_name` label (Loki tells their streams apart by `service_instance_id`), and a selector such as `{service_name="integr8scode-worker"}` returns nothing. The Loki datasource has a derived field that should link `trace_id` to Jaeger, but its regex is `"trace_id":"(\w+)"` and structlog writes `"trace_id": "..."` with a space after the colon. The link never appears. diff --git a/docs/operations/metrics-reference.md b/docs/operations/metrics-reference.md index 8637b806e..1ae987e4f 100644 --- a/docs/operations/metrics-reference.md +++ b/docs/operations/metrics-reference.md @@ -1,6 +1,6 @@ # Metrics Reference -Only the backend (API) process exports metrics today. It pushes them over OTLP to the OpenTelemetry Collector every 60 seconds, and the collector writes them to VictoriaMetrics. The six workers record into the same metric classes but never install a meter provider, so their numbers never leave the process ([#289](https://github.com/HardMax71/Integr8sCode/issues/289)). Every table below says which process records a metric, so you can tell which panels stay empty until that lands. +Only the backend (API) process exports metrics today. It pushes them over OTLP to the OpenTelemetry Collector every 60 seconds, and the collector writes them to VictoriaMetrics. The worker records into the same metric classes but never installs a meter provider, so their numbers never leave the process ([#289](https://github.com/HardMax71/Integr8sCode/issues/289)). Every table below says which process records a metric, so you can tell which panels stay empty until that lands. ## Export path @@ -10,10 +10,10 @@ flowchart LR Receivers["host_metrics, mongodb, kafka_metrics receivers"] --> Collector Collector -->|"prometheus remote write"| VM["VictoriaMetrics"] VM --> Grafana - Workers["k8s-worker, pod-monitor, result-processor, saga-orchestrator, event-replay, dlq-processor"] -.->|"no meter provider, issue 289"| Dropped["no-op"] + Worker["worker"] -.->|"no meter provider, issue 289"| Dropped["no-op"] ``` -`create_app` calls `setup_metrics` (`app/core/middlewares/metrics.py`) once. It installs a global `MeterProvider` whose `PeriodicExportingMetricReader` exports every 60 000 ms through an OTLP gRPC exporter. The exporter targets `OTEL_EXPORTER_OTLP_ENDPOINT`, which `config.toml` sets to `http://otel-collector:4317`. Without an endpoint it returns early and the no-op provider stays. Each metrics class takes its meter from the global provider when DI builds it, which is why a worker that skips `setup_metrics` records into nothing: +`create_app` calls `setup_metrics` (`app/core/middlewares/metrics.py`) once. It installs a global `MeterProvider` whose `PeriodicExportingMetricReader` exports every 60 000 ms through an OTLP gRPC exporter. The exporter targets `OTEL_EXPORTER_OTLP_ENDPOINT`, which `config.toml` sets to `http://otel-collector:4317`. Without an endpoint it returns early and the no-op provider stays. Each metrics class takes its meter from the global provider when DI builds it, which is why the worker, which skips `setup_metrics`, records into nothing: ```python --8<-- "backend/app/core/metrics/base.py:init" @@ -56,20 +56,20 @@ Gauges listed as "in process" are up-down counters the process keeps itself. The | Metric | Type | Labels | Recorded in | |--------|------|--------|-------------| -| `script_executions_total` | counter | status, lang_and_version | `ExecutionService.execute_script` in backend (queued, error); `ResultProcessor` in result-processor (completed, failed, timeout) | +| `script_executions_total` | counter | status, lang_and_version | `ExecutionService.execute_script` in backend (queued, error); `ResultProcessor` in worker (completed, failed, timeout) | | `script_submission_duration_seconds` | histogram | lang_and_version | `ExecutionService.execute_script` in backend: persisting the execution and publishing the request | -| `script_errors_total` | counter | error_type | `ExecutionService` in backend when the publish fails; `ResultProcessor` in result-processor | -| `script_execution_duration_seconds` | histogram | lang_and_version | `ResultProcessor` in result-processor: wall time from the pod's resource usage, or the timeout for timed out runs | -| `script_memory_usage_MiB` | histogram | lang_and_version | `ResultProcessor` in result-processor: peak memory | -| `script_memory_utilization_percent` | histogram | lang_and_version | `ResultProcessor` in result-processor: peak memory over the effective memory limit for the language (admin setting or TOML, raised to the language minimum) | -| `queue_depth` | gauge, in process | none | `ExecutionQueueService` in saga-orchestrator: enqueued minus scheduled | -| `queue_active` | gauge, in process | none | `ExecutionQueueService` in saga-orchestrator: scheduled minus released | -| `queue_enqueue_total` | counter | none | `ExecutionQueueService.enqueue` in saga-orchestrator | -| `queue_schedule_total` | counter | none | `ExecutionQueueService.try_schedule` in saga-orchestrator | -| `queue_wait_time_seconds` | histogram | priority | `ExecutionQueueService.try_schedule` in saga-orchestrator: enqueue to schedule | -| `queue_event_data_lost_total` | counter | none | `ExecutionQueueService.try_schedule` in saga-orchestrator, when the queued event expired in Redis | - -The queue itself lives in Redis, so after a saga-orchestrator restart `queue_depth` and `queue_active` read zero while work is still queued. +| `script_errors_total` | counter | error_type | `ExecutionService` in backend when the publish fails; `ResultProcessor` in worker | +| `script_execution_duration_seconds` | histogram | lang_and_version | `ResultProcessor` in worker: wall time from the pod's resource usage, or the timeout for timed out runs | +| `script_memory_usage_MiB` | histogram | lang_and_version | `ResultProcessor` in worker: peak memory | +| `script_memory_utilization_percent` | histogram | lang_and_version | `ResultProcessor` in worker: peak memory over the effective memory limit for the language (admin setting or TOML, raised to the language minimum) | +| `queue_depth` | gauge, in process | none | `ExecutionQueueService` in worker: enqueued minus scheduled | +| `queue_active` | gauge, in process | none | `ExecutionQueueService` in worker: scheduled minus released | +| `queue_enqueue_total` | counter | none | `ExecutionQueueService.enqueue` in worker | +| `queue_schedule_total` | counter | none | `ExecutionQueueService.try_schedule` in worker | +| `queue_wait_time_seconds` | histogram | priority | `ExecutionQueueService.try_schedule` in worker: enqueue to schedule | +| `queue_event_data_lost_total` | counter | none | `ExecutionQueueService.try_schedule` in worker, when the queued event expired in Redis | + +The queue itself lives in Redis, so after a worker restart `queue_depth` and `queue_active` read zero while work is still queued. ### Events and idempotency @@ -84,7 +84,7 @@ The queue itself lives in Redis, so after a saga-orchestrator restart `queue_dep | `idempotency_processing_duration_seconds` | histogram | event_type, status | `IdempotencyManager`: reserving the key to marking it completed or failed | | `idempotency_keys_active` | gauge, observed at export | none | `IdempotencyMetrics` scans `idempotency:*` in Redis | -Every Kafka publish also shows up in `messaging_publish_messages_message_total`; a publish that fails is marked on the stored event and not counted here. `event_category` equals `event_type` for the current event names. A handler that raises has no counter of its own: it's a `messaging_process_duration_seconds` sample with an `error_type` label, in the FastStream families below. `IdempotencyManager` runs in backend for `Idempotency-Key` requests and in the k8s-worker, result-processor and saga-orchestrator consumers. +Every Kafka publish also shows up in `messaging_publish_messages_message_total`; a publish that fails is marked on the stored event and not counted here. `event_category` equals `event_type` for the current event names. A handler that raises has no counter of its own: it's a `messaging_process_duration_seconds` sample with an `error_type` label, in the FastStream families below. `IdempotencyManager` runs in backend for `Idempotency-Key` requests and in the worker's k8s worker, result processor and saga orchestrator consumers. ### Server-sent events (backend) @@ -102,17 +102,17 @@ The status snapshot a stream sends first comes from the database, not Redis, and | Metric | Type | Labels | Recorded in | |--------|------|--------|-------------| -| `pod_creations_total` | counter | status: success, failed; language | `KubernetesWorker` in k8s-worker | -| `pod_creation_duration_seconds` | histogram | language | `KubernetesWorker` in k8s-worker: ConfigMap, pod and `PodCreated` publish, successful creations only | -| `pod_creations_active` | gauge, in process | none | `KubernetesWorker` in k8s-worker | -| `configmaps_created_total` | counter | status: success, already_exists, failed | `KubernetesWorker` in k8s-worker | -| `pods_by_phase` | gauge | phase | `PodMonitor` in pod-monitor: executor pods seen through LIST and watch; a relist rebuilds it | -| `pod_lifetime_seconds` | histogram | final_phase, language | `PodMonitor` in pod-monitor: pod creation to the terminal event, recorded when the monitor deletes the pod | -| `pod_monitor_events_total` | counter | event_type, pod_phase | `PodMonitor` in pod-monitor: each application event it publishes | -| `pod_monitor_processing_duration_seconds` | histogram | event_type: ADDED, MODIFIED, DELETED | `PodMonitor` in pod-monitor: mapping and publishing one watch event | -| `pod_monitor_reconciliations_total` | counter | none | `PodMonitor` in pod-monitor: each full LIST, on start and after 410 Gone | -| `pod_monitor_watch_errors_total` | counter | error_type: resource_version_expired, api_error, unexpected, processing_error | `PodMonitor` and `workers/run_pod_monitor.py` in pod-monitor | -| `pod_monitor_watch_reconnects_total` | counter | none | `workers/run_pod_monitor.py` in pod-monitor, after a failed watch cycle | +| `pod_creations_total` | counter | status: success, failed; language | `KubernetesWorker` in worker | +| `pod_creation_duration_seconds` | histogram | language | `KubernetesWorker` in worker: ConfigMap, pod and `PodCreated` publish, successful creations only | +| `pod_creations_active` | gauge, in process | none | `KubernetesWorker` in worker | +| `configmaps_created_total` | counter | status: success, already_exists, failed | `KubernetesWorker` in worker | +| `pods_by_phase` | gauge | phase | `PodMonitor` in worker: executor pods seen through LIST and watch; a relist rebuilds it | +| `pod_lifetime_seconds` | histogram | final_phase, language | `PodMonitor` in worker: pod creation to the terminal event, recorded when the monitor deletes the pod | +| `pod_monitor_events_total` | counter | event_type, pod_phase | `PodMonitor` in worker: each application event it publishes | +| `pod_monitor_processing_duration_seconds` | histogram | event_type: ADDED, MODIFIED, DELETED | `PodMonitor` in worker: mapping and publishing one watch event | +| `pod_monitor_reconciliations_total` | counter | none | `PodMonitor` in worker: each full LIST, on start and after 410 Gone | +| `pod_monitor_watch_errors_total` | counter | error_type: resource_version_expired, api_error, unexpected, processing_error | `PodMonitor` and `app/worker.py` in worker | +| `pod_monitor_watch_reconnects_total` | counter | none | `app/worker.py` in worker, after a failed watch cycle | ### Notifications (backend) @@ -133,7 +133,7 @@ The status snapshot a stream sends first comes from the database, not Redis, and ### Event replay (backend) -`EventReplayService`. The replay routes drive it inside backend. The event-replay worker only runs session cleanup against its own session map, which stays empty. +`EventReplayService`. The replay routes drive it inside backend. A TTL index in MongoDB drops finished sessions, so nothing outside the API touches them. | Metric | Type | Labels | Notes | |--------|------|--------|-------| @@ -180,7 +180,7 @@ The status snapshot a stream sends first comes from the database, not Redis, and | `rate_limit_rejected_total` | counter | endpoint, group | | | `rate_limit_bypass_total` | counter | endpoint | users with `bypass_rate_limit` | -### Dead letter queue (dlq-processor) +### Dead letter queue (worker) `DLQManager`, driven by APScheduler every 10 seconds. @@ -194,7 +194,7 @@ The status snapshot a stream sends first comes from the database, not Redis, and | `dlq_message_age_seconds` | histogram | none | `handle_message` | | `dlq_processing_errors_total` | counter | original_topic, event_type, error_type | nothing records it | -`handle_message` runs wherever a router's `DeadLetterMiddleware` catches a failure: backend for notifications, and the k8s-worker, result-processor and saga-orchestrator. Only backend's numbers leave the process until workers export ([#289](https://github.com/HardMax71/Integr8sCode/issues/289)). `dlq_processing_errors_total` was only recorded by a batch retry method that had no callers and is gone, so it stays empty. +`handle_message` runs wherever a router's `DeadLetterMiddleware` catches a failure: backend for notifications, and the k8s worker, result processor and saga orchestrator groups in the worker. Only backend's numbers leave the process until the worker exports ([#289](https://github.com/HardMax71/Integr8sCode/issues/289)). `dlq_processing_errors_total` was only recorded by a batch retry method that had no callers and is gone, so it stays empty. ## FastStream messaging families @@ -251,4 +251,4 @@ sum by (reason) (rate(token_validation_failures_total[5m])) sum(rate(mongodb_cache_operations_total{type="hit"}[5m])) / sum(rate(mongodb_cache_operations_total[5m])) ``` -Once workers export, `sum(pods_by_phase{phase="Pending"})` shows executor pods waiting for a node or a Kueue slot. +Once the worker exports, `sum(pods_by_phase{phase="Pending"})` shows executor pods waiting for a node or a Kueue slot. diff --git a/docs/operations/tracing.md b/docs/operations/tracing.md index 0157b3838..b5b333ec6 100644 --- a/docs/operations/tracing.md +++ b/docs/operations/tracing.md @@ -1,6 +1,6 @@ # Tracing -The backend and every worker export OpenTelemetry traces straight to Jaeger. A request to the API and the Kafka messages it causes show up as one trace, as long as each hop runs inside a span. +The backend and the worker export OpenTelemetry traces straight to Jaeger. A request to the API and the Kafka messages it causes show up as one trace, as long as each hop runs inside a span. ## Setup @@ -12,11 +12,11 @@ The backend and every worker export OpenTelemetry traces straight to Jaeger. A r - sets the W3C TraceContext propagator - instruments HTTPX through `HTTPX2ClientInstrumentor` (the backend uses httpx2), PyMongo, and stdlib logging -The backend resolves `Tracer` in `dishka_lifespan` and then instruments FastAPI with `FastAPIInstrumentor`, skipping `health`, `metrics`, `docs` and openapi.json. Workers never ask for it directly, but the broker provider depends on it, so every process that builds a `KafkaBroker` sets up tracing first. +The backend resolves `Tracer` in `dishka_lifespan` and then instruments FastAPI with `FastAPIInstrumentor`, skipping `health`, `metrics`, `docs` and openapi.json. The worker never asks for it directly, but the broker provider depends on it, so every process that builds a `KafkaBroker` sets up tracing first. config.toml sends traces to `http://jaeger:4317` with a sampling rate of 1.0. The collector also has a traces pipeline (with tail sampling) into Jaeger, but the app doesn't use it. config.test.toml leaves the endpoint empty, so tests create spans and export nothing, and pytest sets `OTEL_SDK_DISABLED=true` on top. -Each worker's `config..toml` overrides `TRACING_SERVICE_NAME`, so Jaeger lists `integr8scode-backend`, `k8s-worker`, `pod-monitor`, `result-processor`, `saga-orchestrator`, `event-replay` and `dlq-processor` as separate services. +`config.worker.toml` overrides `TRACING_SERVICE_NAME`, so Jaeger lists two services: `integr8scode-backend` and `integr8scode-worker`. ## Kafka propagation @@ -43,8 +43,8 @@ sequenceDiagram participant Client participant API as Backend participant Kafka - participant Saga as saga-orchestrator - participant K8s as k8s-worker + participant Saga as Worker: saga orchestrator + participant K8s as Worker: k8s worker Client->>API: POST /api/v1/execute Note over API: server span diff --git a/docs/reference/configuration.md b/docs/reference/configuration.md index 3e904acf6..eba6b5008 100644 --- a/docs/reference/configuration.md +++ b/docs/reference/configuration.md @@ -3,13 +3,13 @@ The backend reads its non-secret configuration from TOML files, not from environment variables. `Settings` in `app/settings.py` builds it in this order, each layer overriding the one before: 1. `config.toml`, the base file committed to git -2. an optional per-worker override such as `config.saga-orchestrator.toml` +2. an optional override, `config.worker.toml` for the worker 3. four secrets from the environment: `SECRET_KEY`, `MONGO_USER`, `MONGO_PASSWORD` and `REDIS_PASSWORD` ```python Settings() # config.toml + secrets Settings(config_path="config.test.toml") # tests -Settings(override_path="config.saga-orchestrator.toml") # a worker +Settings(override_path="config.worker.toml") # the worker ``` `Settings` uses `extra="forbid"`, so a key that isn't a field below stops the process at startup. `OTEL_SERVICE_NAME`, `OTEL_SERVICE_VERSION` and `OTEL_RESOURCE_ATTRIBUTES` are not fields: put them in TOML and nothing starts. @@ -115,7 +115,7 @@ A language can raise its own pod's limits above the system settings. Go gets at |-----|-------|--------|--------------| | `OTLP_TRACES_ENDPOINT` | `http://jaeger:4317` | config.toml | OTLP gRPC endpoint for traces; empty turns export off (the default is empty) | | `TRACING_SAMPLING_RATE` | `1.0` | config.toml | Fraction of traces kept, 0.0 to 1.0 (the default is 0.1) | -| `TRACING_SERVICE_NAME` | `integr8scode-backend` | config.toml | `service.name` on spans; each worker override sets its own | +| `TRACING_SERVICE_NAME` | `integr8scode-backend` | config.toml | `service.name` on spans; `config.worker.toml` sets `integr8scode-worker` | | `TRACING_SERVICE_VERSION` | `1.0.0` | config.toml | `service.version` on spans | | `OTEL_EXPORTER_OTLP_ENDPOINT` | `http://otel-collector:4317` | config.toml | OTLP endpoint for metrics and logs; unset turns both off | | `SERVICE_NAME` | `integr8scode-backend` | config.toml | `service.name` on metrics and logs, Kafka client id, event metadata | @@ -186,20 +186,9 @@ These have no entry in `config.toml`, so the `Settings` default applies. `APP_URL` (`https://localhost` in config.toml) is unused too. `CORS_ORIGINS` is the list in config.toml, which replaces the `Settings` default list. -## Worker overrides +## Worker override -Each worker loads a small override next to `config.toml`: - -| File | Worker | -|------|--------| -| `config.k8s-worker.toml` | K8s worker | -| `config.pod-monitor.toml` | Pod monitor | -| `config.result-processor.toml` | Result processor | -| `config.saga-orchestrator.toml` | Saga orchestrator | -| `config.event-replay.toml` | Event replay | -| `config.dlq-processor.toml` | DLQ processor | - -Each sets `TRACING_SERVICE_NAME` and `KAFKA_CONSUMER_GROUP_ID` to the worker's name. Only the first one has an effect. +The worker loads `config.worker.toml` next to `config.toml`. It sets one key, `TRACING_SERVICE_NAME = "integr8scode-worker"`, so Jaeger shows the worker apart from the API. ## Test configuration diff --git a/docs/testing/kafka-test-stability.md b/docs/testing/kafka-test-stability.md index 32ce4dc04..c57afbe09 100644 --- a/docs/testing/kafka-test-stability.md +++ b/docs/testing/kafka-test-stability.md @@ -4,7 +4,7 @@ Most of what used to make Kafka tests flaky is gone. What's left is how long con ## No producer to manage -`UnifiedProducer` is a store-then-publish outbox. It writes the event to MongoDB, then publishes through `KafkaEventTransport`, which calls `broker.publish()`. If the publish fails, it marks the stored event as failed to publish and re-raises. It has no Kafka client of its own and nothing to start or stop. The one `KafkaBroker` per process (aiokafka 0.14 under FastStream 0.7.7) is started by `dishka_lifespan` in the API and by FastStream's app runner in the workers. +`UnifiedProducer` is a store-then-publish outbox. It writes the event to MongoDB, then publishes through `KafkaEventTransport`, which calls `broker.publish()`. If the publish fails, it marks the stored event as failed to publish and re-raises. It has no Kafka client of its own and nothing to start or stop. The one `KafkaBroker` per process (aiokafka 0.14 under FastStream 0.7.7) is started by `dishka_lifespan` in the API and by FastStream's app runner in the worker. Unit tests don't touch Kafka. The E2E suite runs inside the backend container, and its session-scoped `app` fixture enters the real lifespan. So each xdist worker starts one broker with the SSE and notification subscribers, and stops it when the session ends. diff --git a/mkdocs.yml b/mkdocs.yml index cd806fa54..ee41d1321 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -147,17 +147,17 @@ nav: - SSE: - Architecture: components/sse/sse-architecture.md - Execution Flow: components/sse/execution-sse-flow.md - - Workers: + - Worker: - Overview: components/workers/index.md - Saga Orchestrator: components/workers/saga_orchestrator.md - K8s Worker: components/workers/k8s_worker.md - Pod Monitor: components/workers/pod_monitor.md - Result Processor: components/workers/result_processor.md - - Event Replay: components/workers/event_replay.md - DLQ Processor: components/workers/dlq_processor.md - Saga: - Resource Allocation: components/saga/resource-allocation.md - Dead Letter Queue: components/dead-letter-queue.md + - Event Replay: components/event-replay.md - MongoDB Indexes: components/schema-manager.md - Saved Scripts: components/saved-scripts.md