Event Reliability and Loss Prevention

Data Index uses stdout-based log collection with Vector. This guide describes potential event-loss scenarios and how Vector’s design and configuration mitigate them.

Option reference: vector.dev/docs/reference/configuration/.

Event Flow and Guarantees

Quarkus Flow App
    ↓ (stdout write - OS buffer)
Kernel / container runtime
    ↓ (write to disk)
/var/log/containers/<pod>_<namespace>_<container>.log
    ↓ (Vector kubernetes_logs - offset checkpointed in data_dir)
Vector transforms (parse_json -> filter -> route -> build row)
    ↓
Vector per-sink buffer (MODE 1: in-memory; MODE 2: disk)
    ↓ (postgres / elasticsearch sink - retries with backoff)
Storage Backend (PostgreSQL raw tables / Elasticsearch raw indices)
    ↓ (Normalization: PostgreSQL triggers / ES Transforms)
Normalized Data

Critical points of failure:

  1. App crash before stdout is written

  2. Node termination before the log line reaches disk

  3. Node deletes rotated log files before Vector finishes reading them

  4. Sustained sink back-pressure stalls the source

  5. Vector crash between "sent to storage" and "checkpoint written" (duplication, not loss)

  6. Storage backend unavailable long enough to exhaust the buffer

  7. Malformed JSON dropped by parse_json

Event Loss Scenarios

1. Application Crashes Before Stdout Is Written

Scenario: the app produces a workflow.started event, then the process dies before the line is flushed to its stdout stream.

Risk: Low. Quarkus Flow structured-logging lines are newline-terminated and the JSON console handler is not buffered across lines. If the workflow itself failed mid-execution it will be re-run (idempotent).

kubectl get pods -n workflows --field-selector=status.phase=Failed

2. Node Termination Before Disk Write

Scenario: node gets SIGTERM, then SIGKILL after the grace period, before an in-flight stdout line is persisted to /var/log/containers/.

Risk: Medium.

Mitigation: give workflow pods a longer termination grace period and a small preStop sleep so the runtime flushes stdout:

spec:
  terminationGracePeriodSeconds: 60
  containers:
    - name: workflow-app
      lifecycle:
        preStop:
          exec:
            command: ["/bin/sh", "-c", "sleep 5"]
kubectl get events --field-selector reason=Evicted

3. Rotated Log Files Deleted Before Vector Reads Them

Scenario:

kubelet rotates /var/log/containers/pod.log  (size > containerLogMaxSize)
  pod.log -> pod.log.1 ; new pod.log created
Vector is still reading pod.log.1
kubelet deletes pod.log.6 (containerLogMaxFiles reached)
-> events in pod.log.6 are lost if Vector never opened it

Risk: Low-to-Medium. Vector’s kubernetes_logs source uses oldest_first: true and a very large rotate_wait_secs, so it keeps reading a rotated file to EOF instead of abandoning it on rotation. Loss only happens if the node deletes rotated files faster than Vector drains them (i.e. Vector is badly behind — see scenario 4).

Mitigation - node log retention (cluster-admin, per node):

# /var/lib/kubelet/config.yaml
containerLogMaxSize: 100Mi    # Default: 10Mi
containerLogMaxFiles: 10      # Default: 5

Mitigation - make Vector notice rotation and read faster (in vector.yaml):

sources:
  kubernetes_logs:
    type: kubernetes_logs
    glob_minimum_cooldown_ms: 5000   # re-scan for new/rotated files every 5s (default 60000)
    max_read_bytes: 16384            # bytes per read op (default 2048)

Detect Vector falling behind:

kubectl exec -n logging <vector-pod> -- ls -la /tmp/vector   # offset checkpoints
curl -s http://<vector-pod>:9598/metrics | grep -E 'vector_component_received_events_total|vector_file_'
rate(vector_component_received_bytes_total{component_id="kubernetes_logs"}[5m]) > 1000000   # > 1 MB/s sustained

4. Sustained Sink Back-Pressure Stalls the Source

Scenario: the sink can’t keep up (slow storage, big batch of retries). The per-sink buffer fills; with the default when_full: block Vector stops pulling from kubernetes_logs. Vector does not drop events here - but if the stall lasts long enough that logs rotate and get deleted (scenario 3), those are lost.

Risk: Medium under sustained overload.

Mitigation - larger / disk-backed buffer (per sink in vector.yaml):

sinks:
  postgres_workflow:
    # ...
    buffer:
      type: disk           # survives Vector restart; MODE 2 already uses this
      max_size: 268435456  # 256 MiB
      when_full: block
