Observability and Monitoring Architecture Questions

Guardrails have to exist at all three layers because each one catches a different failure mode. Instrumentation-layer guardrails prevent bad label design from ever shipping; ingestion-layer guardrails catch what slips through in real time before it damages the shared backend; query-layer guardrails contain the blast radius of whatever cardinality already exists.

The three layers

flowchart LR
  A[Instrumentation: SDK label schema] --> B[Ingestion: cardinality meter]
  B --> C{Growth above threshold?}
  C -->|no| D[Accept]
  C -->|yes| E{Decision}
  E -->|reduce signal value| F[Aggregate away]
  E -->|protect budget, keep signal| G[Throttle]
  E -->|hard limit exceeded| H[Reject]
  D --> I[Query Layer]
  F --> I
  G --> I

• Instrumentation layer: require a metric schema/template before a new metric name can ship (declared label keys and, ideally, an expected cardinality bound per key); flag or reject at code-review/CI time any label key with an obviously unbounded domain (request IDs, raw user IDs, full URLs with path parameters unreplaced).
• Ingestion layer: maintain a live, low-memory estimate of distinct series per metric/tenant so growth can be detected within minutes, not after the backend already OOMs.
• Query layer: enforce query-time cost limits (max series a single query can touch, regex-predicate cost estimation) so that even cardinality that did get ingested can't be turned into a denial-of-service against shared query compute.

Detecting a metric about to blow up

The right tool is a HyperLogLog (HLL) sketch per metric name (or per metric x tenant), because it estimates distinct-count cardinality in fixed, small memory regardless of how many series actually exist. HLL's standard error and memory cost are both direct functions of its precision parameter $p$, where the sketch has $m = 2^p$ registers:

import math
for p in (10, 14):
    m = 2 ** p
    se = 1.04 / math.sqrt(m)
    mem = m * 6 / 8
    print(p, m, se, mem)

Result: p=10 gives 1,024 registers, 3.25% standard error, 0.77 KB memory; p=14 gives 16,384 registers, 0.81% standard error, 12.29 KB memory. Running one p=14 sketch per metric name across even 10,000 distinct metric names costs about 123 MB of memory total, cheap enough to keep live for every metric in the system, which is what makes real-time growth detection practical.

With a live baseline and a rolling-window estimate, growth-rate detection is a simple ratio check: if a metric's baseline series count is 5,000 and a 10-minute window shows 500,000, that's a 100x growth ratio against a chosen threshold of, say, 5x, which trips the guardrail well before the metric reaches an operationally dangerous size.

Deciding throttle vs. reject vs. aggregate

