Frequently Asked Questions¶

A living Q&A page drawing from real operator questions and common first-time-reader confusions. Answers are deliberately short — click through for depth. Contributions welcome.

About Drakkar¶

What is Drakkar?¶

A Python framework for building Kafka → subprocess → multi-sink pipelines. You write a handler describing how to turn input messages into executor-task specs; the framework polls Kafka, runs your binaries with backpressure, captures results, and fans them out to one or more sinks (Kafka, Postgres, Mongo, HTTP, Redis, Filesystem). See Architecture and Data Flow.

What workloads is Drakkar designed for?¶

Streaming CPU-bound work done by external CLI tools or compiled binaries, with async Python handling the I/O on either side. The shape of a good fit is:

Ingest: messages arrive on Kafka (async I/O).
Prep (arrange()): optional async lookups — cache, small DB queries, feature-flag checks — that build per-task payloads without blocking the loop.
Work: each task spawns a subprocess running your CPU-heavy binary (ripgrep, a Rust/Go/C++ analyzer, a numerical tool). This is where the CPU goes — outside Python, so the GIL is irrelevant and N subprocesses truly use N cores.
Collect (on_task_complete / on_message_complete / on_window_complete): async, builds sink records.
Emit: records fan out in parallel to one or more sinks (async I/O).

The ExecutorPool semaphore sits at the center, throttling concurrent subprocesses to max_executors so you saturate CPU without oversubscribing it. Backpressure pauses the Kafka consumer when the pipeline is already full. See Performance recommendations and Executor System.

What is Drakkar NOT a good fit for?¶

Pure async-Python workloads with no subprocess stage — a plain aiokafka consumer + async sinks is simpler; Drakkar’s subprocess machinery becomes pure overhead.
Very short tasks where the per-task subprocess launch overhead (~10ms) dominates the actual work. If your task is ≤ 1ms of real work, use precomputed tasks or in-process logic. See Bottleneck: Subprocess Launch.
Ultra-low-latency pipelines (single-digit ms end-to-end) — windowing and batching cost is real.
Exactly-once transactional streams — Drakkar is at-least-once with DLQ safety; see the delivery section below.

How does Drakkar compare to Celery, Faust, Kafka Streams, Benthos?¶

Drakkar is narrower and more opinionated than any of them — the sweet spot is Kafka input + CPU-heavy external binary + multi-sink fan-out with strict per-partition ordering. Rough shape of how the others differ:

Tool	Input sources	Work stage	Python-native?	Notes
Drakkar	Kafka only	Subprocess pool (external binary via stdin/stdout)	Yes (3.13+)	Built-in DLQ, circuit breakers, flight recorder, a cache shared across workers, live operator UI. Offsets commit per-partition after all sinks ack.
Celery	Broker-agnostic (Redis, RabbitMQ, SQS, Kafka via plugin)	In-process Python function	Yes	General task queue; no native Kafka partition ordering; one broker at a time per worker.
Faust	Kafka only	In-process async Python (streams & tables)	Yes	Stream-processing DSL (agents, tables, windows). No first-class subprocess executor; you bring your own.
Kafka Streams	Kafka only	In-process JVM code	No (Java/Scala)	Stateful stream processor with changelog-backed stores. No Python.
Benthos (now Bento/Redpanda Connect)	~50 sources	Config-driven processors (Go)	No	YAML pipeline DSL; excellent fan-out; no user-written Python; handlers are Bloblang/WASM.

Use Drakkar when your handler binary is Rust/Go/C++/a CLI you don’t own, you need per-partition ordering, and you want multi-sink delivery (Kafka + Postgres + HTTP + …) with at-least-once guarantees and a DLQ. Use one of the others when your work is pure in-process Python (Faust), pure Go/config (Benthos), or not Kafka-first (Celery).

What languages can my handler’s executor binary be written in?¶

Any language. The executor launches a subprocess and communicates via stdin/stdout/exit-code — the binary only needs to read a JSON payload from stdin and write results to stdout. See Executor System.

Why Python 3.13+?¶

The floor is driven by PEP 695 generic syntax used across the codebase (e.g. async def get[T: BaseModel](...) in drakkar/cache_memory.py). That syntax parses only on 3.12+, and Drakkar stabilized it on 3.13 where the type parameter machinery is mature and ty / mypy handle it cleanly. Python 3.14 is also supported (the package classifiers list both), and CI runs a non-blocking lane against the next CPython’s release candidate so compatibility problems surface before that version ships. Dropping the 3.13 floor would mean rewriting those signatures to the TypeVar / Generic[T] form — doable, but not currently planned.

Hardware and scaling¶

How much CPU does a worker need? How many cores should I allocate?¶

Depends on task shape (fast/slow), executor pool size, and input throughput. Use the Config Calculator to get a starting point, then tune via Performance Recommendations. Rule of thumb: one Python process can saturate ~1 core for the event loop + expect (max_executors × task CPU share) for subprocess work.