MODE 1 ships the default in-memory buffer on purpose - in Vector 0.54 the postgres sink does not drain a small disk buffer until graceful shutdown. Raise buffer.max_events (memory) before reaching for type: disk, and test.

Mitigation - drop instead of block (explicit trade-off):

    buffer:
      when_full: drop_newest   # prefer losing newest events over stalling the pipeline

Mitigation - scale sink throughput:

    request:
      concurrency: adaptive      # default; or a fixed integer
    batch:
      max_events: 500            # fewer, larger INSERTs / bulk requests
      timeout_secs: 1

Monitor:

vector_buffer_events{component_id="postgres_workflow",stage="0"}          # pending in buffer
vector_buffer_byte_size{component_id="postgres_workflow"}
rate(vector_component_utilization{component_id="postgres_workflow"}[5m])   # ~1.0 => saturated

5. Vector Crash Between Send and Checkpoint

Scenario:

Vector reads events from pod.log and sends them to storage (success)
Vector crashes before persisting the new file offset in data_dir
Vector restarts, resumes from the last persisted offset
-> the same events are read and sent again  (duplication, not loss)

Risk: Low - duplicates are harmless:

  • MODE 1: normalize_workflow_event() uses INSERT …​ ON CONFLICT (id) DO UPDATE; normalize_task_event() upserts on (instance_id, task). Re-inserting a raw row just re-runs the idempotent trigger.

  • MODE 2: ES Transforms aggregate by document key; re-processing converges.

Optionally enable end-to-end acknowledgements so the source offset only advances after the sink confirms the write (reduces the duplication window):

sources:
  kubernetes_logs:
    type: kubernetes_logs
    acknowledgements:
      enabled: true
kubectl get pods -n logging -l app=vector | grep -i CrashLoopBackOff

6. Storage Backend Unavailable

Scenario: PostgreSQL / Elasticsearch refuses connections or times out.

Vector behaviour (defaults): the sink retries the request with exponential backoff - retry_attempts is effectively unlimited, retry_initial_backoff_secs: 1, capped per-request by retry_max_duration_secs: 30. Vector keeps retrying across the whole outage; it does not "give up after N tries". While it retries, the buffer fills and back-pressure applies (scenario 4).

Risk: Low for short outages; Medium for long outages on an in-memory buffer.

Mitigation - more runway during outages:

sinks:
  postgres_workflow:
    request:
      retry_initial_backoff_secs: 1
      retry_max_duration_secs: 120     # allow longer per-request retry windows
    buffer:
      type: disk                       # MODE 2 default; consider for MODE 1 if outages are expected
      max_size: 536870912              # 512 MiB

Mitigation - storage HA: PostgreSQL (Patroni / CloudNativePG / PgBouncer), Elasticsearch (multi-node with replicas). Monitor storage health directly.

7. Malformed JSON

Scenario: an app writes a truncated / non-JSON line. parse_json fails and `abort`s the event - it is intentionally dropped (this is how regular INFO/DEBUG app logs are filtered out).

Risk: Low - Quarkus Flow structured logging emits well-formed JSON.

Capture instead of drop (for debugging): send unparseable lines to the debug console sink rather than aborting.

transforms:
  parse_json:
    type: remap
    inputs: [kubernetes_logs]
    source: |
      parsed, err = parse_json(.message)
      if err != null {
        .parse_error = true          # keep the raw line, tag it
      } else {
        .flow_event = parsed
      }

  # route .parse_error == true to debug_stdout alongside DEBUG_EVENTS
kubectl logs -n logging -l app=vector | grep -iE 'parse_json|VRL|remap.*error'
curl -s http://<vector-pod>:9598/metrics | grep vector_component_errors_total

Reliability Guarantees

Vector provides:

  • At-least-once delivery - the source offset is checkpointed in data_dir; with acknowledgements.enabled: true it only advances after the sink confirms.

  • Automatic retries - unlimited attempts with exponential backoff on transient sink failures.

  • Back-pressure, not silent drop - when_full: block (default) stalls the source rather than discarding buffered events.

  • Crash recovery - resumes from the last persisted offset; duplicates are absorbed by the idempotent triggers / transforms.

Vector does NOT guarantee:

  • Ordering - events can arrive out of order; normalization is order-independent (field-level idempotency / terminal-state precedence).

  • Zero loss on node termination - stdout lines not yet on disk are lost.

  • Unbounded buffering - buffers have limits; a long outage on an in-memory buffer eventually applies back-pressure and, if logs then rotate away, loses data.

  • Cross-node durability - offset checkpoints live in the node’s emptyDir; losing the node loses its checkpoint (re-read from beginning on a fresh pod).

Production Recommendations

