Observability¶

Drakkar provides four integrated observability layers: Prometheus metrics for alerting and dashboards, structured logging for event streams, a browser-based operator UI for real-time inspection, and a SQLite flight recorder for post-mortem analysis. All four are enabled by default and work together – the UI reads from the flight recorder and links out to Prometheus graphs. See Configuration for the full YAML reference.

Prometheus Metrics¶

Metrics are exposed via prometheus_client and served on a dedicated HTTP endpoint. The server starts automatically when metrics.enabled is True (the default).

metrics:
  enabled: true
  port: 9090

Metrics Reference¶

Worker Identity¶

Metric	Type	Labels	Description
`drakkar_worker`	Info	`worker_id`, `version`, `consumer_group`	Worker instance identity, published once at startup. The `consumer_group` label is empty on a worker with `sources.kafka.enabled: false` — it joins no group, so reporting the configured default would place it in every group-scoped dashboard query it has nothing to do with.

Messages¶

Metric	Type	Labels	Description
`drakkar_messages_consumed_total`	Counter	`partition`	Total messages consumed from the source Kafka topic
`drakkar_message_parse_failures_total`	Counter	`partition`	Source messages whose value failed `input_model` deserialization (handled per `sources.kafka.on_parse_error`)
`drakkar_delivery_stalled_offsets_total`	Counter	`partition`	Offsets left uncommitted because sink delivery (including the DLQ fallback) could not be confirmed. The watermark is stalled; messages are redelivered after restart. Alert on this.
`drakkar_suppressed_zombie_deliveries_total`	Counter	`partition`	Sink deliveries suppressed because the task finished after a revoke/shutdown drain timeout — the new partition owner re-processes those messages. A rising rate means `executor.drain_timeout_seconds` is too small for the workload.
`drakkar_messages_unassigned_dropped_total`	Counter	`partition`	Messages received from Kafka for a partition with no registered processor (a revoke raced the poll). The new owner redelivers them; this only signals the race happened.

Executor¶

Metric	Type	Labels	Description
`drakkar_executor_tasks_total`	Counter	`status` (`started`, `completed`, `failed`)	Total executor tasks by outcome
`drakkar_executor_duration_seconds`	Histogram	–	Task execution duration. Buckets: 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2, 5, 10, 30, 60, 120, 300
`drakkar_executor_queue_wait_seconds`	Histogram	–	Time a task waited for a free executor slot before its subprocess could start. Long waits with a busy pool = the pool (or the CPUs behind it) is the bottleneck; long waits with an idle pool = the worker process itself is too slow to start work. Also rides per task as `queue_wait_ms` in `task_started` metadata (timeline hover).
`drakkar_executor_spawn_seconds`	Histogram	–	Time to start the task subprocess: fork/exec plus event-loop scheduling delay around it. Normally single-digit milliseconds; spawn approaching the task duration means the worker process — not the task binary — is the bottleneck. The same figure rides per task as `spawn_ms` in `task_completed` event metadata and shows in the timeline hover.
`drakkar_executor_pool_max`	Gauge	–	Configured executor pool size (`executor.max_executors`)
`drakkar_host_effective_cpus`	Gauge	–	CPUs this worker process can actually use: the affinity mask capped by any cgroup CPU quota, read once at startup. `drakkar_executor_pool_max` above this value means concurrent tasks time-share cores; startup also logs an `executor_pool_exceeds_cpus` warning then.
`drakkar_task_label_value`	Histogram	`label`	Numeric values of the task labels named in `metrics.task_label_histograms`, observed once per completed task. Log-spaced buckets from 0.001 to 1e9 (unitless — a label can hold bytes, line counts, ratios). Values that do not parse as finite numbers are skipped.
`drakkar_executor_pool_active`	Gauge	–	Number of tasks currently running in the executor pool
`drakkar_executor_timeouts_total`	Counter	–	Total tasks that exceeded `task_timeout_seconds` and were killed
`drakkar_executor_output_truncated_total`	Counter	`stream` (`stdout`, `stderr`)	Output streams truncated to `executor.max_stdout_bytes` / `executor.max_stderr_bytes`. One increment per truncated stream per task. Divide by the `drakkar_executor_tasks_total` rate for the fraction of tasks being truncated.
`drakkar_task_retries_total`	Counter	–	Total tasks retried after failure (via on_error returning RETRY)

Average and percentiles come from the histograms with standard PromQL — task duration shown here, and the same shapes work for drakkar_task_label_value by adding its label selector:

# average task duration over 5 minutes
rate(drakkar_executor_duration_seconds_sum[5m])
  / rate(drakkar_executor_duration_seconds_count[5m])

# p50 / p90 / p99 task duration
histogram_quantile(0.99,
  rate(drakkar_executor_duration_seconds_bucket[5m]))

# p90 of a configured task label (metrics.task_label_histograms)
histogram_quantile(0.90,
  rate(drakkar_task_label_value_bucket{label="file_size_bytes"}[5m]))

The Live page’s timeline tab and the Dashboard show the same avg/p50/p90/p99 figures computed browser-side over the tasks currently in memory — no Prometheus needed for a quick read.

Batches¶

Metric	Type	Labels	Description
`drakkar_batch_duration_seconds`	Histogram	–	Window/batch processing duration (arrange through last task completion). Buckets: 0.5, 1, 2, 5, 10, 30, 60, 120, 300, 600

Partitions¶

Metric	Type	Labels	Description
`drakkar_partition_queue_size`	Gauge	`partition`	Messages waiting in the partition’s internal queue
`drakkar_offset_lag`	Gauge	`partition`	Pending (uncommitted) offsets per partition
`drakkar_partition_processor_deaths_total`	Counter	`partition`, `outcome`	Partition loops that exited on an unexpected error. `outcome="restarted"` came back; `outcome="dead"` gave up and now fails `/readyz`. Alert on any non-zero rate — see Deployment
`drakkar_backpressure_active`	Gauge	–	Whether the consumer is paused due to backpressure. `1` = paused, `0` = flowing
`drakkar_total_queued`	Gauge	–	Total messages buffered in all partition queues plus in-flight tasks
`drakkar_assigned_partitions`	Gauge	–	Number of partitions currently assigned to this worker
`drakkar_executor_idle_slot_seconds_total`	Counter	–	Accumulated slot-seconds of executor idle time while messages are waiting in queues. If 2 slots sit idle for 3 seconds while messages are queued, this increments by 6. Only counts when messages are in queues (not yet dispatched), not when the worker has nothing to do.
`drakkar_consumer_idle_seconds_total`	Counter	–	Accumulated seconds with no messages available from Kafka and nothing in queues. Measures time the worker has genuinely nothing to do (consumer lag is zero). Does not count time when the consumer is paused due to backpressure.

Consumer¶

Metric	Type	Labels	Description
`drakkar_consumer_errors_total`	Counter	–	Total Kafka consumer poll errors

Offsets¶

Metric	Type	Labels	Description
`drakkar_offsets_committed_total`	Counter	`partition`	Total offset commit operations

Rebalancing¶

Metric	Type	Labels	Description
`drakkar_rebalance_events_total`	Counter	`type` (`assign`, `revoke`)	Total Kafka rebalance events

Sinks¶

Metric	Type	Labels	Description
`drakkar_sink_deliver_duration_seconds`	Histogram	`sink_type`, `sink_name`	Duration of sink delivery operations. Buckets: 0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1, 5
`drakkar_sink_deliver_errors_total`	Counter	`sink_type`, `sink_name`	Total sink delivery failures
`drakkar_sink_payloads_delivered_total`	Counter	`sink_type`, `sink_name`	Total payloads delivered to sinks
`drakkar_sink_batch_fallbacks_total`	Counter	`sink_type`, `sink_name`	Batched statements that failed and were retried one payload at a time. Delivery stays correct – a failed batch wrote nothing – but costs one round trip per payload instead of one per batch. A climbing rate means batching is not working: check the batch error in the paired `sink_batch_fallback_per_row` warning.
`drakkar_sink_delivery_retries_total`	Counter	`sink_type`, `sink_name`	Total sink delivery retry attempts
`drakkar_sink_deliveries_skipped_total`	Counter	`sink_type`, `sink_name`	Deliveries skipped via `on_delivery_error` returning SKIP
`drakkar_tasks_precomputed_total`	Counter	–	Tasks whose result was supplied by the handler via `ExecutorTask.precomputed`, bypassing the subprocess. Framework is agnostic to the reason (cache hit, lookup, deterministic shortcut). Compare to `drakkar_executor_tasks_total{status="completed"}` for the short-circuit rate.
`drakkar_sink_dlq_messages_total`	Counter	–	Total messages sent to the dead letter queue
`drakkar_dlq_send_failures_total`	Counter	–	Total failed attempts to send messages to the DLQ. What happens next depends on `dlq.on_send_failure`: `drop` (default) commits past the lost payloads, `stall` leaves the offsets uncommitted and pauses the partition — alert on this counter either way.
`drakkar_dlq_unconfigured_drops_total`	Counter	–	DLQ sends dropped because no DLQ topic is configured and the Kafka source is disabled. See the DLQ rule. The first drop also logs `dlq_send_dropped_unconfigured` at warning level.
`drakkar_dlq_dropped_payloads_total`	Counter	`partition`	Payloads dropped because both the sink delivery and the DLQ write failed under `dlq.on_send_failure=drop`. The offset committed and the payloads are lost. Alert on this.
`drakkar_sink_circuit_open`	Gauge	`sink_type`, `sink_name`	Per-sink circuit breaker state: `0.0` closed, `0.5` half-open, `1.0` open. Gauges are zero-initialized at sink registration so a never-tripped sink still appears in scrape output. Sustained `1.0` on a sink means its downstream has been down longer than the cooldown can recover from.
`drakkar_sink_circuit_trips_total`	Counter	`sink_type`, `sink_name`	Transitions into the open state per sink — both the initial failure-threshold trip and every half-open probe failure. A flapping circuit surfaces as a rising rate on this counter, not a single trip plus silent reopens. Alert on `rate(...[5m]) > 0` paired with non-zero `drakkar_sink_circuit_open`.

