Observability and Monitoring Architecture Questions

Keep every region fully functional in isolation (write-local, read-local) and treat cross-region replication as an asynchronous durability mechanism, not a synchronous dependency for local dashboards. That's what lets a region stay observable during a full outage of any other region: nothing on the local read or write path ever blocks on a remote call.

Architecture

flowchart LR
  A[Region 1: Local Agents] --> B[Region 1: Hot Store]
  B -->|async replicate compressed blocks| C[Region 2: Durable Copy]
  B -->|async replicate compressed blocks| D[Region 3: Durable Copy]
  E[Region 2: Local Agents] --> F[Region 2: Hot Store]
  F --> C
  G[Query Router] --> B
  G --> F
  G --> H[Cross-Region Federation: nearest healthy]

• Each region runs a complete local observability stack (ingestion, hot storage, dashboards) so local telemetry never has to leave the region to be queried.
• Long-term/durable copies are replicated asynchronously to the other regions' object storage, batched and compressed, so a region that goes down entirely still has its recent history durable elsewhere.
• A query router directs dashboard reads to the local hot store first; if that region is down, it falls back to the nearest healthy region's replicated copy, and a federation layer merges results for genuinely global queries (e.g., "error rate across all regions").
• Clients (application agents) that can't reach their local ingest endpoint fail over to the nearest healthy region via DNS/client-side retry, so telemetry keeps flowing even during a full regional outage of the ingest path itself.

Why async replication of compressed data, not synchronous cross-region writes

Synchronous cross-region writes would mean every sample write waits on a round trip to at least one other region, multiplying write latency by inter-region network RTT (tens of milliseconds at best, over 100ms for distant region pairs) for every single sample, which is unacceptable at ingestion volumes discussed elsewhere in this domain (hundreds of thousands of samples/sec). Async replication of already-compressed chunks decouples durability from the write's critical path entirely.

Sizing the actual replication cost: 3 regions, each ingesting 300,000 samples/sec locally, with the S13/S14-style compression assumption of about 2 bytes/sample post-compression, replicating to the other 2 regions for durability:

regions = 3
per_region_ingest = 300_000
compressed_bytes_per_sample = 2
egress_cost_per_gb = 0.02   # illustrative unit cost, not a live vendor quote

per_region_bytes_sec = per_region_ingest * compressed_bytes_per_sample        # 600,000 B/s = 600 KB/s
per_region_replication_bytes_sec = per_region_bytes_sec * (regions - 1)       # to 2 peers: 1,200 KB/s
total_replication_bytes_sec = per_region_replication_bytes_sec * regions      # 3.6 MB/s aggregate
total_replication_gb_day = total_replication_bytes_sec * 86400 / 1e9          # 311.0 GB/day
monthly_cost = total_replication_gb_day * 30 * egress_cost_per_gb             # $186.62/month

At about 311 GB/day of aggregate cross-region traffic and an illustrative $186.62/month in egress cost, replicating already-compressed telemetry is cheap; the cost argument for async-and-compressed over synchronous-and-raw isn't marginal, it's roughly two orders of magnitude in bandwidth (compression alone gets you the 16-bytes-to-2-bytes reduction shown in the TSDB storage math elsewhere, before even counting that synchronous writes would need to happen per-sample rather than in batched chunks).

Deduplication and consistency

Because each region's replicated copy and the local hot-store copy can briefly diverge (async lag), the query-merge layer needs to deduplicate by (SeriesID, timestamp) when stitching a cross-region federated result, and prefer the ingest-region's copy as authoritative when both exist (rather than picking arbitrarily), to avoid the same data point appearing twice or a stale replica shadowing a fresher local write.

Cost trade-offs across replication strategies

Trade-offs and pitfalls

• Treating replication lag as zero is the most common mistake in the design write-up; alerting and incident dashboards should always default to local data (which has no replication lag) and only fall back to a remote replica when the local region is actually down, otherwise a transient replication delay looks like a data gap during a real incident.
• Full active-active replication sounds like the "safest" answer but multiplies both storage and cross-region bandwidth cost by the region count for a guarantee (any region can serve any tenant at full fidelity) the requirement doesn't actually ask for; the requirement is "stay observable during outage," which write-local/read-local with async replication already satisfies at a fraction of the cost.
• Data residency constraints (a region legally cannot replicate certain data outside its jurisdiction) break the "replicate everywhere" assumption; the design needs a per-region or per-tenant replication policy, not a single global rule, when residency requirements are in play.
• Deduplication logic that doesn't clearly prefer the ingest-region's copy as authoritative can silently double-count or shadow-out fresher data during federated queries, which is a subtle correctness bug that only shows up as slightly-wrong aggregate numbers, not an obvious failure.

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.