A useful mechanism for the "which label is the culprit" question: if a metric normally has a bounded combinatorial cardinality (e.g., endpoint × status × pod = 50 × 6 × 500 = 150,000 possible series, computed by multiplying each label's distinct-value count) and observed cardinality is far above that bound, the excess growth is coming from a label outside that expected set, and the ingestion layer can pinpoint it by comparing per-label-key distinct-value growth rates rather than only the aggregate metric-level count.

Trade-offs and pitfalls

• A pure hard-reject policy at the ingestion layer is the easiest to build and the worst for reliability: it turns a labeling mistake in one service into a total metrics outage for that service (including its SLO-relevant metrics), which is why aggregate-away and throttle need to exist as intermediate responses.
• HLL is probabilistic; at low precision (small $p$) the standard error is large enough that near-threshold decisions can flap. Pick $p$ based on how close to the threshold you need confident decisions, not just "the default."
• Guardrails without an audit trail (which decision was applied, to which metric, when) make it impossible to tell a team why their metric got throttled, which erodes trust in the guardrail system and encourages people to route around it (e.g., renaming a metric to dodge a quota).
• The instrumentation-layer guardrail is the cheapest one to enforce and the one most often skipped; catching an unbounded label at CI time costs nothing compared to catching it after it has already damaged the shared backend.

Direct answer

Use three correlation identifiers with distinct scopes: job_id for the logical run, task_instance_id for a specific task within it, and attempt_id for a specific retry of that task, and put them in the right telemetry signal for their cardinality. Metrics get labeled only by low-cardinality dimensions like task_name (a fixed, small set), while job_id/task_instance_id/attempt_id live on trace spans and structured logs, which are built to handle high-cardinality identifiers. A durable idempotency ledger keyed on job_id + task_name makes retries and partial failures attribute correctly instead of double-counting or getting lost.

Identifier design

• job_id: one per logical run of the pipeline (a specific execution of "yesterday's ETL," not the pipeline definition itself).
• task_instance_id: one per task within that run, stable across retries of the same task.
• attempt_id: increments on each retry of a task_instance_id.
• All three propagate together through logs, trace spans, and lineage events, so any of the three signals can be joined against the others for a given execution.

Why the label placement matters: cardinality

Cardinality here means the number of distinct label-value combinations a metric can produce; each distinct combination is a separate time series the metrics backend has to store and index. A metric labeled by something that's unique per job run (like task_instance_id) creates a brand-new time series on every single run, forever, which is exactly the failure mode metrics backends are not built for.

Metrics: aggregate counters/histograms keyed only by task_name (a fixed set, e.g. 50 distinct task names), plus a small number of other genuinely low-cardinality dimensions like environment or region. Never job_id, task_instance_id, or attempt_id as a metric label.

Traces: one root span per job_id, child spans per task attempt, with job_id, task_instance_id, attempt_id as span attributes (not labels that create new series). Trace backends are designed for exactly this kind of high-cardinality attribute.

Logs: structured JSON logs carrying the same identifiers, for full-detail debugging of a specific run.

Lineage: events keyed by asset_id, carrying job_id as a join key back to the run.

Idempotency and retry attribution

• A durable status store records terminal outcomes keyed by job_id + task_name (not attempt_id), written via compare-and-swap. A retried attempt that races with an already-succeeded prior attempt loses the CAS and cannot double-apply its effect.
• Job-level SLIs (success rate, throughput) are computed from terminal status only, not from every attempt. A task that fails twice and succeeds on the third attempt counts once, as a success, in the job-level SLI; the retry history is visible in traces/logs for debugging, but doesn't pollute the aggregate metric.
• Partial failures (some tasks in a job succeed, others don't) are recorded per-asset_id in the lineage stream, so a downstream consumer can tell exactly which outputs are trustworthy even if the overall job is marked failed.

Worked example

Assume $N = 10{,}000$ job runs/day, each with $T = 50$ tasks.

If a metric were (incorrectly) labeled by task_instance_id:

Against a typical metrics-backend active-series budget (illustrative figure of 1,000,000 active series for a shared cluster), this single pipeline alone would consume roughly half the entire budget every single day, and it never stops growing since task_instance_id values never repeat.

Labeled correctly by task_name instead:

The correct label placement isn't a minor optimization, it's the difference between a metrics backend that stays healthy indefinitely and one that degrades daily as more job runs accumulate.

flowchart LR
  JobRun[Job Run: job_id] --> Task[Task Attempt: task_name + attempt_id]
  Task --> IdemStore[Idempotency Store: CAS by job_id + task_name]
  Task --> Metrics[Metrics: labeled by task_name only]
  Task --> Trace[Trace Span: job_id + task_instance_id attrs]
  Task --> Lineage[Lineage Event: asset_id + job_id]
  IdemStore --> Finalizer[Job Finalizer: terminal SLI]
  Metrics --> Finalizer
  Lineage --> Catalog[Data Catalog]

Trade-offs and pitfalls

• The most common mistake in batch-job telemetry is putting a per-run identifier directly on a metric label because it's convenient for a one-off debugging query; it's cheap the first time and catastrophic at scale, exactly because cardinality growth is invisible until the backend is already struggling.
• Attempt-level metrics (per-attempt counters) are still useful for detecting retry storms or flaky tasks, but they should be aggregated at task_name granularity (a counter incremented per attempt, not a new series per attempt), keeping the diagnostic value without the cardinality cost.
• Idempotency keyed on job_id + task_name assumes tasks within a job have stable, unique names; a pipeline that dynamically generates task names (e.g. embedding a partition key in the name) reintroduces the same cardinality and idempotency problems this design was built to avoid, so dynamic task naming needs its own bounded-cardinality scheme.
• Computing SLIs from terminal status only is correct for "did the job ultimately succeed," but it can mask a job that's technically succeeding while burning through an unhealthy number of retries; that's why attempt-count needs its own aggregated signal (task-level, not job-level) even though it isn't the primary SLI.

Time-series compression works because consecutive samples in a series are usually similar to each other: timestamps arrive at near-regular intervals, and values tend to change by small amounts (or not at all) between samples. Block-chunking, delta-of-delta timestamp encoding, XOR-based float compression, and label dictionary encoding each exploit a different piece of that similarity.

How each technique works

Block-chunking: samples are grouped into fixed time-window or fixed-count blocks (e.g., 2-hour windows), each with its own compressed byte stream and a small metadata header (start/end timestamp, min/max value). This lets a query engine skip whole blocks that fall outside a requested range without decompressing them, and it bounds how much has to be re-encoded when new data arrives (you only ever append to the current open block).

Delta-of-delta timestamp encoding: instead of storing each timestamp, store the delta from the previous timestamp, then the delta of that delta. For a series scraped at a fixed interval, the delta between consecutive timestamps is constant, so the delta-of-delta is exactly zero almost every time, which collapses to a single control bit per sample.

XOR-based float compression (Gorilla): XOR the current value's bit pattern against the previous value's bit pattern. If the value hasn't changed much, most of the leading bits (sign, exponent, high mantissa) and trailing bits (low mantissa) of the XOR result are zero, and only a short "meaningful" middle span needs to be stored explicitly.

Dictionary encoding for labels: label keys and values repeat across enormous numbers of series (env=prod appears on millions of series), so each distinct string is stored once in a dictionary and every series references it by a small integer ID instead of repeating the string.

Worked example: counter vs. sparse gauge

The following is executed Python (struct-level IEEE-754 bit manipulation, no external data) so the numbers are exactly reproducible:

import struct

def bits(f):
    return struct.unpack('>Q', struct.pack('>d', f))[0]

def leading_zeros(x, width=64):
    return width if x == 0 else width - x.bit_length()

def trailing_zeros(x, width=64):
    if x == 0: return width
    c = 0
    while (x & 1) == 0:
        x >>= 1; c += 1
    return c

def xor_meaningful_bits(prev, cur):
    x = bits(prev) ^ bits(cur)
    if x == 0: return 0, 64, 0
    lz, tz = leading_zeros(x), trailing_zeros(x)
    return lz, tz, 64 - lz - tz

# dense monotonic counter: http_requests_total, +1 per 15s scrape
print(xor_meaningful_bits(184320.0, 184321.0))
# sparse gauge, small fluctuation: temperature idling
print(xor_meaningful_bits(21.4, 21.9))
# sparse gauge, large jump: temperature spikes
print(xor_meaningful_bits(21.9, 87.2))

Output:

(28, 35, 1)   # counter +1 step: 1 meaningful bit
(16, 47, 1)   # gauge, small change: also 1 meaningful bit
(9, 0, 55)    # gauge, large jump: 55 meaningful bits (almost the full double)

And for delta-of-delta timestamps on a realistically jittered 15-second scrape, with the full input pinned so the result is exactly reproducible (10 explicit timestamps, which by the $D_i$ formula above yields 8 delta-of-delta values, since each output needs 3 consecutive timestamps):

timestamps = [
    1720000000, 1720000015, 1720000030, 1720000045, 1720000060,
    1720000075, 1720000089, 1720000105, 1720000120, 1720000134,
]
deltas = [timestamps[i] - timestamps[i - 1] for i in range(1, len(timestamps))]
delta_of_delta = [deltas[i] - deltas[i - 1] for i in range(1, len(deltas))]
print("deltas:", deltas)
print("delta-of-delta:", delta_of_delta)

Output:

deltas:          [15, 15, 15, 15, 15, 14, 16, 15, 14]
delta-of-delta:  [0, 0, 0, 0, -1, 2, -1, -1]

4/8 are exactly 0 (single control bit each): the first four scrape gaps land exactly on the 15-second interval, then jitter shows up as a 14s gap, a 16s gap, and two more gaps that miss the interval by one second, each producing a nonzero delta-of-delta.

Which techniques help most for which shape

The counter case shows why Gorilla-style encoding is the default for metrics generally: even though the value itself is always changing (climbing), the bit pattern delta between consecutive floats stays small as long as the relative step size is small, which is true for almost all real counters. The gauge case shows the failure mode: a big jump changes the exponent bits, which wipes out both the leading-zero and trailing-zero runs simultaneously, so the encoding degrades toward storing the value nearly raw.

Trade-offs and pitfalls

• A common wrong turn is assuming XOR compression is "for gauges" and delta-encoding is "for counters" as a rule; the real determinant is how much the bit pattern changes step to step, which correlates with relative magnitude change, not with the metric type label.
• Partial decompression for range queries requires the block-level index (min/max timestamp, min/max value) to be cheap to read without decoding the compressed body; if you skip that index to save space, every range query degrades to a full block scan.
• Block size is a real tuning knob, not a footnote: too small and per-block metadata overhead dominates; too large and a single noisy stretch (like the sparse-gauge jump) drags down the compression ratio for the whole block and increases decompression latency for small range queries.
• Dictionary encoding for labels needs a compaction/garbage-collection story; a dictionary that only grows (never reclaims IDs for labels that stop being used) becomes its own unbounded-cardinality problem over long retention.

Direct answer

Fix this at the architecture level, not by throwing more compute at the same query plan. Build a layer of materialized views (precomputed, stored query results that refresh as new data lands, instead of being recomputed from scratch on every request) that pre-aggregate the group-bys and time windows dashboards actually use, add a query rewriter (a planner step that intercepts an incoming query and swaps it for a cheaper, equivalent one) that transparently substitutes a matching materialized view for the raw scan whenever the rewrite preserves aggregate semantics, and cache the result of the highest-frequency top-K queries with a TTL tied to the refresh cadence. Decide what to precompute with a cost-benefit rule: materialize a query shape when the daily rows it saves scanning outweighs the daily rows its incremental refresh costs to maintain, not by intuition about which dashboards "feel slow."

Structured elaboration

Materialized views (MVs): store pre-grouped, pre-aggregated rows keyed by the dimensions a panel actually displays (for example, top-20 services by error rate), refreshed incrementally as new raw data lands, not recomputed from scratch each time.

Pre-aggregation windows: keep multiple resolutions so a query can pick the coarsest one that still covers its time range: 1-minute rollups for the last hour, 5-minute for the last day, 1-hour for the last month. This bounds how many rows any query has to touch regardless of the underlying cardinality.

Partitioning (the piece the naive scan is missing): partition the raw and rollup tables by time first, then by a bounded set of high-selectivity dimensions (service, region). This lets both the raw fallback path and the MV refresh job skip whole partitions instead of scanning the full high-cardinality series space, which is what turns a query over "recent, high-cardinality metrics" from a full-table scan into a bounded one.

Query rewriting: a planner stage intercepts the incoming query, checks whether its group-by, filter, and time window are a subset of an existing MV's coverage, and rewrites the query to read the MV instead of raw data. Only rewrite when the operation is safe (sums, counts, min/max compose across MVs cleanly; distinct counts and unbounded percentiles generally do not without a sketch-based MV, which needs its own merge logic).

Caching top-K queries: cache the actual result set for the highest-frequency parameterized queries (dashboard panel + time range), keyed by a hash of the query shape and the current rollup epoch, invalidated on the next refresh rather than on a wall-clock timer.

flowchart LR
    Q[Incoming dashboard query] --> P{Query rewriter}
    P -- matches an MV --> MV[(Materialized view / rollup)]
    P -- no safe match --> RAW[(Partitioned raw store)]
    MV --> CACHE{Top-K result cache}
    CACHE -- hit --> R[Response]
    CACHE -- miss --> R
    RAW --> R
    ING[Streaming ingest] -- incremental refresh --> MV

Selection criterion, precisely: precompute a query shape when

where $f$ is how often that shape is queried per day. This is the freshness-versus-speed trade-off made concrete: the left side is what you save on reads, the right side is what you pay to keep the view fresh.

Worked example

A dashboard panel groups by service across $K = 100{,}000$ series matched by its filter, over a $W = 3600$s (1-hour) window, at a 15s scrape interval.

Without a materialized view, every execution scans one row per series per scrape tick in the window:

With a materialized view that stores 1-minute rollups already grouped down to the top 20 series the panel displays:

At $f = 500$ views/day for this panel, the daily rows saved by reading the MV instead of raw:

The MV refreshes incrementally every minute (1,440 refreshes/day), and each refresh only has to process the new minute's raw rows for the panel's series ($K \times 4$ samples/minute):

Savings exceed maintenance cost by roughly 20.8x here, so this panel clears the bar comfortably. Solving the selection inequality for the breakeven view frequency:

So the concrete criterion for this panel shape is: materialize it once it's viewed more than roughly 24 times a day; below that, the refresh overhead isn't earning its keep and the raw fallback path is cheaper.

Trade-offs & pitfalls

Combining all three, with the rewriter deciding per-query which to use, is what actually removes the timeout; any one alone leaves a gap.

Common wrong turns: materializing every group-by combination a dashboard could theoretically ask for (storage and refresh cost grow combinatorially, and most of those shapes are never queried, i.e. $f \approx 0$, which fails the selection criterion outright); rewriting queries onto an MV whose aggregation isn't actually composable for the requested operation (silently wrong distinct-counts or percentiles are worse than a slow correct answer); and caching on a fixed TTL instead of the rollup epoch, which either serves stale data past a refresh or invalidates a cache entry that's still perfectly valid.

Direct Answer

Pick the weakest delivery guarantee that still lets you reconstruct correctness at the aggregation layer, rather than paying the latency and complexity cost of strong guarantees everywhere. For most telemetry pipelines that means at-least-once delivery plus idempotent, deduplicated writes: true end-to-end exactly-once delivery across independent systems does not really exist, what you can build is a system that behaves as if it were exactly-once because duplicate deliveries have no effect on the result.

Structured Elaboration

At-most-once
A point may be silently lost, never duplicated. Cheapest and lowest-latency, but every dropped point is gone. Acceptable only for high-volume, low-value telemetry where a small, unbiased loss rate does not meaningfully change an aggregate.

At-least-once
No steady-state data loss, but retries can produce duplicates. This is the practical default for most producers (send, wait for ack, retry on timeout, occasionally re-deliver something that actually did arrive). Correctness at the aggregation layer requires deduplication: an idempotent write keyed by a unique identifier, so a duplicate delivery does not change the result.

Exactly-once
The goal is a metric aggregation that reflects each event's effect precisely once. In practice this is built, not given by any single component: it requires end-to-end idempotency (deterministic dedup keys), transactional or checkpointed writes (a write-ahead log tying the aggregate update and the consumer offset commit together atomically), and durable state. The cost is coordination overhead and added write latency.

Eventual consistency
Aggregates may be temporarily stale or incomplete, converging to correct only after all in-flight and late-arriving events settle. Handled with event-time windowing, watermarks that define how late an event can arrive before it's excluded, and a correction mechanism (recompute or emit a delta) for events that show up after their window closed.

Making a weaker guarantee correct in practice

• Deduplication IDs: a producer-assigned, monotonically increasing sequence number per stream. The consumer tracks a high-water mark per stream and discards anything at or below it.
• Idempotent writes: aggregation writes as upserts keyed by (metric, labels, timestamp window), so replaying the same input overwrites rather than accumulates.
• Write-ahead logs: the consumer writes its aggregate update and its new checkpoint offset in a single atomic step, then acks the upstream offset. On crash and restart, it replays only the un-checkpointed range, which is safe because that replay is itself idempotent.

Checkpoint and dedup flow

flowchart LR
  EVT["Event with sequence id"] --> DEDUP{"seq > high-water mark?"}
  DEDUP -->|"yes, new"| APPLY["Apply to Aggregate"]
  DEDUP -->|"no, duplicate"| DISCARD["Discard"]
  APPLY --> CKPT[("Atomic: Aggregate + Checkpoint")]
  CKPT --> ACK["Ack Offset Upstream"]

Worked Example

Assume 3 producers each sending 100 events into a 1-minute rollup window, for a correct total of 300 events. One producer's network connection times out on 2 of its events, so it retries both, and (because at-least-once delivery does not deduplicate at the transport layer) both the original and the retried copy arrive.

Without deduplication, the consumer counts:

A small number for one window, but a systematic bias that compounds across every window and every retried event at scale, not a one-time rounding error.

With deduplication (each producer assigns a monotonically increasing sequence number per stream, and the consumer tracks a high-water mark per producer, discarding anything at or below the mark it has already seen):

The dedup mechanism restores the exact correct count using only transport-layer at-least-once delivery, no exactly-once broker feature required.

Trade-offs and Pitfalls

Sequence-number-based dedup (a high-water mark per stream) only works correctly for pure retries, redelivery of the same sequence number. It breaks if the transport can reorder events beyond a small window, since a legitimately new event with a lower sequence number than one already seen would be incorrectly discarded as a duplicate. If your transport allows significant reordering, dedup by explicit unique event ID with a time-bounded lookup window instead of a strict high-water mark.

Exactly-once semantics inside a single system (say, a Kafka Streams application with transactional writes) do not extend automatically to an external sink. The moment you write to an external time-series database or object store, you are back to needing idempotent writes at that boundary, transactional guarantees inside the stream processor do not make the external write exactly-once by themselves.

Choosing at-least-once plus dedup as the default is right for the vast majority of telemetry, but not for every case: a billing-critical count (metered usage that directly drives a customer invoice) may warrant paying the extra coordination cost for stronger guarantees specifically at that boundary, while everything else stays on the cheaper default.