Handler Hooks¶

Metric	Type	Labels	Description
`drakkar_handler_duration_seconds`	Histogram	`hook` (`arrange`, `on_task_complete`, `on_message_complete`, `on_error`, `on_window_complete`, etc.)	Duration of user handler hook execution. Buckets: 0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1, 5, 30
`drakkar_handler_hook_errors_total`	Counter	`hook`	User hooks that raised instead of returning. The framework contains the failure and keeps the partition flowing, so this counter is the only signal that a handler is broken — alert on any non-zero rate

Periodic tasks¶

Covers both user-defined @periodic methods and the framework’s system loops (cache.flush, cache.sync, cache.cleanup, …), keyed by task name.

Metric	Type	Labels	Description
`drakkar_periodic_task_runs_total`	Counter	`name`, `status` (`ok`, `error`)	Periodic task executions by name and outcome. Alert on a non-zero `status="error"` rate — the wrapper logs and keeps scheduling by default, so this counter is the durable failure signal.
`drakkar_periodic_task_duration_seconds`	Histogram	`name`	Duration of periodic task executions. Buckets: 0.001, 0.01, 0.1, 0.5, 1, 5, 10, 30, 60.

Offload pool¶

CPU-bound hook work moved off the event loop via handler.offload().

Metric	Type	Labels	Description
`drakkar_offload_running`	Gauge	–	Offloaded computations currently executing on the offload thread pool
`drakkar_offload_queued`	Gauge	–	Offloaded calls waiting for a free pool thread. Sustained non-zero values mean `offload.max_threads` is undersized for how often hooks offload concurrently
`drakkar_offload_duration_seconds`	Histogram	`hook`	Execution time of offloaded computations, queue wait excluded. Buckets reach 120s — offloaded work is expected to be seconds-scale

Recorder¶

Emitted only when ui.enabled=true. Track flight-recorder health – buffer depth, drops under load, and flush latency. Alert on sustained non-zero dropped_events_total or p99 flush_duration_seconds exceeding flush_interval_seconds.

Metric	Type	Labels	Description
`drakkar_recorder_buffer_size`	Gauge	–	Current depth of the in-memory event buffer. Updated on every `_record()` call and after each flush. A sustained value near `max_buffer` signals the flush loop can’t keep up with incoming events – raise `flush_interval_seconds` or `max_buffer`, or investigate disk latency.
`drakkar_recorder_dropped_events_total`	Counter	–	Events evicted from the ring buffer because it was full at record time. Non-zero means event history has gaps; scale the buffer up or speed up flushes. This was previously a silent failure mode – the counter makes it alertable.
`drakkar_recorder_flush_duration_seconds`	Histogram	–	Duration of the full flush body (`executemany` + `commit`). Exposes disk-I/O latency tail. Buckets: 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5. Alert on p99 exceeding `flush_interval_seconds` – flushes overlapping the next scheduled flush starve the buffer.
`drakkar_recorder_flush_retries_total`	Counter	–	Flush attempts that failed with `aiosqlite.OperationalError` and re-queued their batch at the front of the buffer. A non-zero `rate(...[5m])` signals a transient DB-health issue (WAL lock, disk pressure); if it climbs without stabilising, the recorder is about to start dropping batches. Pair with `drakkar_recorder_flush_batches_dropped_total` to distinguish “retrying” from “giving up”.
`drakkar_recorder_flush_batches_dropped_total`	Counter	–	Batches discarded after `max_flush_retries` consecutive failures. This is silent recorder data loss — alert on any non-zero rate. Investigate disk space, filesystem health, and WAL contention.
`drakkar_recorder_requeue_overflow_total`	Counter	–	Events silently dropped from the recorder buffer tail when an `OperationalError`-triggered requeue exceeded `max_buffer` capacity (concurrent records filled the deque during the flush). Any non-zero value is data loss on the flush-retry path — alert alongside `flush_batches_dropped_total`.
`drakkar_recorder_ws_dropped_events_total`	Counter	–	Events dropped for a live WebSocket subscriber whose queue was full. Affects only that subscriber, which is signalled to resync from the DB, so the UI self-heals; sustained growth means live views cannot keep up with event volume.
`drakkar_recorder_archive_bytes`	Gauge	–	Total size on disk of this cluster’s compressed `.db.gz` archives in its history folder (`<shared.dir>/history/<cluster>/`), refreshed on every archive pass including the ticks that archive nothing. Archives are the only recorder artifact nothing else reclaims: bounded by `history.archive.retention_days`, and unbounded while that is `0`. Alert against your volume budget.
`drakkar_recorder_archive_files`	Gauge	–	Number of this cluster’s `.db.gz` archives in its history folder. Read with `drakkar_recorder_archive_bytes` to tell a slow accumulation of small windows from a few very large ones.
`drakkar_recorder_unsealed_files`	Gauge	–	Own recorder files closed but not yet sealed into the history folder: files waiting for the background move, plus the backlog of failed seals and moves. Alert when it stays above `0` for two rotation intervals: the history folder is unwritable, or another process holds the file. See Sealing.
`drakkar_recorder_seal_failures_total`	Counter	–	Recorder files whose sealing failed; they stay in the backlog and are retried. Counts a failed seal and a failed move into the history folder (log `recorder_seal_failed`, field `stage`).
`drakkar_recorder_archive_failures_total`	Counter	–	Archive passes that failed after taking the archive lock: a window failed, the pass was fenced because another worker took the lock over, or an unexpected error.
`drakkar_recorder_annotations_total`	Counter	–	Handler annotations accepted and written to the recorder. See Annotations.
`drakkar_recorder_annotations_dropped_total`	Counter	`reason`	Annotations discarded before reaching the recorder. `reason` is `oversize` (payload alone exceeded `annotation_max_bytes`), `budget_exhausted` (the hook invocation had spent `annotation_max_bytes_per_call`), `no_context` (called outside a framework-invoked hook — a handler bug), or `unserializable`. Payloads are dropped whole, never truncated. The accompanying warning log falls silent after five drops in one hook invocation to protect the log pipeline, so alert on this counter, not on log volume.

Cache¶

Emitted only when cache.enabled=true. Memory gauges are maintained as running sums — Prometheus scrape reads a single int, never walks the in-memory dict. DB gauges are refreshed by the cache.cleanup loop (default every 60s).

