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.