The architecture is a pipeline of five stages, each trading some fidelity for cost, plus a reconstruction path at query time for the cases where someone genuinely needs the detail back. The stages: label reduction, aggregation and rollups, sampling, storage tiering, and retention.

Pipeline

flowchart LR
  A[Raw High-Cardinality Metrics] --> B[Label Reduction]
  B --> C[Aggregation and Rollups]
  C --> D[Sampling]
  D --> E[Storage Tiering]
  E --> F[Retention Enforcement]
  F --> G[Query Layer]
  G --> H[On-demand Rehydration: top-K exemplars]

• Label reduction: the single highest-leverage step. Cardinality is combinatorial across independent label dimensions: if endpoint has 80 distinct values, status has 6, pod has 800, and region has 4, the bounded series count is their product.

Adding one more label with effectively unbounded cardinality (like a raw request_id, close to one unique value per request) multiplies this by request volume instead of by a fixed factor, turning a bounded 1.5M-series metric into an unbounded one. Label reduction means identifying and removing or bucketing exactly those unbounded dimensions before ingestion, keeping the combinatorially-bounded ones.

• Aggregation and rollups: pre-compute coarser-grained series (per-service instead of per-pod, for example) at ingestion time so most dashboard and alert queries never touch the raw high-cardinality series at all.
• Sampling: for label values that are useful individually but too numerous to keep in full (e.g., per-customer-ID series for a B2B product with thousands of customers), keep exact series for the top-K by volume or spend, and sample or bucket the long tail.
• Storage tiering: recent raw data on fast storage, older data downsampled and moved to cheaper storage, as in a standard hot/warm/cold retention policy.
• Retention enforcement: hard expiry so the pipeline's cost stays bounded over time regardless of how ingestion volume trends.

Reconstructing detail on demand

The trick that makes this acceptable to users is that "finer detail" doesn't mean "keep everything forever," it means "keep enough breadcrumbs to go get the detail from a cheaper source when someone actually asks." Two mechanisms:

1. Exemplars: attach a sampled trace ID or raw event reference to an aggregated metric bucket, so a spike in the aggregate can be drilled into by fetching the handful of exemplar traces/logs that were kept in full, even though the metric itself was aggregated.
2. On-demand reprocessing: if raw pre-aggregation data still exists in a cheap cold tier (e.g., unindexed compressed blobs), a rare deep-dive query can trigger an offline reprocessing job rather than requiring the hot path to keep everything queryable in real time.

Sizing against a cost ceiling

Given an illustrative monthly storage budget of $5,000 and an illustrative unit cost of $0.02/GB-month (a labeled assumption for this worked example, not a live vendor quote):

Using the 1,536,000 bounded series from the label-reduction step above, at a 5-minute downsampled resolution with 8 bytes/point (consistent with the multi-aggregate downsampling estimate used for retention-tier design):

series = 1_536_000
bytes_per_point = 8
interval_s = 300
points_per_day = 86400 / interval_s
bytes_per_day = series * points_per_day * bytes_per_point   # 3.539 GB/day
max_days = (250_000 * 1e9) / bytes_per_day

That affords roughly 70,600 days of 5-minute-resolution history under the budget, which is obviously far beyond any real retention need, so the budget is not actually the constraint at this series count and resolution. Repeating the same calculation for raw 15-second resolution at 2 bytes/sample instead gives 17.69 GB/day and about 14,128 affordable days, roughly 5x shorter than the downsampled case (the 20x fewer points at 5-minute resolution is partly offset by needing 4x more bytes per point to store min/max/sum/count instead of a single value, netting exactly 20/4 = 5x). The concrete lesson: at this series count, the budget comfortably covers years of retention either way, so cost pressure at 1.5M series is not what forces sampling or tiering, it forces label reduction to happen so the series count never gets to the point where the arithmetic above breaks down (e.g., adding the unbounded request_id label would blow past 250 TB in a matter of hours).

Trade-offs and pitfalls

• Treating sampling as the first line of defense (instead of label reduction) is the most common mistake: sampling a metric whose cardinality is unbounded because of a labeling error still leaves an unbounded number of series, each just sampled less; the series count itself, not just the sample rate, is what needs to be bounded first.
• Aggregation destroys the ability to answer "which specific instance caused this" without exemplars; a design that aggregates without keeping any drill-down path trades away debuggability that's expensive to get back later.
• A fixed cost ceiling naturally reframes the problem: it's not "how do we store everything cheaper," it's "what data can we afford to keep at what fidelity," and that framing should drive which stage of the pipeline (label reduction vs. sampling vs. tiering) absorbs the cost pressure, since as shown above they don't interact linearly.
• Query-time reconstruction only works if the cheap cold tier is actually queryable, even slowly; if cold data is written in a format nothing can read without a bespoke recovery process, "reconstruct on demand" is really "data is gone" with extra steps.

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.

Direct Answer