Metric	Type	Labels	Description
`drakkar_cache_hits_total`	Counter	`source` (`memory`, `db`)	Cache reads served from memory or from the DB fallback. `peek()` is not counted — only `get()` increments this.
`drakkar_cache_misses_total`	Counter	–	Cache reads that returned `None` (key absent or expired in both memory and DB).
`drakkar_cache_writes_total`	Counter	`scope` (`local`, `cluster`, `global`)	Total `Cache.set()` calls, labelled by the entry’s scope.
`drakkar_cache_deletes_total`	Counter	–	Total `Cache.delete()` calls regardless of whether the key was present. Note: delete is local-only — see sharp edge.
`drakkar_cache_evictions_total`	Counter	–	Entries popped from the in-memory dict due to the LRU cap (`max_memory_entries`). Sustained high rate signals cap undersizing.
`drakkar_cache_flush_entries_total`	Counter	`op` (`set`, `delete`)	Entries drained from the dirty map to SQLite per flush cycle, by op type. Measures flush throughput, not rows actually modified (LWW may reject a SET).
`drakkar_cache_flush_failures_total`	Counter	–	Flush cycles that raised. Consecutive failures mean cache writes accumulate in memory (lost on shutdown if the final drain also fails); after 5 in a row the worker logs `cache_flush_failing_repeatedly` at ERROR. Alert on a rising rate.
`drakkar_cache_cleanup_removed_total`	Counter	–	Rows removed from the SQLite DB by the cleanup loop (entries whose TTL elapsed). A sudden spike means many entries expiring together.
`drakkar_cache_sync_entries_fetched_total`	Counter	`peer`	Rows pulled from a peer over HTTP (cache exchange) and accepted, per peer. Counted after the page commits. Rows the worker rejected are not counted here; they go to `drakkar_cache_sync_errors_total`.
`drakkar_cache_sync_entries_upserted_total`	Counter	`peer`	Rows from a peer passed to the LWW upsert of the local DB, per peer. Counted with the same rows as `sync_entries_fetched_total`, so the two are equal; rows that LWW keeps unchanged are still counted.
`drakkar_cache_sync_errors_total`	Counter	`peer`	Per-peer exchange errors: one per failed request (timeout, refused connection, a status other than 200, an answer that is not the expected JSON), one per page that could not be applied locally, and one per rejected row (wrong scope or not the peer’s own origin). The loop keeps running.
`drakkar_cache_peer_sync_timeouts_total`	Counter	–	Worker-level count of exchange cycles that hit `fleet.cache_exchange.cycle_deadline_seconds` (or the derived `interval_seconds * 0.9` default). One tick per cycle that was cut short by the deadline; the cut peers are not counted as failures. A sustained non-zero rate signals slow peers, or more peers than one interval can pull — raise the deadline and the interval, or raise `max_concurrent_peers`.
`drakkar_cache_entries_in_memory`	Gauge	–	Entries currently in the in-memory dict. Running sum, adjusted per set/delete/evict/cleanup.
`drakkar_cache_bytes_in_memory`	Gauge	–	Sum of `size_bytes` across in-memory entries. Running sum (see above).
`drakkar_cache_entries_in_db`	Gauge	–	Rows in the local `<worker>-cache.db`. Refreshed by the cleanup loop.
`drakkar_cache_bytes_in_db`	Gauge	–	Sum of `size_bytes` across DB rows. Refreshed by the cleanup loop.

Cache flush/sync/cleanup loop durations are captured by the existing drakkar_periodic_task_duration_seconds{name="cache.flush|cache.sync|cache.cleanup"} histogram — no dedicated cache-only timing histograms.

Each HTTP request of the exchange is also counted in the fleet metrics drakkar_fleet_peer_requests_total{endpoint="cache_changes", outcome} (ok, http_error, timeout, transport_error, bad_body) and timed in drakkar_fleet_peer_request_duration_seconds{endpoint}. See Fleet registry — Metrics. The exchange logs fleet_peer_request_failed, fleet_peer_backoff and fleet_peer_rows_rejected (Fleet registry — Log events), cache_peer_sync_failed when a pulled page cannot be applied locally (fields peer, error), and cache_peer_sync_cycle_timeout when a cycle hits its deadline.

Webapp¶

Emitted only when sources.http.enabled=true. Track per-client request volume, latency, capacity headroom, and drop rates. Status labels are drawn from a closed set documented in Webapp → Status codes: ok | timeout | error | rate_limited | auth_failed | shutdown | not_ready | capacity. The client label is bounded by the configured client list plus the fixed unauthenticated sentinel for auth-failed / pre-auth gate hits, so cardinality stays small.

Metric	Type	Labels	Description
`drakkar_webapp_requests_total`	Counter	`client`, `status`	One increment per HTTP request, labelled by the matched client name and the terminal outcome. Replaces a separate `webapp_rate_limited_total` – query `drakkar_webapp_requests_total{status='rate_limited'}` instead.
`drakkar_webapp_request_duration_seconds`	Histogram	`client`, `status`	Server-side wall-clock duration of HTTP requests. Buckets: 0.005, 0.01, 0.05, 0.1, 0.5, 1, 2, 5, 10, 30 – covers sub-second work and the default 30s `request_timeout_seconds` budget. The duration starts at route entry and ends at response emission, so it includes auth + rate-limit + body parsing as well as the runner.
`drakkar_webapp_inflight`	Gauge	–	Number of HTTP requests currently being processed by the runner on T1. Incremented on runner entry, decremented in `finally`. Alert on `drakkar_webapp_inflight > sources.http.max_concurrent` to spot semaphore-leak bugs – a permit accidentally held past response would surface here.
`drakkar_webapp_dropped_after_timeout_total`	Counter	`client`	Requests whose pipeline result was dropped on T1 after T2 had already returned a 504 to the caller. Incremented at each cancellation gate (post-execute and pre-`on_http_request_complete`). A high rate signals either a too-tight `request_timeout_seconds` or a slow subprocess pipeline – correlate with `drakkar_executor_duration_seconds`.
`drakkar_webapp_rpm_limit`	Gauge	`client`	Configured rpm cap per client. Informational gauge set once at webapp startup and never updated thereafter (the cap is a config field; operators edit YAML and restart). Read alongside `drakkar_webapp_requests_total{status='rate_limited'}` to confirm deployed limits match the workload’s expectations.

Shutdown¶

Metric	Type	Labels	Description
`drakkar_uncommitted_offsets_at_stop`	Gauge	–	Snapshot at `_shutdown` start: count of Kafka offsets that were registered in-flight but not yet committed when shutdown began. Summed across all assigned partitions via each `OffsetTracker.pending_count`. Always set (even to `0`) so the gauge reflects the most recent shutdown rather than a stale prior value.
`drakkar_inflight_at_stop`	Gauge	–	Snapshot at `_shutdown` start: number of in-flight executor subprocesses (`ExecutorPool.active_count`) running user code when shutdown began. Always set, including to `0`.
`drakkar_drain_timeout_hit_total`	Counter	–	Incremented each time an input source’s drain exceeded `executor.drain_timeout_seconds` before its in-flight work finished — the Kafka source’s partition processors or the HTTP source’s in-flight requests. A nonzero rate signals workers being killed mid-flight — either the timeout is too tight or handlers are stuck.
`drakkar_suspected_oom_kills_total`	Counter	–	Incremented at startup when the previous run left a watchdog file at `{storage.dir}/{worker_id}.watchdog` whose body lacked the `CLEAN_EXIT` marker. That signature means the prior process was killed before reaching the normal shutdown path — typically OOM-killer SIGKILL, kubelet pod-pressure eviction, or kernel panic. See OOM / SIGKILL detection below.

OOM / SIGKILL detection¶

Drakkar writes a small watchdog file at {storage.dir}/{worker_id}.watchdog during startup and removes it as the last step of a clean shutdown. The file’s presence and contents at the next startup let the framework distinguish three termination modes:

Watchdog state at startup	Previous run	Action
File absent	First run, or prior shutdown completed cleanly and unlinked the file.	No log, no metric.
File present, body == `CLEAN_EXIT`	Prior shutdown wrote the marker but a process death between the write and the unlink left the file behind. Still a clean exit.	No log, no metric.
File present, body empty (or anything else)	Prior process was killed before `_shutdown` reached `mark_clean` — almost always SIGKILL from the OOM-killer, a kubelet pod-pressure eviction, or a kernel panic.	Structured warning `previous_run_ended_unexpectedly` (category `watchdog`) with `worker_id` and `watchdog_path` fields, plus `drakkar_suspected_oom_kills_total` increment.

The watchdog file is created in AppLifecycle._async_run once the worker is committed to running (the actual write() happens just before subscribe() returns and the poll loop starts) and deleted as soon as the drain phase has been accounted for in _shutdown. A drain_timeout is treated as a clean (if slow) shutdown for the watchdog’s purposes — its dedicated counter drakkar_drain_timeout_hit_total already covers that case. Conflating “drain was slow” with “process was SIGKILLed” would muddle dashboards, so the OOM counter drakkar_suspected_oom_kills_total is reserved for the genuinely-empty-body signature: process killed before reaching mark_clean.

The file lives in storage.dir (default /tmp/drakkar), the same folder used by the recorder and cache engine. If storage.dir is empty — fully disk-less deployment — the watchdog is disabled for that run (no file written, no metric increment), since the alternative of falling back to the worker’s CWD risks both breaking the “no on-disk state” promise and writing into a read-only container volume. The disable is logged at startup as watchdog_disabled_no_storage_dir so operators can see the OOM signal is off.

If the watchdog write itself fails (read-only mount, permission denied, no space), the worker logs a structured warning watchdog_write_failed with path and error, then disables the watchdog for the rest of the run and continues startup. The watchdog is observability-only — losing the OOM signal is preferable to crashing the worker over a misconfigured volume mount.

Operator interpretation: any nonzero rate on drakkar_suspected_oom_kills_total should be correlated with container-runtime pod-restart events. A spike usually maps to one of: container memory limit too low, a memory leak in handler code, a sibling pod evicting this worker, or a node-level OOM. Pair the metric with drakkar_uncommitted_offsets_at_stop and drakkar_inflight_at_stop from the previous run (Prometheus retains the last-set value across restarts) to gauge how much in-flight work was lost.

Some feature-specific metrics are documented on their own pages rather than here: runtime health (event-loop lag gauges and stall counters), task cost & throughput, and consume pause.

User-Defined Metrics¶