How much memory per worker?¶

There’s no single number — memory is dominated by three tunables plus your subprocess. A useful back-of-envelope:

Buffered Kafka messages: max_poll_records × average message size × (number of partitions × window_size) in worst case. Defaults (max_poll_records=100, window_size small) keep this in the tens of MB.
In-memory cache: capped by cache.max_memory_entries (default 10_000). Each entry is a Python dict + serialized Pydantic JSON; budget roughly 1-10 KB/entry for typical payloads. Set max_memory_entries: null for unbounded (the config layer emits a startup warning if you do — monitor RSS).
Flight recorder buffer: small, bounded by ui.recorder.flush_interval_seconds × event rate. Events are buffered in memory and flushed to SQLite; a full buffer is tens of MB.
Subprocess children: max_executors live processes, each using whatever your binary consumes. This usually dwarfs the Python RSS.

For a typical worker (max_executors=80, cache enabled with defaults, recorder on), the Python process sits in the 150-400 MB range. The subprocess pool adds whatever your binary needs, times max_executors. Measure under load; there’s no substitute. Raw capacity knobs are documented in Configuration and the Cache memory cap page.

Can I run multiple workers on one host?¶

Yes. Each worker needs a unique worker_id (set via the env var named in worker_name_env, default WORKER_ID). Recorder and cache DBs are per-worker, so their paths must differ — typically achieved by including the worker_id in the directory path.

How do I scale horizontally?¶

Run more workers in the same Kafka consumer group. The group rebalances partitions across all members; parallelism is capped at the partition count. See Scaling Horizontally and Architecture.

What’s the maximum throughput per worker?¶

Plan for ~400 tasks/sec/worker at most. Your rate follows from the pool size — roughly max_executors / average task duration — but past ~400 completions/sec the single Python event loop that orchestrates everything becomes the bottleneck, and adding slots only grows the queue. See Bottleneck: Event Loop for the breakdown of where the GIL time goes.

The ceiling is soft, and it mostly moves down: the more CPU-bound work your arrange() and completion hooks do on the loop (compiling or applying regexes, nested loops over large collections, building large payloads), the more loop stalls you introduce, and each stall delays every in-flight task’s completion handling. Heavy hooks can drag the practical maximum well below 400/sec — keep them cheap and move heavy work to handler.offload().

Knobs that stretch what one worker delivers within that envelope (all documented in performance.md):

Precomputed tasks (cache hits skip the subprocess entirely) remove the per-task launch cost, though every completion still pays the loop’s orchestration cost.
Batching messages per task (one subprocess launch for N messages via stdin) amortizes the launch cost and — more importantly — raises message throughput without raising the completions/sec the loop has to absorb.
Larger window_size enables larger batches in arrange().
orjson cuts recorder encoding overhead (see Available optimization: orjson).

Past the ceiling, scale horizontally: more workers, each with its own event loop. And if your workload is inherently high-rate, small, CPU-bound tasks, a Go implementation of Drakkar (drakkar-go) exists with a byte-for-byte identical config and API surface — not yet publicly released — whose runtime schedules orchestration across all cores instead of one GIL-held loop; consider porting your business logic to it for that profile.

Use the Config Calculator for a starting point, then measure in staging. Target metrics live in Monitoring Throughput.

What benefit does separating pipelines by partition give?¶

Kafka partitions become independent processing lanes: each partition has its own PartitionProcessor with its own window, message tracker, and offset tracker (Data Flow — Phase 2). Benefits: - Ordering preserved within a partition (key-based grouping stays intact). - A slow message on partition 3 doesn’t block partitions 0/½/4. - Offsets commit per-partition, so retries don’t disturb neighbours.

Handler basics¶

How do I write a handler?¶

Subclass BaseDrakkarHandler[InputModel, OutputModel] and implement at least arrange(). See Quick Start — Define a handler and the full Handler System reference.

What hooks exist and when are they called?¶

on_startup → on_ready → per-message: arrange → (per task) on_task_complete → on_message_complete → (per window) on_window_complete. Error paths: on_error, on_delivery_error. Partition lifecycle: on_assign, on_revoke. See Hook Reference.

What’s a “precomputed” task?¶

A task whose result the handler already knows (cache hit, deterministic shortcut) — the executor skips the subprocess and emits synthetic task_started/task_completed events. Useful for memoization. See Precomputed task results.

When does a message’s offset get committed to Kafka?¶

After all tasks generated from that message terminate (success or failure after retries exhausted) AND on_message_complete has run. Failures go through on_error → DLQ, not a “skipped commit”. See Offset Commit Semantics.

Can I build per-message aggregates from multiple tasks?¶

Yes — that’s what on_message_complete is for. You get a MessageGroup with all the task outcomes. See Fan-out: One Message → Many Tasks → One Aggregate.