Baseline (shipped defaults)

  • kubernetes_logs with checkpointing in data_dir

  • Sinks: in-memory buffer (MODE 1) / disk buffer (MODE 2), unlimited retries, batch.timeout_secs: 1

  • Expected loss: negligible under normal conditions; bounded by scenario 2 (node kill) and scenario 4 (sustained overload).

Hardened

sources:
  kubernetes_logs:
    type: kubernetes_logs
    glob_minimum_cooldown_ms: 5000
    max_read_bytes: 16384
    acknowledgements:
      enabled: true

sinks:
  postgres_workflow:      # and postgres_task
    request:
      retry_max_duration_secs: 120
    batch:
      max_events: 500
      timeout_secs: 1
    buffer:
      type: disk
      max_size: 536870912   # 512 MiB
      when_full: block

Plus node-level containerLogMaxSize: 100Mi / containerLogMaxFiles: 10, and terminationGracePeriodSeconds: 60 on workflow pods.

Near-zero loss

If any loss is unacceptable, bypass log collection:

  • MODE 3 (Kafka) - Quarkus Flow publishes CloudEvents to Kafka; Data Index consumes them directly with a dead-letter topic. No stdout, no log files.

  • App-level dual write - the app writes to storage directly in addition to stdout (couples the app to Data Index).

Monitoring and Alerting

Key Metrics (Vector internal_metrics on :9598)

# Vector up
up{job="vector"}

# Events in vs out (should track closely)
rate(vector_component_received_events_total{component_id="kubernetes_logs"}[5m])
rate(vector_component_sent_events_total{component_id=~"postgres_.*"}[5m])

# Buffer pressure
vector_buffer_events{component_id=~"postgres_.*",stage="0"}
vector_component_utilization{component_id=~"postgres_.*"}      # ~1.0 => saturated

# Retries / errors
rate(vector_component_errors_total[5m])

Data-side sanity check (MODE 1)

SELECT
  (SELECT COUNT(*) FROM workflow_events_raw) AS raw_workflow_events,
  (SELECT COUNT(*) FROM workflow_instances)  AS workflows,
  (SELECT COUNT(*) FROM task_events_raw)     AS raw_task_events,
  (SELECT COUNT(*) FROM task_instances)      AS tasks;

Recommended Alerts

  • Critical: Vector pod not Running; vector_component_errors_total rising for a sink; buffer at capacity (vector_buffer_events flat at max).

  • Warning: sink vector_component_utilization > 0.8 for 10m; source ingest rate > 1 MB/s sustained; retry rate > 10/min.

Event Loss Detection (MODE 1)

Workflows that started but never finished

SELECT id, name, status, started_at, ended_at
FROM workflow_instances
WHERE status = 'RUNNING'
  AND started_at < NOW() - INTERVAL '1 hour';

Task count mismatch

-- 'simple-set' is expected to have 2 task executions
SELECT wi.id, wi.name, COUNT(ti.task) AS task_count
FROM workflow_instances wi
LEFT JOIN task_instances ti ON ti.instance_id = wi.id
WHERE wi.name = 'simple-set'
GROUP BY wi.id, wi.name
HAVING COUNT(ti.task) <> 2;

Disaster Recovery

1. Check Vector for drops / errors

kubectl logs -n logging -l app=vector --tail=2000 | grep -iE 'error|fail|drop|buffer'
curl -s http://<vector-pod>:9598/metrics | grep -E 'vector_component_(errors|discarded)_total'

2. Are the events still in the container logs?

kubectl exec -n logging <vector-pod> -- sh -c 'grep eventType /var/log/containers/*_workflows_*.log* | tail -100'

(the Vector image is distroless - if sh is unavailable, read the files from a debug pod or the node.)

3. Manual replay into the raw tables (MODE 1) - the triggers normalize on INSERT:

kubectl exec -n logging <vector-pod> -- \
  sh -c "grep 'io.serverlessworkflow.workflow' /var/log/containers/<pod>.log.2" > missed.jsonl

while IFS= read -r ev; do
  kubectl exec -i -n postgresql postgresql-0 -- \
    psql -U dataindex -d dataindex -c \
    "INSERT INTO workflow_events_raw (tag, time, data) VALUES ('replay', NOW(), '$(printf '%s' "$ev" | sed "s/'/''/g")'::jsonb);"
done < missed.jsonl

Summary

stdout-based collection can lose events at the edges (node kill before disk write; sustained overload that outruns log rotation). Vector’s defaults - keep reading rotated files, unlimited retries, block-on-full - keep this small. For workloads where any loss is unacceptable, use MODE 3 (Kafka) instead of log collection.