You can declare Prometheus metrics as class attributes on your handler – see Custom Prometheus Metrics for the handler-side setup. The framework auto-discovers them via discover_handler_metrics(), which scans the handler’s class hierarchy (MRO) for any prometheus_client metric instances (Counter, Gauge, Histogram, Summary, Info).

from prometheus_client import Counter, Histogram
from drakkar.handler import BaseDrakkarHandler

class MyHandler(BaseDrakkarHandler[MyInput, MyOutput]):
    # These are automatically discovered and shown in the operator UI
    items_parsed = Counter(
        'myapp_items_parsed_total',
        'Total items parsed from executor output',
    )
    parse_duration = Histogram(
        'myapp_parse_duration_seconds',
        'Time spent parsing executor output',
    )

    async def on_task_complete(self, result):
        with self.parse_duration.time():
            items = parse_output(result.stdout)
        self.items_parsed.inc(len(items))
        ...

User metrics are displayed alongside framework metrics on the UI’s metrics page (/debug), tagged with source user to distinguish them from the built-in drakkar_* metrics.

Scrape Configuration¶

Metrics are exposed at :9090/metrics by default (configurable via metrics.port).

Static TargetsKubernetes Service Discovery

# prometheus.yml
scrape_configs:
  - job_name: drakkar
    static_configs:
      - targets:
          - 'worker-1:9090'
          - 'worker-2:9090'
          - 'worker-3:9090'

# prometheus.yml
scrape_configs:
  - job_name: drakkar
    kubernetes_sd_configs:
      - role: pod
    relabel_configs:
      - source_labels: [__meta_kubernetes_pod_label_app]
        regex: drakkar
        action: keep
      - source_labels: [__meta_kubernetes_pod_ip]
        target_label: __address__
        replacement: '${1}:9090'

Structured Logging¶

Drakkar uses structlog for structured, context-rich logging. By default, log output goes to stderr (standard Unix convention), but you can redirect to stdout or a file.

Configuration¶

logging:
  level: INFO       # DEBUG, INFO, WARNING, ERROR, CRITICAL
  format: json      # json | console
  output: stderr    # stderr | stdout | file path

json (default) – produces one JSON object per line, suitable for ingestion by Elasticsearch, Loki, Datadog, or any log aggregator.
console – human-readable colored output for local development.

Output Destination¶

The output field controls where logs are written:

Value	Behavior
`stderr` (default)	Standard error stream
`stdout`	Standard output stream
File path	Append to the specified file (created if missing, parent directories created automatically)

File paths support template variables:

Variable	Substitution
`{worker_id}`	Worker name (from `WORKER_ID` env var or config)
`{cluster_name}`	Cluster name (from config or env var)

# Examples
logging:
  output: stdout

logging:
  output: /var/log/drakkar/{worker_id}.log

logging:
  output: /var/log/{cluster_name}/{worker_id}.jsonl

Environment variable override: DK_LOGGING__OUTPUT=/var/log/drakkar/worker.log

Automatic Context Fields¶

Every log line includes these fields, bound once at startup:

Field	Source	Example
`timestamp`	ISO 8601 timestamp	`2026-04-06T12:34:56.789Z`
`level`	Log level	`info`
`service_name`	Hardcoded	`drakkar`
`worker_id`	`worker_name_env` env var	`worker-3`
`consumer_group`	`sources.kafka.consumer_group` config; omitted entirely when the Kafka source is disabled	`drakkar-workers`
`module`	Python module name	`drakkar.partition`

Per-Context Binding¶

The framework binds additional fields for specific operations using structlog.contextvars. These appear on all log lines within that context:

partition – bound during partition processing
hook – bound during handler hook execution (e.g., arrange, on_task_complete, on_message_complete)
task_id – bound during task execution and result handling

Usage in Handler Code¶

import structlog

logger = structlog.get_logger()

class MyHandler(BaseDrakkarHandler[MyInput, MyOutput]):
    async def arrange(self, messages, pending):
        await logger.ainfo(
            "Arranging batch",
            category='handler',
            message_count=len(messages),
            pending_count=len(pending.pending_task_ids),
        )
        ...

Output (JSON format):

{
  "timestamp": "2026-04-06T12:34:56.789Z",
  "level": "info",
  "event": "Arranging batch",
  "service_name": "drakkar",
  "worker_id": "worker-3",
  "consumer_group": "drakkar-workers",
  "module": "myapp.handler",
  "category": "handler",
  "message_count": 50,
  "pending_count": 12
}

Operator UI¶

The operator web UI is a FastAPI application served in a separate thread so that CPU-intensive executor tasks on the main event loop do not block the interface. Enabled by default on port 8080.

ui:
  enabled: true
  port: 8080

Pages¶

`/` – Dashboard¶

The landing page with a high-level overview of worker health:

Partition tiles – one tile per assigned partition showing queue depth and status.
Pool utilization – active tasks vs. max workers.
Event counters – consumed, completed, failed, produced, committed totals.
Consumer lag – total lag across all assigned partitions.
Custom links – configurable via ui.custom_links with template variables ({worker_id}, {cluster_name}, {metrics_port}, {debug_port}).
Prometheus graph links – when ui.prometheus_url is configured, each stat card links directly to a Prometheus graph filtered to this worker. Worker-scoped links are grouped by category (Throughput, Latency, Health, Errors). Cluster-wide links appear when ui.prometheus_cluster_label is set.

`/partitions` – Partition Overview¶

Per-partition stats table:

Queue size (messages waiting)
Pending offsets (uncommitted)
Consumer lag (from Kafka high watermark)
Committed offset and high watermark
Consumed, completed, and failed event counts
Live/dead status indicator

Lag is the high watermark minus the committed offset. The committed offset comes from the broker. If the broker does not report one for a partition, the worker uses the last offset that it committed for that partition itself. When neither is known, the committed offset shows as - (null in the API) and lag counts from where the consumer starts to read under auto.offset.reset: the low watermark for earliest (the default), the high watermark for latest. The dashboard’s total lag is the sum of these per-partition values.

`/partitions/{id}` – Partition Detail¶

Paginated event browser for a single partition. Shows all recorded events (consumed, arranged, task_started, task_completed, task_failed, committed) in reverse chronological order with 50 events per page.

`/sinks` – Sink Delivery Stats¶

Overview of all configured sinks with live delivery statistics:

Delivered count and payload count
Error count and retry count
Last delivery timestamp and duration
Last error message and timestamp
Links to external sink UIs (when ui_url is configured on a sink)

`/live` – Live Pipeline View¶

Real-time view of the processing pipeline, fed by WebSocket (/ws). Organized in tabbed sections:

Arrange – active arrange() calls with partition, duration, and message labels.
Executors Timeline – visual timeline of running, pending, and recently finished tasks. Slot-based layout matching max_executors. Tasks shorter than ws_min_duration_ms are hidden (except failures, which are always shown). See Duration Thresholds for threshold tuning.
Collect – recently completed tasks with exit codes and durations.

Pool utilization bar shows active, waiting, and available slot counts.

`/history` – Event Browser¶

Filterable, paginated event browser across all partitions:

Partition filter dropdown
Event type toggles (consumed, arranged, task_started, task_completed, task_failed, committed, etc.)
100 events per page with forward/backward pagination

`/debug` – Debug Tools¶

Multi-purpose debug page with:

Metrics viewer – live snapshot of all registered Prometheus metrics (framework and user), accessible via the /api/v1/debug/metrics JSON endpoint.
Periodic tasks – status of all @periodic handler methods: last run time, duration, success/error status, total counts, and a sparkline of recent runs.
Cross-worker message trace – two search modes:
- By partition:offset – enter partition:offset (e.g., 5:42) to trace a message through the full pipeline.
- By label – search inputs are auto-generated for each label key found in the database (e.g., request_id, pattern). Enter a value to find all tasks and events matching that label. Useful for tracing a specific request across the partitions of this worker. A partition:offset trace searches the current worker’s own files, or the sealed history of its cluster on request. A label search searches the current worker’s live file. See Cross-Worker Trace.
Database management – list this worker’s own recorder files and the shared history folders (sealed files, archives, merge results) with event counts and sizes, download history files, merge several of them into one, and view per-file breakdown by event type. See The Databases page.
Live file download – GET /api/v1/debug/download-live answers a consistent snapshot of this worker’s live recorder file: a copy made with the SQLite backup API on its own read-only connection, in rollback-journal format, so it opens on its own. The temp copy is removed at once. 404 in memory-only mode, 500 {"error": "Snapshot failed: ..."} when the copy fails. Behind ui.auth_token like the other debug routes.
Message Probe – replay a single pasted message through the handler pipeline (arrange → executor → on_task_complete → on_message_complete → on_window_complete) with zero footprint: no sink writes, no offset commits, no event-recorder rows, no cache writes, nothing for peers to pull. The report shows every task’s stdout/stderr/exit code, each hook’s returned CollectResult, the payloads that would have been written to each sink, every cache call, a timeline waterfall, and any exceptions raised at any stage. Posts to /api/v1/debug/probe with a JSON body containing the message’s value (and optional key, partition, offset, topic, timestamp, use_cache).