Can one task cover multiple messages (fan-in)?¶

Yes. An ExecutorTask can list multiple source_offsets; it only completes the window when all its source messages’ trackers have settled. See Multi-message tasks (fan-IN).

Cache and state¶

What does the built-in cache give me?¶

Framework-provided self.cache on every handler: in-memory LRU + write-behind per-worker SQLite + an optional cross-worker cache exchange with last-write-wins. Survives restarts (per-worker persistence), optionally shares across the fleet. See Cache — What it solves.

How do workers discover each other’s caches?¶

Through the fleet registry — no peer list in the cache config, no Kafka topic. Each worker’s cache file is private to it, in its own storage.dir. Each worker publishes a registry card with the URL of its UI server and, while it serves the exchange, features.cache: the marks of the newest cluster and global rows it originated. Every fleet.cache_exchange.interval_seconds a worker reads the cards and pulls new rows from each peer with GET /api/v1/peer/cache/changes, page by page. It pulls only peers the registry lists, that are not offline or starting, and that have a mark past its cursor. See Cache exchange.

What happens without a registry (provider none, or the UI off): nothing is pulled and the cache operates single-worker. With the static provider the listed peers are pulled every cycle: their cards carry no marks to skip a pull with. No error — it just behaves like a local cache. If you want a cluster-wide cache, give the workers one card folder (the default filesystem provider in shared.dir) and make sure every worker can reach the UI port of the others at the URL on their cards.

Fallbacks when a peer is unreachable: each peer is pulled in its own task. A failed request (timeout, refused connection, a status other than 200, a malformed answer) increments drakkar_cache_sync_errors_total{peer=...}, logs fleet_peer_request_failed once, and backs that peer off for 1, 2, 4, 8, then 16 cycles; one success resets it. One bad peer cannot break the whole cycle. See Cache — Peer unreachable.

Trust boundary: a worker sends exchange requests to the URL on every card in the card folder, and the /api/v1/peer/ route is public — ui.auth_token does not cover it. A worker applies only rows that the peer originated itself, in the scopes it asked for, but anyone who can write the card folder can plant a card whose URL serves rows under that card’s name. Restrict the card folder to the worker user (see Why are peer workers trusted? and Security posture).

What if two workers write the same cache key?¶

Last-write-wins based on updated_at_ms, with origin_worker_id as a lexicographic tiebreak when timestamps collide. Details in Consistency model.

Is cache delete propagated across workers?¶

No — delete is LOCAL only in the current implementation. Other workers’ copies stay until they expire or are overwritten. See Delete is local-only (the main sharp edge).

What’s the difference between `CacheScope.LOCAL` and `CLUSTER`?¶

LOCAL lives only in the worker that wrote it. CLUSTER is pulled by other workers in the same cluster through the cache exchange. See Choosing a scope.

How does cache interaction with SQLite actually work?¶

Reads are in-memory LRU first; miss → fallback to _reader_db (separate aiosqlite connection so reads never queue behind a write). Writes go into _dirty dict and are flushed periodically to _writer_db as a batch. WAL mode so readers don’t block writers. See How it flows.

Sinks and delivery¶

What sinks ship with Drakkar?¶

Kafka, Postgres, Mongo, HTTP, Redis, Filesystem, and a dedicated DLQ (Kafka topic). See Sinks and the per-type config in Configuration — Sinks.

What delivery guarantee does Drakkar provide?¶

At-least-once. A failed sink write triggers DLQ + on_delivery_error; a failed task triggers on_error + offset commit only after a final outcome. Duplicates after crash/retry are possible — design your sinks to be idempotent or tolerate duplicates. See Delivery and error handling.

Does Drakkar support exactly-once?¶

No. Kafka EOS transactions aren’t wired in; workloads that require exactly-once should key their output writes idempotently (Kafka sinks use producer-side dedup via message keys; Postgres/Mongo sinks can use ON CONFLICT / upsert semantics; HTTP sinks should send an idempotency key). There is no roadmap item for first-class transactional EOS — the design cost (coordinating the Kafka consumer group, the producer transaction, and arbitrary external sinks) conflicts with Drakkar’s “many sinks, one pipeline” philosophy. If you need strict EOS between Kafka and Kafka only, use Kafka Streams or a transactional producer directly.

What happens when a sink is unreachable?¶

The sink’s internal retry policy kicks in; exhausted retries call on_delivery_error on your handler and route the record to the DLQ topic. See Dead letter queue.

How do I add multiple sinks of the same type?¶

Name each in config (sinks.kafka.results, sinks.kafka.audit); your handler routes records by setting sink='results' on each payload. See Multiple named sinks.

How do I add a custom sink type?¶

