Data & Pipelines

Data flow through an inference gateway

The mental picture for the senior interview:

client
  │  HTTP/gRPC request
  ▼
[edge: TLS, request ID, auth]
  │
  ▼
[gateway: rate-limit, validate, route]──▶[metrics: counters/histograms]
  │                                       (Prometheus / OTLP)
  │
  ▼
[backend pool: vLLM / Triton / Anthropic]
  │
  ├──▶ [structured logs: JSON lines]
  │     (Vector / Fluent Bit -> Loki / Elastic)
  │
  ├──▶ [traces: OTel spans]
  │     (OTel Collector -> Tempo / Jaeger)
  │
  └──▶ [audit events: append-only]
        (Kafka topic / Postgres)
  ▼
response stream back to client

Three parallel data planes leave every request: metrics (aggregated), logs (per-event), traces (per-span). Plus the audit event plane for compliance-relevant actions.

Trace propagation

Already covered structurally in 09. The data-pipeline view:

• Process-internal: spans linked via the tracing crate context.
• Across processes (HTTP/gRPC): W3C trace context headers (traceparent, tracestate).
• Through queues (Kafka): trace context as a Kafka message header. The consumer continues the trace.
• Sampling: head-based vs tail-based. Tail-based (decide to keep a trace after seeing its outcome — e.g., always keep errors) is more useful but costs more.
• Storage: Tempo, Jaeger, Honeycomb, Datadog. Retention is short by default (days); important traces archived separately.

Metrics pipeline (Prometheus)

Prometheus pull-based scraping is the default at fintechs. Your Rust service exposes /metrics in Prometheus text format; a scraper polls every 15s. From there:

service /metrics ──▶ Prometheus ──▶ Cortex/Thanos (long-term, multi-cluster) ──▶ Grafana
                  │
                  └──▶ Alertmanager ──▶ PagerDuty / Slack

In Rust, use prometheus or metrics + metrics-exporter-prometheus:

use metrics::{counter, histogram};

pub async fn handle(req: GenerateReq) {
    let t = std::time::Instant::now();
    let result = call_backend(&req).await;
    let status = if result.is_ok() { "ok" } else { "err" };
    counter!("requests_total", 1, "model" => req.model.clone(), "status" => status);
    histogram!("request_seconds", t.elapsed().as_secs_f64(),
        "model" => req.model, "status" => status);
}

Cardinality discipline: don't use unbounded labels (user_id, request_id) on metrics. Prometheus stores one time series per label combination; unbounded labels OOM the scraper.

Log aggregation

Service writes JSON-line logs to stdout. A node-level agent (Vector, Fluent Bit, Filebeat) ships them to a backend (Loki, Elasticsearch, CloudWatch, BigQuery).

• Don't log to a file the service manages — let the platform do it. Stdout/stderr is the k8s contract.
• Structured JSON. Every log line is parseable. tracing-subscriber's fmt::layer().json() does this.
• Sampling for high-volume logs. 100% sampling for errors; lower for routine info.
• Don't log full prompts/responses. Log hashes + length + cost. Store bodies separately with access controls.
• Sensitive-field redaction happens at the log layer if anything makes it through the application layer.

Event sourcing for agent actions

An agent run is naturally event-sourced: the state is derived by replaying a sequence of events (RunStarted, ModelCalled, ToolCalled, ApprovalGranted, StepCompleted, RunCompleted). For audit and replay, this is gold.

• Events are append-only. Never updated, never deleted (within retention).
• State is a projection. The "current run state" is what you get by folding the event stream.
• You can replay. Given the events, you can rebuild any state at any point. Critical for audit ("what did this look like at 2:14pm?").
• Storage: Kafka for hot stream; periodic snapshot to S3; Postgres index for fast "give me events for run R" queries.