`/task/{id}` – Task Detail¶

Detailed view of a single task’s lifecycle:

Status – running, completed, or failed
Labels – user-defined labels from ExecutorTask.labels
Duration – wall-clock execution time
PID – OS process ID
CLI – reconstructed command line (binary_path + args)
stdout/stderr – captured process output (subject to output_min_duration_ms threshold)
Source offsets – which Kafka offsets this task covers
Event timeline – chronological list of all events for this task_id (started, completed/failed, task_complete)

API¶

The pages above are backed by a JSON/WebSocket API under /api/v1. Endpoints useful beyond the pages themselves:

GET /api/v1/identity – worker identity for the SPA and external tooling. Since contract v1.2 the response also carries backend ("python" or "go"), backend_version, ui_version, and ui_source.
GET /api/v1/openapi.json – the vendored OpenAPI 3.1 spec describing the /api/v1 surface. Served from drakkar/uiserver/openapi.yaml; the canonical source is docs/openapi-v1.yaml in the drakkar-ui repo, and tests/test_openapi_parity.py pins the served route table to the spec.
GET /api-docs – a self-hosted Swagger UI over that spec (no CDN assets). It moved off /docs in contract v1.22; that path now serves the operator docs site.

Both the spec and Swagger UI endpoints are auth-gated exactly like the rest of their route class when ui.auth_token is set.

Managing the UI bundle cache¶

The drakkar-ui console script (installed with the package) manages the shared per-user bundle cache – the same engine the worker runs at startup, so you can prime or inspect the cache without starting a worker:

drakkar-ui where                    # report the cache location + what would be served
drakkar-ui update                   # download the latest UI release into the cache
drakkar-ui fetch --version=v0.2.0   # download a specific release

All subcommands accept --repo=owner/name, --cache-dir=DIR, and --api-base=URL (GitHub Enterprise); a GITHUB_TOKEN in the environment raises rate limits and unlocks private repos. Cached versions are never re-downloaded – release tags are immutable. just drakkar-ui <args> wraps the same command for repo-local use.

When neither the cache nor the release source can supply a bundle (an offline host whose cache is still empty), the worker runs API-only: /api/v1/..., the Kubernetes probes and the event WebSocket are all unaffected, and every page request answers 503 naming the ways to supply a bundle. There is no built-in HTML fallback – the UI is one versioned artifact, and a single download at any point in a worker’s life fills a cache that every later start reads.

For an air-gapped deployment, stage the bundle once with drakkar-ui fetch --version=vX.Y.Z (into a cache directory the workers share), or point ui.release.repo at an internal mirror that publishes the same release assets. ui.release.enabled: false turns the UI off deliberately, leaving the same API-only worker.

Flight Recorder¶

The flight recorder is a SQLite-based event log that captures the full processing lifecycle. It provides the data backing for the operator UI and survives worker restarts (within retention limits).

Configuration¶

storage:
  dir: /tmp/drakkar                   # This worker's own files; '' disables disk persistence
ui:
  recorder:
    store_events: true                # Write processing events to the events table
    store_config: true                # Write worker config into each file
    store_state: true                 # Periodic state snapshots
    state_sync_interval_seconds: 10   # Snapshot frequency for worker_state
    flush_interval_seconds: 5         # Buffer flush interval
    max_buffer: 50000                 # In-memory event buffer size
    max_flush_retries: 3              # Failed-batch retries before the batch is dropped
    rotation_interval_hours: 1        # Rotate to a new DB file every N hours (1 = 1 hour)
    store_output: true                # Include stdout/stderr in event records
    output_min_duration_ms: 500       # Tasks faster than this store no args/stdout/stderr
    event_min_duration_ms: 0          # Task events faster than this are not persisted at all
    store_stdin: false                # Store each task's stdin in task_started metadata (failures always store it)
    stdin_max_bytes: 65536            # Byte cap for stored stdin; 0 = unlimited
    annotations_enabled: true         # Accept handler-emitted annotations
    annotation_max_bytes: 16384       # Largest single annotation; 0 = unlimited
    annotation_max_bytes_per_call: 262144  # Per hook invocation; 0 = unlimited
    annotation_log_max_bytes: 2048    # Cap on a dropped payload's log copy; 0 = unlimited

The four annotation_* settings govern handler annotations — diagnostic records your handler attaches to a window, message, or task. They are stored as ordinary events rows, so they rotate, archive, and (optionally) expire with everything else — see Archiving below.

Database Schema¶

Each database file can contain up to three tables, controlled by the store_events, store_config, and store_state flags:

`events` – Processing Events¶

Stores processing events across the full pipeline lifecycle.

Key columns: ts, dt, event, partition, offset, task_id, args, stdout, stderr, exit_code, duration, output_topic, metadata (JSON), pid, labels (JSON), origin ('kafka' or 'http', default 'kafka'), client_name (webapp client name; NULL for Kafka-origin rows), request_id (webapp framework id; NULL for Kafka-origin rows).

Indexed on (partition, offset), ts, dt, task_id, event, labels (partial, where not null), origin, and request_id (partial, where not null).

Event types:

Event	When recorded	Key fields
`consumed`	Message polled from Kafka	`partition`, `offset`
`arranged`	`arrange()` completes for a window	`partition`, `metadata` (message_count, task_count, offsets, message_labels, window_id)
`annotation`	A handler calls `self.annotate(...)` — see Annotations	`partition`, `offset` (message scope), `task_id` (task scope), `labels`, `metadata` (kind, scope, hook, window_id, offsets, data)
`task_started`	Subprocess launched (after semaphore acquired)	`task_id`, `partition`, `args`, `pid`, `labels`, `metadata` (source_offsets, slot, queue_wait_ms; stdin + stdin_truncated when `store_stdin` is on, else stdin_bytes size marker)
`task_completed`	Subprocess finished with exit 0	`task_id`, `duration`, `exit_code`, `stdout`, `stderr`, `pid`, `labels`, `metadata` (spawn_ms; plus `cost`/`speed` for throughput-counted tasks)
`task_failed`	Subprocess failed (non-zero exit, timeout, crash)	`task_id`, `duration`, `exit_code`, `pid`, `labels`, `metadata` (exception; stdin always stored, capped at `stdin_max_bytes`)
`task_complete`	`on_task_complete()` hook finishes for one task	`task_id`, `partition`, `duration`, `metadata` (output_message_count)
`message_complete`	`on_message_complete()` hook finishes for one source message	`partition`, `offset`, `duration`, `metadata` (task_count, succeeded, failed, replaced, output_message_count)
`window_complete`	`on_window_complete()` hook finishes for one arrange() window	`partition`, `duration`, `metadata` (window_id, task_count, output_message_count)
`produced`	Kafka payload delivered to output topic	`partition`, `offset`, `output_topic`
`sink_delivered`	Sink delivery succeeds	`metadata` (sink_type, sink_name, payload_count, duration)
`sink_error`	Sink delivery fails	`metadata` (sink_type, sink_name, error, attempt)
`committed`	Kafka offset committed	`partition`, `offset`
`assigned`	Partition assigned during rebalance	`partition`
`revoked`	Partition revoked during rebalance	`partition`
`partition_stalled`	Partition paused because an offset stalled (`dlq.on_send_failure=stall`)	`partition`
`runtime_health`	Event-loop health transition or periodic sample — see Runtime Health	`metadata` (kind: transition/sample, state, lag_ms, unit_count)
`runtime_stall`	Event loop resumed after a stall; carries the stack traces sampled while it lasted	`duration`, `metadata` (duration_ms, stacks: [{stack, location, count}], dropped_stacks, unit_count)
`runtime_lag_episode`	A degraded/stalled span ended (or hit `runtime_health.episode_max_seconds`); carries stacks aggregated across the whole span and a verdict — see Runtime Health	`duration`, `metadata` (duration_ms, peak_lag_ms, lag_sum_ms, cpu_ms, cpu_ratio, verdict: blocked/cpu_bound/starved/inconclusive, stall_count, sample_count, stacks, dropped_stacks, unit_count, plus optional evidence: cpu_throttled_ms, psi_cpu_some_avg10, load1)
`runtime_probe`	Opt-in stack probe (`runtime_health.probe_interval_seconds > 0`) — a flight-recorder profiler sample, recorded regardless of health state	`metadata` (lag_ms, unit_count, stacks: [{stack, location, count}])
`resource_sample`	Periodic snapshot of what the worker consumed — and what the host is fighting over (Host Pressure) — every `state_sync_interval_seconds`	`metadata` (rss_bytes, threads, open_fds, cpu_self_pct, cpu_children_pct, interval_s, load1, load5, psi_cpu_some_avg10, psi_io_some_avg10, psi_io_full_avg10, psi_mem_some_avg10, psi_mem_full_avg10, cpu_throttled_periods, cpu_throttled_ms, nfs_mounts: [{mount, ops, rtt_ms, retrans}] — each omitted when its source is unavailable; `cpu_children_pct` counts reaped subprocesses, so it is bursty; the first sample carries no cpu or delta fields)
`offload`	One `handler.offload()` call completed — CPU-bound hook work run on the offload thread pool; see Offload	`partition` (NULL outside hooks), `offset` / `task_id` (annotation-style anchoring), `duration`, `metadata` (hook, function, queued, status: ok/error/cancelled, window_id, offsets, error)
`periodic_run`	Periodic task execution completes	`task_id` (task name), `duration`, `exit_code` (0=ok, 1=error), `metadata` (status, error)
`webapp_request_received`	One HTTP request passed auth + rate-limit + body parsing and is about to dispatch to T1	`origin='http'`, `client_name`, `request_id`, `metadata` (started_at, body_bytes)
`webapp_request_completed`	An HTTP request returned 200 to the caller	`origin='http'`, `client_name`, `request_id`, `duration`, `metadata` (status=’ok’, duration_ms)
`webapp_request_timeout`	T2’s `asyncio.wait_for` tripped its `request_timeout_seconds` budget; client received a 504	`origin='http'`, `client_name`, `request_id`, `duration`, `metadata` (duration_ms)
`webapp_request_rate_limited`	Per-client rpm window full; client received a 429	`origin='http'`, `client_name`, `metadata` (rpm_limit, requests_in_window)
`webapp_request_auth_failed`	`Authorization` header missing or naming a non-configured token; client received a 401	`origin='http'`, `metadata` (token_prefix – redacted to first 4 chars)
`webapp_request_dropped_after_timeout`	T1 reached a cancellation gate after T2 had already 504’d. Marks pipeline work the framework managed to skip thanks to cooperative cancellation.	`origin='http'`, `client_name`, `request_id`