Drakkar discovers third-party sinks via Python’s standard importlib.metadata entry points. Subclass BaseSink[YourPayloadT] from drakkar.sinks.base, register it under [project.entry-points."drakkar.sinks"] in your plugin’s pyproject.toml, and reference it in your worker config under sinks.custom.<type>.<instance>. See Custom sinks (plugin API) for the full contract — BaseSink interface, entry-point declaration, the config layout _build_sinks consumes, and failure modes.

Deployment and operations¶

How do I configure a worker?¶

YAML file + env overrides (DK_ prefix, __ nesting). See Configuration Loading.

How do I stagger a rolling deploy so my fleet doesn’t cascade-rebalance?¶

Leave sources.kafka.startup_align_enabled: true (default). Each worker delays its first subscribe() until the next shared wall-clock boundary, so 10 workers booting over 15 seconds all join the consumer group at the same moment. See Staggered startup alignment.

How many workers should I run per consumer group?¶

Up to the number of Kafka partitions. Extra workers sit idle (Kafka only assigns one consumer per partition). See Scaling Horizontally.

Can I hot-reload config without restarting?¶

No. Config is loaded once at startup and held in the DrakkarApp instance for its lifetime — there is no SIGHUP handler and no config-watch loop. The signal handlers today (drakkar/app.py) are SIGINT and SIGTERM, both of which trigger graceful shutdown. To change config, rely on rolling deploys: Kafka’s cooperative-sticky rebalance keeps non-revoked partitions running during the rollout, and sources.kafka.startup_align_enabled (default on) prevents a fleet-wide cascade rejoin. See Staggered startup alignment.

Where do logs go?¶

Stdout by default; configurable to file/syslog/JSON via structlog. See Structured Logging.

How do I wire up Prometheus scraping?¶

Drakkar exposes metrics on /metrics at the configured port. See Scrape Configuration and the full metric list in Metrics Reference.

How do I deploy in Kubernetes?¶

Drakkar does not ship blessed Helm charts or Kustomize overlays yet — a reference chart is on the roadmap. What the framework does provide to make k8s easy:

Graceful shutdown on SIGTERM (drains in-flight windows, commits settled offsets, disconnects sinks).
Structured logs on stdout (JSON via structlog) — ready for any k8s log collector.
Prometheus metrics on a configurable port for prometheus-operator ServiceMonitor.
Stateful considerations: each worker needs a unique worker_id (set via the env var named in worker_name_env, default WORKER_ID) and a per-worker DB path (recorder + cache SQLite files). Use a StatefulSet with a volumeClaimTemplate mounted at storage.dir for each worker’s own files, and mount one shared ReadWriteMany volume (NFS, EFS) at shared.dir on every pod for the recorder history, archives, merge results and registry cards. A volume that survives a restart lets the next start with the same worker name move a file that a crash or a short stop left in storage.dir. The cache file stays private in storage.dir; the cache exchange runs over HTTP between the pods, so each pod’s UI port must be reachable from the others at the URL on its registry card (set ui.host: 0.0.0.0, or ui.public_url).

Typical shape: one StatefulSet (or Deployment with a PVC per replica) × N replicas = N Kafka consumer-group members. Replica count ≤ partition count. Env-var overrides follow the DK_ prefix + __ nesting convention (see Configuration Loading).

Operator UI and observability¶

How does the operator web UI affect the running pipeline?¶

The UI runs on a separate thread with its own asyncio event loop, so most read-only endpoints don’t interfere. Real effects to be aware of:

Message Probe monkey-patches handler.cache process-wide while a probe runs — production hooks during that window see a DebugCacheProxy and their cache.set() calls are silently suppressed.
Probe consumes ExecutorPool slots exactly like production messages. Close it with ui.probe_enabled: false.
Merge writes files. POST /api/v1/debug/merge creates a merged-<ts>-<token>.db in <shared.dir>/merged/. The archive pass deletes it after history.merged_retention_days (default 32); with archiving off nothing reclaims it. Close it with ui.merge_enabled: false.
GIL contention between the UI thread and the main loop under heavy UI use.
aiosqlite connections are per-thread workers; UI reads don’t block pipeline writes thanks to SQLite WAL mode.

See Observability — Operator UI for the endpoint inventory.

Is the UI safe to expose to a team of operators?¶

To a team of operators on a trusted network, yes – that is what it is for. To anything wider, no: see Security posture.

Auth is opt-in by default. No endpoint stops a worker, replays Kafka messages, mutates sinks, or commits offsets, and Drakkar is intended to run inside a private contour, so the framework starts unauthenticated when ui.auth_token is empty (the default) and emits a structured ui_unauthenticated warning at startup naming the host:port and the two opt-in paths (ui.auth_token in YAML or DK_UI__AUTH_TOKEN env var).

Most endpoints are read-only, but two are not: POST /api/v1/debug/probe runs caller-supplied bytes through the live handler, and POST /api/v1/debug/merge writes a file into the shared folder. Set ui.probe_enabled: false / ui.merge_enabled: false to close them — independently of auth_token, so this works both when you cannot set a token and when you have one and want defence in depth. The startup warning names whichever is still enabled.