Partition by the entity that ordering has to be preserved for (tenant plus series key), lean on Kafka's per-partition ordering plus consumer-group parallelism (multiple consumers sharing the work by each owning a subset of partitions) across partitions for throughput, and make every downstream write idempotent using a deterministic key derived from the source offset, not wall-clock arrival time. That last point is what makes reprocessing safe: replaying the same offset range twice produces the same output twice, not duplicated output.

Structured Elaboration

Pipeline topology

flowchart LR
  RAW[["Raw Topic: 32 partitions"]] --> ENR["Enrich + Join"]
  ENR --> ROLL["Rollup Aggregator"]
  ROLL --> TSDB[("TSDB")]
  ROLL --> OBJ[("Object Storage")]
  REPLAY["Replay Job: run_id"] -.-> RAW
  ROLL -.->|"shadow output"| MANIFEST[("Manifest: promote on validate")]
  MANIFEST -.-> TSDB

Partitioning keys and ordering
Partition key: hash of tenant_id + series_key. All events for a given series land in the same partition, in publish order, which Kafka preserves per partition. Ordering guarantees stop at the partition boundary: there is no cross-partition ordering guarantee, and the design should not need one, since joins and rollups key on the same tenant_id + series_key and therefore only ever need to reason within a partition.

Idempotency and deduplication
Every event carries (source_partition, source_offset) as a natural, monotonic, unique key. Downstream sinks write as upserts keyed by (metric, labels, timestamp) rather than blind appends. Given deterministic aggregation logic (the same input set always produces the same output), an upsert keyed this way is safe to repeat: replaying the same input just overwrites the same output value, it does not add to it.

Enrichment and joins
Enrichment reads from a keyed local state store (updated from a compacted changelog topic of reference data, a Kafka topic that keeps only the latest value per key so it behaves like a durable running table) rather than an external synchronous call, so the join does not add per-event network latency and does not fail the pipeline if a reference-data service has a bad moment.

Rollups
Windowed rollups (grouping events into time buckets called windows; tumbling windows are fixed, back-to-back, non-overlapping buckets, sliding windows overlap and shift continuously) run in the stream layer with event-time watermarks (a marker saying no more late data is expected before a given timestamp, so a window can safely close), so a metric arriving slightly out of order still lands in the correct window instead of silently going into whatever window happens to be currently open.

Reprocessing historical data without duplicates
Reprocessing reads a bounded offset range from the immutable raw topic and writes its output to a separate, run-tagged path (a shadow output, not the live output). The run is validated against expectations (row counts, spot-check aggregates) and only then "promoted," an atomic manifest pointer swap that makes the new output the one queries read. The live output is never partially overwritten by a job that might fail halfway.

Worked Example

Assume a topic ingest rate of 500,000 events/sec after enrichment (slightly bigger messages than raw, average 300 bytes). Aggregate bandwidth:

Stream processing (enrichment plus join) is more CPU-heavy per byte than a raw pass-through, so budget a lower per-partition throughput target of 5 MB/s (a stated design assumption, not S0's raw-ingest figure of 10 MB/s):

Consumer lag and slow-consumer recovery. Per-partition steady rate: $500{,}000 / 32 = 15{,}625\ \text{events/s}$. Assume a consumer sustains 20,000 events/s/partition when caught up (a stated design assumption), giving headroom of $20{,}000 - 15{,}625 = 4{,}375\ \text{events/s}$.

If a consumer stalls for 10 minutes (600 s), the backlog that accumulates on its partition is:

Draining that backlog at the spare headroom rate takes:

That is over 3.5x the length of the original stall. A consumer with only 28% steady-state headroom recovers far slower than the outage that caused the backlog, which is the concrete reason to autoscale additional consumer instances on rising lag rather than relying on fixed per-partition headroom alone.

Trade-offs and Pitfalls

Choosing tenant_id + series_key as the partition key optimizes for join and rollup locality, but a single very high-volume tenant or series can create a hot partition that no amount of consumer parallelism fixes, since one partition is only ever read by one consumer in the group at a time. Detecting and handling that (a salted key for known outliers, or a dedicated topic for the outlier tenant) is a real operational concern, not a hypothetical.

Idempotent upserts only stay correct if the aggregation function is genuinely deterministic given the same inputs. A rollup that includes any non-deterministic element, current wall-clock time at write, or an ordering-dependent tie-break, silently breaks the safety of reprocessing: two runs over the same offset range could legitimately produce different output, and the shadow-and-promote validation step is exactly what catches that before it reaches the live path.

Fixed per-partition consumer headroom is cheap to reason about but expensive to actually hold idle most of the time. Autoscaling on lag is more efficient but adds complexity (scale-up latency, and new consumers need to rebalance partition assignment, which briefly pauses processing on the partitions being reassigned), a real cost to weigh against the alternative of just running fatter, mostly-idle consumers.