The /ws stream additionally carries a throughput frame — the three-window cost/speed aggregate behind the throughput strips. It is broadcast only and never written to events, so it has no row above: buffering a periodic aggregate would grow the table with rows no query reads back. A fixture pins the full vocabulary, stored and broadcast-only alike, against one shared fixture — see drakkar-ui/docs/api-contract-v1.md, “Recorder event-type vocabulary”.

For HTTP-origin rows, partition=-1 is used as a synthetic partition value so they remain distinct from any real Kafka partition without requiring a separate table. Filter by origin='http' to see all webapp activity, or by request_id=... to walk the full lifecycle of one HTTP request (received → task_started → task_completed → completed/timeout/dropped).

Fields subject to duration thresholds: args, stdout, stderr are omitted for fast tasks below output_min_duration_ms. Events below event_min_duration_ms are not stored at all. Stdin content is gated by store_stdin / stdin_max_bytes rather than by duration — failed tasks always store it. See Local Databases for the full retention-knob table.

Recorder upgrade story (delete pre-webapp DBs)

The webapp release added three columns to the events table – origin, client_name, request_id – without a migration framework. Recorder DBs are observability-only, already rotated and disposable, so a one-shot delete on upgrade is preferable to a migration runner that exists only to add three optional columns.

On upgrade, operators delete pre-existing per-worker recorder DBs in storage.dir before restarting workers. New rotation-cycle DBs include the new columns automatically.

The framework detects the schema mismatch at startup: the recorder runs PRAGMA table_info(events) immediately after opening each DB and raises RecorderSchemaError when origin / client_name / request_id are missing. The exception propagates through AppLifecycle._async_run and aborts startup with the actionable next step in the message: delete the offending DB(s) under storage.dir and restart. Without this fail-fast check, pre-webapp DBs would surface a confusing OperationalError: no such column mid-request on the first HTTP event the recorder tried to write.

`worker_config` – Worker identity¶

Single-row table written at startup (and after each rotation). Contains the full worker identity and configuration: worker_name, cluster_name, ip_address, debug_port, debug_url, kafka_brokers, source_topic, consumer_group, binary_path, max_executors, task_timeout_seconds, max_retries, window_size, sinks_json, env_vars_json. On a worker with the Kafka source disabled, source_topic and consumer_group are written empty; the column set does not change, so peers reading the file need no special case.

This table names the worker and cluster that wrote each file, for history, merges and archives.

Secrets are redacted before they reach disk

The recorder SQLite file is downloadable via the operator UI. To avoid publishing credentials through that path, three redactions are applied before any env data is written:

kafka_brokers — embedded credentials are stripped from SASL URIs. SASL_SSL://alice:s3cret@host:9094 becomes SASL_SSL://***:***@host:9094. Host and port remain visible.
worker_config.env_vars_json — values for env var names matching secret patterns (*PASSWORD*, *PASSWD*, *SECRET*, *TOKEN*, *KEY*, *API_KEY*, *CREDENTIAL*, *_DSN, *AUTH*, *PRIVATE*, *CERT*, *SALT*) are replaced with ***. For other names, any URL-shaped value has its embedded user:pass@ credentials stripped.
per-task env metadata — task.env values written by your handler in arrange() are sanitized with the same patterns before the task_started event is recorded. The original task object is not mutated; only the recorded copy is redacted (the subprocess still receives the real values). This closes the last env-related leak into the UI.

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. Rename the var if you need the value visible. The ExecutorConfig.env (framework-level) values are never written to the recorder at all — they only reach the subprocess environment.

These patterns are deliberately broader than executor.env_inherit_deny (see Parent-env filtering): an over-matched redaction here just costs *** in a downloadable debug DB, while an over-matched inheritance-deny pattern would withhold a variable a subprocess actually needs.

This protects operators who add secret-named vars to expose_env_vars without thinking, DSNs passed through otherwise-innocuous var names (UPSTREAM_URL, CACHE_URL, etc.), and per-task env values that handlers assemble from inbound message payloads.

`worker_state` – Periodic Snapshots¶

Appended every state_sync_interval_seconds (default: 10). Captures a point-in-time snapshot: uptime_seconds, assigned_partitions, partition_count, pool_active, pool_max, total_queued, cumulative counters (consumed_count, completed_count, failed_count, produced_count, committed_count), paused flag, health_state and loop_lag_ms from the runtime health monitor, and a throughput JSON column with the windowed task throughput snapshot (NULL when throughput.cost_label is unset).

File Layout and Rotation¶

The live database file follows the naming pattern:

{storage.dir}/{worker_name}-{timestamp}.db

For example: /tmp/drakkar/worker-1-2026-04-06__14_55_00.db. Only the worker that writes it opens it.

Rotation: every rotation_interval_hours (default: 1), the recorder opens a new timestamped DB file before closing the old one (no query gap). The worker config is re-written to the new file. The old file is then sealed: switched to a rollback journal, given a meta file with its statistics, and closed. A background job moves it into the history folder of the worker’s cluster:

{shared.dir}/history/{cluster}/{worker_name}-{timestamp}.db
{shared.dir}/history/{cluster}/{worker_name}-{timestamp}.db.meta.json

shared.dir empty means storage.dir. {cluster} is the cluster name with every character outside [a-zA-Z0-9_-] replaced by _ (default when it is empty). A clean stop seals the live file too, and moves it while shutdown time remains. A file whose seal or move fails stays in storage.dir and is retried on every rotation tick. The full rules — the cross-filesystem copy, the backlog, leftovers of a crashed process, the meta file format — are in Local Databases → Sealing. Rotation itself deletes nothing — see Archiving, next.

Archiving¶

Age- and count-based retention (the old retention_hours / retention_max_events keys) are gone. In their place, a periodic archive pass folds each finished UTC time window’s sealed files into one compressed, merged database per cluster (<cluster>-<from>__<to>.db.gz, in the same cluster history folder), and deletes only the sealed files it successfully merged — see Local Databases → Archiving for the full byte-level spec. The short version:

Defaults (ui.recorder.rotation_interval_hours: 1, and history.archive enabled: true, window_hours: 24, due_after_hours: 48, retention_days: 30): files rotate hourly and are sealed into the history folder; a UTC day’s window becomes due exactly 48h after it closes and is archived on the next hourly tick after that, merging its sealed files into one .db.gz per cluster and deleting them and their meta files. That 48h due delay is the safety margin, not a bug — but it means a single recorder file’s lifespan, counted from when it was created rather than from window close, can reach up to ~72h (up to 24h as part of its own window, plus the ~48h due delay after that window closes). Archives themselves are deleted once their window ended more than retention_days ago; 0 keeps them forever and warns at startup. Track the footprint with drakkar_recorder_archive_bytes.
Every worker runs the pass after each rotation tick, in the background, on the history folder of its own cluster only.
Windows key on file start time, not event time. A sealed file belongs to the window holding its own start timestamp; an archive can carry a handful of events timestamped slightly past its own window end.
Only sealed files are candidates. Every *.db in a cluster history folder is sealed and never changes, so the pass needs no in-use check and reads its sources with immutable=1.
One archiver per cluster. Every worker of the cluster tries to create history/<cluster>/.archive.lock with an exclusive create; one wins, and the others skip the tick and lose nothing. A lock older than history.archive.lock_stale_seconds (default 7200) belongs to a dead archiver and is taken over (recorder_archive_lock_taken_over); a fresh lock that two racing takers moved aside is put back, and recorder_archive_lock_put_back_failed reports when a third worker’s lock came first. Ages are measured on the file server’s clock. Right before it publishes, the holder checks that the lock still holds its token; if another worker took it over, the pass publishes and deletes nothing (recorder_archive_fenced).
Sources die last. A sealed file is deleted only once its archive is durably on disk. A file the merge cannot read is renamed to <name>.unreadable instead of being deleted, and a pre-existing archive at the final name is folded into the new one rather than overwritten. A pass that took the lock and failed counts in drakkar_recorder_archive_failures_total.
Merge results expire. The pass also deletes UI merge results in <shared.dir>/merged/ older than history.merged_retention_days (default 32; 0 keeps them), logging recorder_merged_pruned.
Opting out (history.archive.enabled: false) disables the pass entirely — nothing then deletes sealed recorder files or merge results automatically, and a startup log line (recorder_archiving_disabled, field folder) says so.
Renamed key: rotation_interval_minutes is now rotation_interval_hours, with a unit change (1 = 1 hour, not 1 minute). rotation_interval_minutes, retention_hours and retention_max_events are gone and are ignored if still present — see Renamed and removed keys.