For multi-operator setups, set a strong ui.auth_token:

Bearer token on protected endpoints. Once the token is set, /api/v1/debug/databases, /api/v1/debug/merge, /api/v1/debug/download/{path}, /api/v1/debug/download-live, and /api/v1/debug/probe all require an Authorization: Bearer <token> header (or ?token=<token> query parameter). Comparison uses secrets.compare_digest to avoid timing side-channels.
WebSocket auth + origin check. With a token configured, the live-event WebSocket at /ws also requires the same auth_token; invalid tokens close with code 4401. The Origin header is validated on every handshake, token or not, against allowed_ws_origins (explicit allowlist) or the Host header (same-origin fallback). An untokened worker still accepts the cross-origin handshakes the cluster view is built on, but not a page that is off loopback reaching a worker over loopback — see Security. The same rule decides the cross-origin reads that the UI sends to every worker of the cluster (trace, label trace, databases list): with a token set, list every worker origin in allowed_ws_origins — see Cross-origin reads.
Read-only views remain accessible — both before and after enabling auth. /, /live, /partitions, /sinks, /history and the per-task views are SPA routes backed by read-only flight-recorder endpoints; the extra gating listed above is on the mutating / data-exposing ones.
The peer route stays public. GET /api/v1/peer/cache/changes needs no token, because other workers call it without one. It serves the cluster and global cache rows this worker originated — see The public peer route.

Even with auth, the read-only pages expose task stdout/stderr, task env (after redaction), cache contents, and live event streams. Restrict access to operators. Concurrent Message Probes serialize on an internal lock, and the probe temporarily replaces handler.cache — keep this in mind if you have many operators debugging the same worker simultaneously.

Why does my worker emit a `ui_unauthenticated` warning at startup?¶

You have ui.enabled=true and ui.auth_token is empty (the default). That’s an intentional, supported configuration — Drakkar treats the UI as opt-in-auth because no endpoint touches the pipeline and it is meant for private-network deployments. The warning is informational; the worker continues starting normally. It also names any side-effecting endpoint still enabled (probe, merge) so the exposure is explicit rather than implied.

To silence it (i.e. require auth), pick one of:

Set ui.auth_token to a strong random value (python -c "import secrets; print(secrets.token_urlsafe(32))") — recommended whenever the UI is reachable from anywhere outside a fully-trusted operator network.
Set ui.enabled=false — if the worker doesn’t need the flight recorder at all (no observability cost reduction without removing the UI).

See Authentication for the field semantics and the implementation at drakkar/app_security.py::warn_if_ui_unauthenticated.

What is the Message Probe tab?¶

Paste a raw Kafka message value; the framework runs it end-to-end through your handler (arrange → executor → on_task_complete → on_message_complete → on_window_complete) with zero intentional footprint: no sink writes, no offset commits, no recorder rows, no cache writes. Shows every task’s stdin/stdout/stderr/exit code/duration plus the sink payloads that would have been produced. The full contract (headers, cache proxy behavior, concurrent-probe serialization) is documented on the Operator UI page.

If your handler registers a probe_details_model, the probe also gets a User-defined tab showing your own fields — see Probe User Details.

How do I trace a specific message through the pipeline?¶

Use /debug → Message Trace tab, search by partition:offset or by label value. The flight recorder stores every lifecycle event per message. See Observability — Operator UI.

Do UI readers slow down the pipeline?¶

Mildly. Heavy read traffic increases SQLite WAL checkpoint frequency and burns a little Python GIL time. Under normal operator use (a few tabs refreshing) the effect is negligible. See Bottleneck: Recorder and UI.

What happens to old recorder database files? Where did `retention_hours` go?¶

They get archived, not deleted outright. ui.recorder.retention_hours and ui.recorder.retention_max_events are gone — a config that still sets either is ignoring an unknown key. At rotation each worker seals its finished file and moves it into <shared.dir>/history/<cluster>/. A periodic pass merges each finished UTC time window’s sealed files into one compressed <cluster>-<from>__<to>.db.gz in that folder and deletes only the sealed files it merged; with the defaults, that lands 48-72h after the data was recorded. history.archive.retention_days (default 30) expires old archives too; set it to 0 to keep them forever, which the worker warns about at startup because nothing else reclaims that disk. Download them from Debug → Databases tab, or gunzip one directly into an ordinary merged recorder SQLite database. See Archiving for the full mechanics, and set history.archive.enabled: false if you’d rather prune the history folders yourself.

What does the shared folder need from the filesystem?¶

Each worker writes its live recorder file only to its own storage.dir. Only finished, sealed files cross to shared.dir, in a background job, so a slow network mount never holds up rotation or the pipeline. A local disk per worker plus one network mount for shared.dir is the usual layout; when the two are on different filesystems the worker copies the file across, fsyncs it, and only then deletes the source.