#[derive(serde::Serialize, serde::Deserialize)]
#[serde(tag = "type")]
pub enum RunEvent {
    Started { run_id: String, user: String, policy: String, at_ms: u64 },
    ModelCalled { step: u32, prompt_hash: String, resp_hash: String, model: String, cost_usd: f32 },
    ToolCalled { step: u32, tool: String, idem_key: String, args_hash: String, result_hash: String, ok: bool },
    ApprovalRequested { step: u32, tool: String },
    ApprovalDecided { step: u32, approver: String, decision: bool },
    Completed { final_status: String, total_cost_usd: f32 },
    Failed { error: String, retryable: bool },
}

Postgres for run state

The durable home for orchestration state. Schema sketch:

CREATE TABLE runs (
  run_id      UUID PRIMARY KEY,
  user_id     TEXT NOT NULL,
  policy      TEXT NOT NULL,
  status      TEXT NOT NULL,  -- pending|running|awaiting_approval|complete|failed|cancelled
  step_count  INTEGER NOT NULL DEFAULT 0,
  budget_usd  REAL NOT NULL,
  spent_usd   REAL NOT NULL DEFAULT 0,
  deadline    TIMESTAMPTZ NOT NULL,
  state       JSONB NOT NULL,     -- current messages, partial outputs
  version     BIGINT NOT NULL DEFAULT 0, -- optimistic concurrency
  created_at  TIMESTAMPTZ NOT NULL DEFAULT NOW(),
  updated_at  TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE TABLE run_events (
  event_id    BIGSERIAL PRIMARY KEY,
  run_id      UUID NOT NULL REFERENCES runs(run_id),
  seq         INTEGER NOT NULL,
  event       JSONB NOT NULL,
  at_ms       BIGINT NOT NULL,
  UNIQUE (run_id, seq)
);

CREATE INDEX run_events_run_id ON run_events (run_id, seq);

CREATE TABLE idempotency (
  key         UUID PRIMARY KEY,
  scope       TEXT NOT NULL,    -- (run_id, step_id) or similar
  status      TEXT NOT NULL,    -- in_progress | done | failed
  result      JSONB,
  created_at  TIMESTAMPTZ NOT NULL DEFAULT NOW(),
  expires_at  TIMESTAMPTZ NOT NULL
);

Patterns:

• Optimistic concurrency via version: UPDATE runs SET ... , version=version+1 WHERE run_id=$1 AND version=$2. If 0 rows updated, retry.
• Connection pool via sqlx or deadpool-postgres. Size: typically 10-50 per service replica; tune by load.
• Migrations via sqlx-cli or refinery. Versioned, forward-only.
• Read replicas for "list my runs" queries; primary for state mutations.

Redis for ephemeral state

Postgres is durable but not fast enough for hot path. Redis fills the gap:

• Rate-limit token buckets — INCR/EXPIRE or Lua script.
• Idempotency in-progress markers with short TTL (durable record goes to Postgres on completion).
• Distributed locks (with caution — Redlock is debated; for stronger guarantees use a leased lock in Postgres or a real consensus store).
• Cached prompt-prefix → backend routing decisions.
• Pub/sub for kill-switch and cancellation broadcasts.
• Per-tenant counters for cost accounting (eventually consistent with the source of truth in Postgres).

Use deadpool-redis or fred for async Redis in Rust. fred has more advanced cluster/sentinel support; deadpool is simpler.

Kafka / streaming pipelines

For event streams that need to be consumed by multiple downstream systems (analytics, eval-data collection, audit archive, ML training pipelines):

• Topic per event family. agent.runs.events, gateway.requests, tool.calls.
• Schema registry. Avro / Protobuf with versioning. Don't ship free-form JSON.
• Partitioning by run_id for run events (preserves order per run).
• Retention per topic — short for hot streams, long for audit.
• Consumer groups for parallel processing.
• Exactly-once delivery is a property of the consumer + idempotent sink, not the broker.

In Rust: rdkafka (librdkafka bindings) is the mainstream client.

Data quality at the boundaries

Garbage in, garbage out. The places to enforce contracts:

1. Public API request validation. Strict schemas; reject malformed.
2. Tool args validation. JSON schema, validated before dispatch.
3. Model response shape. If you expect JSON, parse strictly. On failure, repair-prompt or escalate.
4. Event publishing. The event you publish must conform to the schema-registry contract. Fail fast on bad events, don't poison the stream.
5. DB writes. Postgres CHECK constraints, foreign keys. Make the DB the last line of defense against corruption.

✓ Check yourself

Answer from memory before opening each one.

1. You can afford to keep about 5% of traces. How do you make sure the failing requests are among those kept, and what does your method cost?

   Show answer

   Sample on the outcome: always keep errors and sample 1 to 10 percent of successes. That is tail-based sampling, which decides after the trace completes, so the collector must buffer whole traces first, costing memory and complexity. Head-based sampling is cheap but decides before it knows the result. See Trace propagation.

2. A colleague adds a user_id label to the request counter to find noisy users. What happens, and where does that detail belong?

   Show answer

   Prometheus stores a time series per label combination, so an unbounded label multiplies series until the scraper runs out of memory. Keep metric labels to bounded sets such as model and status, and put per-user detail in logs and traces. See Metrics pipeline.

3. Agent-run traces break at the Kafka hop: the consumer's spans start new, unrelated traces. What is missing?

   Show answer

   Trace context was not carried in the message. Put the traceparent value in a Kafka message header on the producer side, and have the consumer continue the trace from it. The same rule applies to any queue. See Trace propagation.

4. An auditor asks what run R looked like at 2:14pm. How do you answer from an event-sourced store, and what keeps events for one run in order on Kafka?

   Show answer

   Fold the run's events up to that timestamp, since state is a projection of an append-only stream. Partitioning the topic by run_id keeps one run's events in order. See Event sourcing and Kafka.

5. Two executors update the same run row at the same moment. Describe the statement and what the loser sees.

   Show answer

   Each issues an UPDATE that increments the version column and filters on the version it read. The first changes one row. The second matches zero rows, learns its copy is stale and re-reads before trying again. See Postgres for run state.

6. Where do you store an in-progress idempotency marker, the final result and a distributed lock, and what do you do when a lock must be strongly safe?

   Show answer

   The marker goes in Redis with a short TTL because it is hot and loss is tolerable. The final result goes in Postgres as the durable record. For a lock that must be strongly safe, avoid relying on Redis alone and use a leased lock in Postgres or a consensus store, since Redlock's guarantees are debated. See Redis for ephemeral state.

Takeaway

• Every request leaves four planes of data: metrics for aggregates, logs per event, traces per span and an append-only audit plane for compliance.
• Keep cardinality in logs and traces and out of metric labels; sample errors always and successes at a low rate.
• Carry trace context everywhere a request can cross a boundary, including Kafka message headers.
• Model agent runs as an append-only event stream with state as a projection, which gives audit and replay for free.
• Postgres holds run state with a version column for optimistic concurrency; Redis holds hot, ephemeral state, with the durable record written to Postgres.
• Exactly-once comes from an idempotent consumer sink, not the broker; enforce schemas and validate data at every boundary.

The data layer has to be shipped and operated safely, which is the next chapter. → Deployment & ops

Further reading

• Sampling — OpenTelemetry docs. Head-based and tail-based sampling and their trade-offs.
• Trace Context — W3C Recommendation, 2021. The headers to place in HTTP calls and message headers.
• Metric and label naming — Prometheus documentation. Why label cardinality matters and how to name metrics.
• Event Sourcing — Martin Fowler. State as a fold over an event log, and what it costs.
• Apache Kafka documentation — Apache Software Foundation. Partitioning, ordering, delivery semantics and consumer groups.
• Distributed Locks with Redis — Redis documentation. The Redlock algorithm and the debate over its guarantees.