storage:
  dir: /var/lib/drakkar
shared:
  dir: /mnt/fleet                    # history folders live here; '' = storage.dir
ui:
  recorder:
    rotation_interval_hours: 1       # 1 = every hour (was rotation_interval_minutes)
history:
  archive:
    enabled: true                    # merge sealed files into windowed .db.gz archives
    window_hours: 24                 # one archive per cluster per UTC day; must be >= rotation_interval_hours
    due_after_hours: 48              # archive a window this long after it ends
    retention_days: 30               # delete archives older than this; 0 = keep forever
  merged_retention_days: 32          # delete UI merge results older than this; 0 = keep them

Archives are downloaded from the same place as the other history files — Debug → Databases tab — backed by GET /api/v1/debug/archives (the archives of every cluster folder, each with folder and path) and GET /api/v1/debug/download/{path}, where path is relative to shared.dir (for example history/search/search-2026-10-12_00-00__2026-10-13_00-00.db.gz). See Merging Databases below for why archives never appear as merge candidates. A downloaded archive is a plain gzip file: gunzip it and the result is an ordinary merged recorder SQLite database, readable with the sqlite3 CLI or any tool that already reads a merged .db.

Two operational gotchas worth knowing:

A cluster with no running worker is never archived or expired. Each worker works only on its own cluster’s history folder, so a retired cluster’s sealed files and archives stay until you delete them.
A corrupt file at an archive’s own final name stalls that window. Every pass that considers the window fails to decompress it, logs recorder_archive_failed, and retries next tick — indefinitely, without touching the (safe, undeleted) sealed sources. Recovery is manual: remove or rename the bad file.

Buffer Mechanics¶

Events are not written to SQLite immediately. They accumulate in an in-memory ring buffer (collections.deque with maxlen=max_buffer, default 50,000). A background task flushes the buffer to disk every flush_interval_seconds (default: 5) as a batch INSERT. This design:

Avoids per-event disk I/O (thousands of events/sec would bottleneck SQLite)
Keeps the main event loop fast (recording an event is a deque append, ~1us)
Provides a short window of recent events even when storage.dir is empty (in-memory only mode)

If the buffer fills before the next flush (e.g., during a burst), oldest events are evicted (ring buffer behavior). Drops are no longer silent — the drakkar_recorder_dropped_events_total counter (see Recorder metrics) increments on every evicted event, and the drakkar_recorder_buffer_size gauge shows the current depth. Alert on sustained non-zero drops and raise max_buffer or shorten flush_interval_seconds when they appear.

The buffer is written in chunks of 5000 rows, and the loop is handed back between chunks: building a row tuple costs one lookup per column, so an unchunked flush of tens of thousands of events would stall the loop for tens of milliseconds and show up as loop lag in runtime health. A flush also starts early, before flush_interval_seconds elapses, once the buffer is half full — a burst is written out rather than left to reach max_buffer and evict.

A flush starts early on size as well as on count, once the buffered payload passes roughly 64 MiB. max_buffer bounds the buffer by event count, which says nothing about how much memory it holds: with store_output on, every task at or above output_min_duration_ms has its whole stdout and stderr buffered, and executor.max_stdout_bytes is unlimited by default. A handler producing hundreds of kilobytes per task therefore reaches gigabytes of resident buffer long before it reaches 25,000 events. The size trigger is not configurable: it is a safety floor rather than a tuning knob, and flushing early is never wrong. Nothing is truncated — the stored row keeps the whole output, which is what the task-detail page shows.

Live WebSocket clients never receive captured output — the live views show stdout_size, not the text — and the queued frames do not hold it either: the output is dropped when the event is handed to the fan-out, not when it is encoded. Otherwise a browser tab on a slow link could pin up to 10,000 events’ worth of full output in its subscriber queue, on top of the buffer above.

The chunk is also the unit of loss. A chunk that still fails after max_flush_retries consecutive attempts is dropped and drakkar_recorder_flush_batches_dropped_total ticks; the rest of the buffer stays and is retried on the next tick. Size max_buffer for how long a database stall you want to ride out:

max_buffer >= event_rate x flush_interval_seconds x (max_flush_retries + 1)

At 5000 events/s with the defaults (5 s interval, 3 retries) that is 100 000, twice the default max_buffer. Raise it if your workers must survive a 15-second SQLite stall — the usual cause is a network filesystem — without losing a chunk.

Buffering happens only when the worker will actually write the events. In memory-only mode (storage.dir empty) and with store_events: false there is no flush loop, so events are streamed to the live WebSocket view and not buffered at all: the buffer gauge stays at 0 and the drop counter never moves. A non-zero drop count therefore always means real flush pressure.

On shutdown, the buffer is flushed one final time before closing the database.

Merging Databases¶

The UI’s database section (/debug) allows selecting multiple history files — sealed files and earlier merge results — and merging them into a single file. This is useful for:

Post-mortem analysis – combine files from multiple workers and time periods into one queryable database
Cross-worker correlation – merged files contain events from all selected workers, with worker_name preserved in the worker_config table
Archival – download a merged file for offline analysis with any SQLite tool

POST /api/v1/debug/merge takes paths relative to shared.dir (history/<cluster>/<name>.db or merged/<name>.db). The merge process copies all events from the selected files into a new file <shared.dir>/merged/merged-<YYYY-MM-DD__HH_MM_SS>-<16 hex>.db; the answer carries its path. It does not remove duplicates. The workers table lists every source worker; a selected merged file keeps its own workers, events and states. The file is written under a hidden temporary name and appears in the database list only when it is complete. It can be downloaded. merged/ is created with mode 0700, and the archive pass deletes merge results older than history.merged_retention_days (default 32 days; 0 keeps them).

A worker’s own files in storage.dir — the live file and the backlog — are not merge sources and are not downloadable: use a file once it is sealed into the history folder.

Archives are never merge candidates. The Databases tab’s archives (see Archiving above) are read-only — no selection checkboxes, no path into /api/v1/debug/merge. POST /api/v1/debug/merge accepts only .db paths, so an archive path passed to it by hand answers 400.

Duration Thresholds¶

Drakkar provides four independent duration thresholds that control what gets logged, broadcast, and stored. These are essential for high-throughput workers where fast tasks would otherwise flood every observability channel.

ui:
  log_min_duration_ms: 500
  ws_min_duration_ms: 500
  recorder:
    event_min_duration_ms: 0
    output_min_duration_ms: 500

`log_min_duration_ms` (default: 500)¶

Controls which completed/failed tasks produce a structlog entry. Tasks finishing faster than this threshold are silent in the log stream. Keeps logs focused on slow or problematic tasks.

`ws_min_duration_ms` (default: 500)¶

Controls which tasks appear in the live UI (/live page). When a task starts, its WebSocket broadcast is deferred. If the task completes before the threshold, neither the start nor the completion event is sent to connected browsers – the task is invisible in the live view.

Exception: failed tasks are always sent to WebSocket regardless of duration. This ensures failures are never hidden.

`event_min_duration_ms` (default: 0)¶

Controls which tasks are persisted to the SQLite flight recorder. When set to a value greater than 0, tasks completing faster than the threshold are not written to the events table. Set to 0 to store all events (the default).

This is useful for workers that process millions of fast tasks where full event recording would consume excessive disk space.

`output_min_duration_ms` (default: 500)¶

Controls whether args, stdout, and stderr are stored for a task. Tasks completing faster than this threshold have their output omitted from the event record – the event itself is still recorded (subject to event_min_duration_ms), but without the potentially large output fields.

Reduces storage for fast tasks where the output is uninteresting.

Worker Autodiscovery¶

A shared folder between workers is not required for Drakkar to function. Each worker operates independently – its own Kafka consumer, executor pool, sinks, and flight recorder work fine with a local directory.