The shared folder needs two things:

The same folder on every worker. The mount path may differ: no file stores an absolute path.
An exclusive create and a rename inside one folder. The archive lock (history/<cluster>/.archive.lock) is taken with an exclusive create, and a sealed file appears in the history folder by a rename. It does not need flock.

Clock skew between worker hosts does not matter: lock ages are measured on the file server’s clock (a worker touches a marker file and reads its modification time), never by comparing two hosts’ clocks.

Watch drakkar_recorder_unsealed_files: a value above 0 for two rotation intervals means files cannot reach the shared folder. See Sharing folders across workers.

Can I customize the debug UI?¶

Yes, without writing any client-side code:

Probe User Details — a handler-defined probe_details_model adds a User-defined tab to the Message Probe.
UI Enrichment — links, badges, formats, hints, and detail panels on any probe field or table column.
Declared UI Pages — a handler’s own dashboard page and nav entry, built from built-in data sources (events, annotations, tasks, metrics).

For presentation none of those built-ins can express, custom cell renderers let a deployment-owned JavaScript module take over one cell’s rendering.

Can I change task colors or keep more history on the timeline?¶

Yes, via ui.timeline config — no client-side code needed:

History depth: history_factor (times the executor pool size) governs how many tasks are kept; max_age_minutes (up to 24h) caps how far back in time.
Bar colors: color_rules, a first-match-wins list of conditions on task labels or built-in fields (status, duration, exit code, …), each mapped to a named color or a #rrggbb hex.
Label roles: labels binds up to five task labels to on-screen roles — a tag, a caption, a numeric highlight threshold, a substring filter, and vertical batch markers. A viewer can also override any role for their own browser via the timeline’s gear icon, independent of the backend config.

See Timeline Tuning for the full grammar, the color palette, and a worked example.

Failure modes¶

What happens if my handler crashes mid-message?¶

The exception is caught, logged, the task is marked failed, on_error fires on the handler. If on_error returns RETRY and max_retries isn’t exhausted, the task runs again. If all retries fail, the task is terminal and the message’s offset commits only after all its tasks settle. See Error Handling — on_error Hook.

What if a subprocess hangs?¶

executor.task_timeout_seconds kills the subprocess when asyncio.wait_for raises TimeoutError; the executor’s finally block calls proc.kill() + proc.wait() and the result is recorded with exit_code=-1 and stderr='task timed out'. Then the normal on_error path handles it. See Execution Flow — Enforce Timeout.

What if the Kafka broker goes down?¶

librdkafka (the underlying client) handles reconnects transparently — the consumer automatically rejoins when brokers recover. Drakkar does not expose a distinct “broker-down” state in the UI or Prometheus: there is no drakkar_kafka_connected gauge today. What you will see: drakkar_messages_consumed_total rate drops to zero while processing continues on already-polled messages. Pair a “no new messages” alert (rate(drakkar_messages_consumed_total[5m]) == 0 while expecting traffic) with broker-level monitoring (Kafka’s own JMX / kafka_exporter metrics) to catch broker issues. There is no hard fail-fast option — the design assumption is that brokers recover and transient disconnects should not restart the worker.

What if a sink is unreachable for a long time?¶

Per-batch retries kick in first; exhausted retries fire on_delivery_error and route the record to the DLQ. After failure_threshold consecutive terminal failures (default 5), the sink’s circuit breaker trips and subsequent batches route directly to the DLQ without even attempting the sink — so you stop paying the retry-timeout cost on every batch while the downstream is down. After cooldown_seconds (default 30s) the breaker promotes to half-open and sends a single probe; success closes it, failure reopens with a fresh cooldown. Watch drakkar_sink_circuit_open and drakkar_sink_circuit_trips_total to alert on the condition. See also Dead letter queue.

How do I replay messages from the DLQ?¶

Drakkar does not run a built-in replay worker — replays are operator-driven. Use scripts/replay_dlq.py as the reference tool: it reads DLQ entries and republishes their preserved original_payloads to a target Kafka topic (defaults to your source topic), with --dry-run, --filter, --limit for inspection and a consumer group that is unique per invocation so each run drains from the beginning. The script preserves the serialized payload bytes; it does NOT currently preserve original partition / key / headers — if downstream routing depends on those, point --target-topic at a retry topic whose producer re-derives them. See the DLQ payload schema for the fields the script reads.

A message is stuck in “in-flight” forever — what do I do?¶

Check the Executors tab on /live for stuck tasks; their task_timeout_seconds should eventually fire. When a task overruns the timeout, asyncio.wait_for raises TimeoutError, the executor’s finally block calls proc.kill() + proc.wait(), and the task takes the normal on_error path. If the subprocess itself doesn’t die after proc.kill(), the task coroutine stays pending — the UI server exposes /api/v1/debug/processors, which includes a stuck_tasks list with a live coroutine stack (top 5 frames) per in-flight task; that’s the rescue-visibility tool.