You can also disable the flight recorder database entirely by setting storage.dir: "" (empty string). The web UI and WebSocket live streaming still work (events are held in memory only), but these features are disabled: event history (/history), message trace, label search, database download/merge and periodic task history. Worker discovery needs a card folder for the fleet registry: with storage.dir: "" set shared.dir (or another registry provider) to keep the workers list. The live pipeline view (/live) and dashboard counters continue working since they read from in-memory state.

When workers share one shared.dir (e.g., a shared NFS mount, Kubernetes PVC, or Docker volume; with shared.dir empty, one shared storage.dir), several additional UI features become available:

Worker switcher – jump between worker UIs from a dropdown in the nav bar, without remembering ports or IP addresses (the switcher comes from the fleet registry, whose default card folder is in the shared folder)
Cross-worker message trace – trace a message by partition:offset through the sealed history of every worker in the cluster
Database merge – combine sealed recorder files from multiple workers into a single SQLite file for post-mortem analysis
Cluster-wide database browser – see every cluster’s history folder (sealed files, archives, merge results) in one view

Without a shared folder, each worker’s UI only sees its own data. All other functionality (processing, sinks, metrics, logging) works identically.

How It Works¶

Workers find each other through the fleet registry:

Each worker publishes a small card – worker name, cluster, the URL of its UI server, state (starting, running, draining), release – right after its UI server starts, at every state change, and every fleet.registry.heartbeat_seconds (default 10).
With the default filesystem provider the card is a file in <shared.dir>/fleet/workers/ (<storage.dir>/fleet/workers/ when shared.dir is empty), so workers that mount the same folder see each other.
Every fleet.registry.refresh_seconds (default 15) a worker reads the other cards into memory. GET /api/v1/workers answers from that snapshot; it never opens a peer’s database.
Workers are grouped by cluster name – only workers in the same cluster appear together. Unclustered workers show in a separate group.
At a clean stop a worker deletes its card.

Liveness¶

A crashed or OOM-killed worker never deletes its card, so it stays listed until the card is pruned (fleet.registry.prune_after_seconds, default one day). To tell such workers apart from healthy ones, every row of /api/v1/workers carries two liveness fields:

online – true when the peer’s card is no older than fleet.registry.offline_after_seconds (default 30), false when it is older, and null when the provider cannot tell (a static peer list, or a worker whose own first card write has not succeeded yet). The worker serving the request always reports itself online.
last_seen_ts – the serving worker’s current time minus the card’s age, in epoch seconds; null when the age is unknown.

The age is measured on the file server’s clock, so clock skew between worker hosts does not matter (see Liveness). An offline worker stays listed – its recorded history is still reachable – it is just marked as gone.

Worker Switcher¶

The UI navigation bar includes a worker dropdown that lists all discovered workers, grouped by cluster. Clicking a worker navigates to the URL on its card: ui.public_url if configured, otherwise http://<outbound ip>:<ui.port>/. A worker that is not running shows its state, and every worker shows its release when the card carries it. The dropdown refreshes every 10 seconds.

Setup¶

The minimal config for worker discovery:

# every worker mounts the same folder; the registry keeps its cards there
shared:
  dir: "/shared"
# optional: set cluster_name to group workers
cluster_name: "my-cluster"

Without shared.dir, workers that share one storage.dir see each other too (the cards then live in <storage.dir>/fleet/workers/).

In Kubernetes, use a shared PVC:

volumes:
  - name: recorder-shared
    persistentVolumeClaim:
      claimName: drakkar-recorder

# mount on all worker pods (the mount path may differ per pod)
volumeMounts:
  - name: recorder-shared
    mountPath: /shared

In Docker Compose, use a named volume:

volumes:
  shared: {}

services:
  worker-1:
    volumes:
      - shared:/shared
  worker-2:
    volumes:
      - shared:/shared

Cross-Worker Trace¶

The trace feature on the /debug page searches for a message’s lifecycle in the current worker’s files and, on request, in the sealed history of every worker in the cluster. Two search modes are available:

By partition:offset – given a partition and offset, finds the consumed event and all related events (task_started, task_completed, task_failed, task_complete, message_complete, produced, committed).

By label – given a label key and value (e.g., request_id=abc-123), finds all tasks whose labels match and returns their full event lifecycle. This is useful when you know a business identifier but not which partition or offset it landed on. Label search inputs appear automatically on the /debug page for each label key found in the database.

What each mode searches:

By partition:offset, by default: the current worker’s own files – its live database and its closed files in storage.dir that are not yet sealed into the history folder.
By partition:offset with history=1: only the sealed files in the history folder of the worker’s own cluster (<shared.dir>/history/<cluster>/), newest first. The search stops after ui.trace.history_max_files files (default 256) or ui.trace.history_max_seconds (default 10). Sealed files never change, so the search opens them with immutable=1 and takes no locks.
By label: the current worker’s live database only.

No worker opens another worker’s live database. Events of other workers come from their sealed files, or from the workers themselves: the UI sends the trace, label-trace and databases requests to every worker of the cluster that is not offline, and lists the workers that did not answer as unreachable. Those reads are cross-origin. With ui.auth_token set, list the origin of every worker in ui.allowed_ws_origins, or only the serving worker answers (see Cross-origin reads).

Each returned event carries a worker_name field identifying which worker processed it. Labels are stored as JSON in the labels column with a partial index (WHERE labels IS NOT NULL) for efficient filtering.

Periodic Tasks¶

The @periodic decorator schedules handler methods to run at fixed intervals alongside the main processing loop. See also Periodic Tasks in the handler reference.

Declaration¶

from drakkar.periodic import periodic

class MyHandler(BaseDrakkarHandler[MyInput, MyOutput]):
    @periodic(seconds=60)
    async def refresh_cache(self):
        """Reload lookup table from database every minute."""
        self.lookup = await load_lookup_table()

    @periodic(seconds=300, on_error='stop')
    async def health_check(self):
        """Check downstream service health. Stop if it fails."""
        await ping_downstream()

Behavior¶

Discovery: at startup, the framework inspects the handler instance for methods decorated with @periodic. Only async functions are accepted – decorating a sync function raises TypeError.
Scheduling: each periodic task runs in its own asyncio.Task within the main event loop. The loop is: sleep(seconds) then await coro_fn(). The next interval starts only after the current invocation finishes, preventing overlapping runs.
Lifecycle: periodic tasks are started after on_ready() completes and cancelled during graceful shutdown (before partition processors are drained).
Error handling:
- on_error='continue' (default) – the exception is logged, and the task continues scheduling on the next interval.
- on_error='stop' – the exception is logged, and this specific periodic task is permanently cancelled. Other periodic tasks and the main processing loop are unaffected.

Observability¶

Prometheus Metrics¶

Metrics Reference¶

Worker Identity¶

Messages¶

Executor¶

Batches¶

Partitions¶

Consumer¶

Offsets¶

Rebalancing¶

Sinks¶

Handler Hooks¶

Periodic tasks¶

Offload pool¶

Recorder¶

Cache¶

Webapp¶

Shutdown¶

OOM / SIGKILL detection¶

User-Defined Metrics¶

Scrape Configuration¶

Structured Logging¶

Configuration¶

Output Destination¶

Automatic Context Fields¶

Per-Context Binding¶

Usage in Handler Code¶

Operator UI¶

Pages¶

/ – Dashboard¶

/partitions – Partition Overview¶

/partitions/{id} – Partition Detail¶

/sinks – Sink Delivery Stats¶

/live – Live Pipeline View¶

/history – Event Browser¶

/debug – Debug Tools¶

/task/{id} – Task Detail¶

API¶

Managing the UI bundle cache¶

Flight Recorder¶

Configuration¶

Database Schema¶

events – Processing Events¶

worker_config – Worker identity¶

worker_state – Periodic Snapshots¶

File Layout and Rotation¶

Archiving¶

Buffer Mechanics¶

Merging Databases¶

Duration Thresholds¶

log_min_duration_ms (default: 500)¶

ws_min_duration_ms (default: 500)¶

event_min_duration_ms (default: 0)¶

output_min_duration_ms (default: 500)¶

Worker Autodiscovery¶

How It Works¶

Liveness¶

Worker Switcher¶

Setup¶

Cross-Worker Trace¶

Periodic Tasks¶

Declaration¶

Behavior¶

`/` – Dashboard¶

`/partitions` – Partition Overview¶

`/partitions/{id}` – Partition Detail¶

`/sinks` – Sink Delivery Stats¶

`/live` – Live Pipeline View¶

`/history` – Event Browser¶

`/debug` – Debug Tools¶

`/task/{id}` – Task Detail¶

`events` – Processing Events¶

`worker_config` – Worker identity¶

`worker_state` – Periodic Snapshots¶

`log_min_duration_ms` (default: 500)¶

`ws_min_duration_ms` (default: 500)¶

`event_min_duration_ms` (default: 0)¶

`output_min_duration_ms` (default: 500)¶