Rescue procedure when a task is truly wedged past the timeout:

Open /live → Executors tab and look for tasks whose oldest_running_sec exceeds task_timeout_seconds.
Hit /api/v1/debug/processors to get the coroutine stack for each wedged task via the stuck_tasks field.
Check worker logs for the task’s task_id — you’ll usually find the downstream call (sink, cache) that’s blocked.
If the stack shows the subprocess already exited and the coroutine is wedged in sink or recorder code, a worker restart is the safest recovery — Kafka’s cooperative-sticky rebalance will reassign the partition to another worker, which will re-poll the un-committed offset range and retry the work.
There is no in-UI “force complete” button today by design: marking a wedged task as succeeded could silently ack a half-applied side effect. Operators are expected to restart the worker instead.

Architecture deep-dive¶

Why is the executor subprocess-based instead of async-native or threaded?¶

Four reasons, all aligned with the “CPU in external binaries, Python orchestrates I/O” design goal:

Language portability — the worker binary can be Rust/Go/C++/a CLI tool you don’t own; it just reads a JSON payload from stdin and writes results to stdout.
Real parallelism — N subprocesses truly use N cores. Python threads would still contend on the GIL for any CPU burst; asyncio.to_thread is only useful for blocking I/O.
OS-level CPU isolation — one task’s bug (segfault, OOM, infinite loop) cannot corrupt the worker process.
Clean timeouts and cancellation — the executor just SIGKILLs a runaway subprocess. Cancelling in-process Python threads is notoriously unreliable.

The cost is ~10ms of launch overhead per task (Bottleneck: Subprocess Launch) — a fair trade when your task does 10ms–10s of real CPU work. For sub-millisecond tasks, use precomputed results to skip the subprocess entirely.

Why SQLite for the recorder and cache?¶

Zero-operational-cost embedded store with WAL-mode concurrency, good enough for the write volumes these components generate, and easy to inspect/backup/ship. The framework separates reader and writer connections so read endpoints never queue behind the writer’s flush. See Cache — How it flows.

How many event loops does a worker run?¶

Two: the main loop (Kafka consumer + partition processors + executor pool + periodic tasks + sink manager) and the UI server loop (uvicorn on a separate thread). asyncio primitives created on one are not safe to use on the other — in particular, the recorder’s aiosqlite connection and the cache reader connection are bound to the main loop. The UI server uses a dispatch-to-loop helper to marshal roughly 20 read-side endpoints (message trace, probe, task listings, cache inspection, etc.) back onto the main loop via asyncio.run_coroutine_threadsafe; pure-Python work (template rendering, counters, constants) stays on the UI loop.

What happens during a Kafka rebalance?¶

Revoked partitions’ PartitionProcessors drain their in-flight windows then stop; newly-assigned partitions spawn fresh processors. on_revoke / on_assign hooks fire so handler code can flush partition-scoped state. See Partition Assignment and Revocation.

How does backpressure work?¶

The poll loop compares in-flight message count to high/low watermarks (multiples of max_executors). Above high-watermark: pause the Kafka consumer so no new batch is polled. Below low-watermark: resume. See Backpressure Formula and Backpressure deep-dive.

What’s the difference between the recorder DB and the cache DB?¶

They’re separate SQLite files with separate schemas and separate reader/writer connections. The recorder stores per-message lifecycle events (flight recorder for the UI); the cache stores operator-written key/value pairs for memoization. See UI / Flight Recorder and Cache.

Security and trust model¶

Drakkar makes five explicit trust assumptions – each one an architectural trust boundary, not a latent bug. Read this before a production deploy.

The short version, and where the boundary actually sits, is on its own page: Security posture. Drakkar is an internal tool; its UI and API belong on a trusted network, and the controls below are defence in depth rather than a perimeter. This section covers the individual assumptions in detail.

Why is the handler binary trusted?¶

Drakkar launches executor.binary_path as a subprocess and pipes the message bytes to its stdin without validation. There’s no sandbox, no signature check, and no attempt to filter input — the binary is assumed to be operator-provided code you audit the same way you audit the rest of your deployment.

See drakkar/executor.py::ExecutorPool._launch for the launch code. The binary runs with the worker’s OS privileges, plus any env overrides from ExecutorConfig.env or per-task env. If you need defense-in-depth against a compromised binary, run the worker under a restricted user / with seccomp / inside a container — Drakkar itself offers no in-process sandbox.

Why are peer workers trusted?¶

Every worker reads, archives and serves the sealed recorder files in <shared.dir>/history/ (see drakkar/recorder/archive.py), and the cache exchange sends requests to the URL on every registry card and applies the cluster and global rows it gets back (see drakkar/cache/exchange.py). There’s no per-write signature, no auth check between workers, no schema-level integrity verification. Anyone who can write to the shared folder can plant recorder files that your workers will archive and serve, or plant a card whose URL serves cache rows under that card’s name. The worker filters those rows to the card’s own origin and the scopes it asked for, which stops local rows and rows that claim another worker, but not rows under the planted name.

Treat shared.dir (and the card folder, when it lives elsewhere) as a shared-trust boundary: any principal with write access to that directory has the same trust level as the workers themselves. On a shared filesystem (NFS, EFS), restrict directory permissions to the worker user.

How is the UI protected?¶

Auth is opt-in (see Is the UI safe above):

Default loopback bind (ui.host='127.0.0.1') — the UI is only reachable from the host out of the box, regardless of auth.
Startup warning when unauthenticated. With ui.enabled=true and an empty auth_token, the worker emits a ui_unauthenticated structured warning at startup naming the unauthenticated bind and the two opt-in paths (YAML key + env var). The worker continues starting — Drakkar treats this as a private-contour-friendly default, not a misconfiguration.
Bearer token when auth_token is set, Origin check always. With a token, every UI page and API endpoint requires Authorization: Bearer <token> (or ?token= for browser navigation); only /healthz and /readyz stay public for Kubernetes probes, and the routes under /api/v1/peer/ for the cache exchange between workers. The WebSocket stream authenticates inside its handshake, and validates the Origin header on every handshake whether or not a token is set. See drakkar/uiserver/server_helpers.py::origin_allowed and drakkar/uiserver/server.py::token_matches.

If you put the UI on a non-loopback host outside a private network, set a strong auth_token and consider a reverse proxy with TLS.

Why doesn’t Drakkar validate Kafka message payloads?¶

Parse failures are policy-driven via sources.kafka.on_parse_error. The default (skip) sets msg.payload=None and stamps msg.parse_error, logging a message_parse_failed warning and ticking drakkar_message_parse_failures_total; the handler decides what to do with the message in arrange(). Set dlq to exclude unparseable messages from arrange() and route them to the DLQ topic as ParseFailurePayload records (a failed DLQ write then follows the dlq.on_send_failure strategy — see the next question), or raise to fail fast on schema-broken deployments. A malicious producer can trigger parse failures but cannot execute code in the worker.

What happens when the DLQ itself fails?¶

The error-handling philosophy is: never wedge the pipeline silently — let the handler decide via hooks, default to log + DLQ. When the DLQ write also fails, the payloads have nowhere safe to go and the dlq.on_send_failure strategy applies:

drop (default) — the loss is logged at CRITICAL and counted in drakkar_dlq_dropped_payloads_total; the message counts as processed, its offset commits, and the pipeline keeps moving.
stall — the offset is left uncommitted (drakkar_delivery_stalled_offsets_total) and the partition is paused so no new messages are fetched. Note that not committing alone would not stop consumption — Kafka consumers fetch from their in-memory position, so the explicit pause is what bounds duplicate processing and tracker memory. The partition resumes after a worker restart or reassignment, replaying from the last committed offset.

Both strategies tick drakkar_dlq_send_failures_total and emit a CRITICAL dlq_send_failed log. See When the DLQ itself fails.

What redactions apply to per-task env?¶

Two surfaces expose env vars:

The recorder’s worker_config table — framework-level ExecutorConfig.env is never written to the recorder (it’s omitted from the JSON payload entirely). Environment variables listed in expose_env_vars are captured by name, and secret-shaped names (*PASSWORD*, *PASSWD*, *SECRET*, *TOKEN*, *KEY*, *API_KEY*, *CREDENTIAL*, *_DSN, *AUTH*, *PRIVATE*, *CERT*, *SALT*) are redacted to ***. Non-matching values with embedded URL credentials (user:pass@host) have the credentials stripped.
The recorder’s per-task env metadata — task.env written by your handler is sanitized with the same secret-name patterns before being stored. The original task object is not mutated; only the recorded copy is redacted. See drakkar/recorder_helpers.py::sanitize_env_value for the regex.

The contract is “aggressive redact, accept false positives”: PASSWORD_RESET_URL is redacted because it matches *PASSWORD*, even though a reset URL isn’t a credential. Operators who need to expose these exact names should rename them — a leaked secret is a worse outcome than a logged URL.

This recorder list is deliberately broader than ExecutorConfig.env_inherit_deny (see Environment Variables), which controls what reaches your subprocess rather than what shows up in a debug DB. Withholding a variable from a running binary can break it silently, so that list stays conservative and does not include *AUTH*, *PRIVATE*, or *CERT*.

Contribute to this page¶

Missing a question you’ve asked or answered? Send a PR adding it here — even questions without a confident answer are useful: they mark the project’s documentation debt visibly, and get resolved faster when they’re written down.