“Monitor the health of the cluster” sounds like a cron job that pings every box and pages someone when one goes quiet. Then you put 50,000 nodes behind it and ask for sub-second detection, and every easy choice breaks. Poll 50K nodes from one monitor and you cannot finish a sweep in a second. Trust a single missed heartbeat and every GC pause or 200ms network blip pages you at 3am. Auto-restart on the first miss and a flaky switch takes down a rack, then your remediation logic restarts all of it in a storm and you have turned a blip into an outage. The whole problem is doing three hard things at once: detect a real failure in under a second across 50K nodes, be sure it is real and not a false alarm, and act on it without making things worse.

The core insight is that “is this node dead” is not a lookup, it is a statistical inference over a stream of heartbeats, and “should I remediate” is not a reflex, it is a guarded state machine that has to survive network partitions, correlated failures, and its own bugs. Let me build it properly.

Functional Requirements (FR)

In scope:

  • Liveness tracking. Know, within a second, whether each of 50K nodes is up, down, or suspect. This is the uptime signal.
  • Resource and service health. Collect per-node resource usage (CPU, memory, disk, network, file descriptors, load) and the status of the critical services/daemons running on each node (up, degraded, crash-looping). Health is more than “process responds to ping.”
  • Failure detection. Turn the raw signals into a health verdict per node and per service, with a confidence level, distinguishing a dead node from a slow one from a partitioned one.
  • Alerting. Fire alerts (to humans / paging) when a node or service crosses a health threshold, deduplicated and grouped so one rack failure is one alert, not 40.
  • Auto-remediation. For a defined set of failure signatures, trigger an automated action (restart service, drain and cordon node, reboot, reprovision) safely, with rate limits, blast-radius caps, and a circuit breaker.
  • Query and history. Answer “is node X healthy now,” “what was its CPU over the last 6h,” “how many nodes are unhealthy right now” for dashboards and capacity planning.

Explicitly out of scope (own the scope):

  • The scheduler/orchestrator itself. We report node health and can call a drain/reprovision API; we do not rebuild Kubernetes or the placement engine. We feed it, we do not replace it.
  • Application-level tracing and logs. Distributed tracing, log aggregation, and APM are a separate pipeline. We ingest metrics and health signals, not full request traces.
  • Long-term analytics / ML anomaly detection. We do threshold and statistical failure detection; a learned anomaly model is an extension I note, not the core.
  • Provisioning new capacity. We can trigger reprovisioning via an external API; the image build and bare-metal provisioning pipeline is upstream.

The decision that drives everything: liveness is latency-critical and cheap per sample, while resource metrics are voluminous but latency-tolerant. These are two different problems wearing one name. Treating them the same - one fat message on one path - forces you to either make the metric pipeline sub-second (impossibly expensive) or make failure detection slow (defeats the point). So the design splits them from the first hop.

Non-Functional Requirements (NFR)

  • Scale: 50K nodes. Each runs ~5-20 monitored services. ~200 resource metrics per node. The monitored fleet can grow to 100K without a redesign.
  • Latency (the headline): sub-second liveness detection. A hard-down node must be declared suspect within ~1s and down within ~2-3s (after confirmation). Resource-metric freshness can be 10s. Dashboard queries p99 under 500ms.
  • Availability: the monitoring plane must be more available than what it monitors - target 99.99%. A monitor that is itself down is a monitor that misses the outage it exists to catch. No single monitor failure may blind any node.
  • Consistency: health state is eventually consistent but must converge fast (sub-second). We favor availability over strict consistency for the health verdict - a briefly stale “up” is tolerable, a monitoring plane that stalls waiting for a quorum write on every heartbeat is not. Remediation decisions, by contrast, need a consistent, serialized view so two controllers do not both reboot the same node.
  • Durability: metric history is durable (time-series store, tiered retention). The live health state is a rebuildable in-memory view; losing it costs a few seconds of re-detection, not data.
  • Correctness: no false “down” that triggers destructive remediation on a healthy node, and no missed real failure. These two error types trade off directly, and the whole detection design is about tuning that trade-off, not pretending it away.

Back-of-the-Envelope Estimation (BoE)

Real numbers, with arithmetic, because they pick the architecture.

Liveness (the fast, cheap path):

50,000 nodes, heartbeat every 250ms for sub-second detection
  -> 50,000 / 0.25s = 200,000 heartbeats/sec
Per heartbeat on the wire: node_id(8) + seq(8) + ts(8)
  + summary_flags(8) = ~32 bytes, ~100 bytes with framing/headers
200,000 * 100 bytes = ~20 MB/sec = ~160 Mbps for liveness alone

200K tiny messages/sec is trivial in bytes but real in packet rate, and it must never queue behind anything slow. This is why liveness gets its own lightweight path.

Resource metrics (the heavy, slow path):

50,000 nodes * ~200 metrics * 1 sample / 10s
  = 50,000 * 200 / 10 = 1,000,000 datapoints/sec
Raw per datapoint: series_id(8) + value(8) + ts(8) = 24 bytes
1M * 24 bytes = 24 MB/sec raw ingest
Compressed (Gorilla / delta-of-delta, ~1.5-2 bytes/point):
  1M * 2 bytes = ~2 MB/sec sustained to storage

Storage/year for metrics:

2 MB/sec * 86,400 s/day = ~173 GB/day compressed
* 365 = ~63 TB/year at full resolution
With tiered downsampling (raw 7d, 1-min 30d, 5-min 1y):
  hot tier ~1.2 TB, warm ~5 TB, the bulk aged/rolled up

Full-resolution retention is the cost driver; downsampling and TTL keep it bounded. The point: metrics are a genuine big-data storage problem, liveness is not.

Detection fan-in:

200K heartbeats/sec spread over M monitor shards.
At M = 50 monitor shards -> 4,000 heartbeats/sec/shard,
  each tracking 1,000 nodes -> a per-node timer check is cheap.
A single monitor tracking all 50K at 250ms cadence would need
  200K timer evaluations/sec on one box - workable but a SPOF and
  a hot spot; sharding is about failure isolation more than CPU.

Alerts and remediations (rare but bursty):

Steady state: a handful of node failures/day in a healthy 50K fleet.
Correlated event (rack/switch/AZ loss): 40-2,000 nodes go suspect
  within the same second. The system must absorb that burst,
  group it into ONE incident, and NOT fire 2,000 remediations.

The headline: liveness is 200K tiny messages/sec that must be fast and isolated, metrics are 1M datapoints/sec that must be cheaply stored, and remediation is rare but must stay sane during correlated bursts. Three different regimes, three different designs.

High-Level Design (HLD)

Every node runs a lightweight agent. The agent does two independent things: it emits a fast, tiny heartbeat on a dedicated liveness path, and it samples ~200 resource/service metrics and ships them on a separate, batched telemetry path. Heartbeats fan into sharded failure detectors that maintain per-node liveness with a statistical (phi-accrual) model and reach a confirmed verdict via a small quorum of observers. Metrics fan into an ingest tier and land in a time-series database. A health-state service fuses liveness plus metric-derived health into a single per-node/per-service verdict, which drives alerting and the remediation controller. The controller is a guarded state machine that calls out to the orchestrator to restart, drain, or reprovision - rate-limited and blast-radius-capped.

  ┌──────────────────────────────────────────────────────────────┐
  │  Node (x50,000): AGENT                                         │
  │   ├─ heartbeat sender  (every 250ms, tiny)                     │
  │   └─ metric sampler    (every 10s, ~200 metrics, batched)      │
  └───────┬───────────────────────────────────┬──────────────────┘
          │ heartbeats (fast path)             │ metrics (bulk path)
          ▼                                    ▼
  ┌───────────────────────┐          ┌──────────────────────────┐
  │  Failure Detectors    │          │  Metric Ingest Tier      │
  │  (sharded by node_id) │          │  (stateless, batched,    │
  │  phi-accrual per node │          │   Kafka -> writers)      │
  │  timer wheels         │          └───────────┬──────────────┘
  └─────┬─────────────────┘                      │
        │ suspect/up per node                    ▼
        │ (to >=2 observers)            ┌──────────────────────┐
        ▼                               │  Time-Series DB      │
  ┌───────────────────────┐            │  (sharded, tiered,   │
  │  Quorum / Confirm      │◀───────────│   Gorilla-compressed)│
  │  (>=2 detectors agree, │            └───────────┬──────────┘
  │   monitor-lease check) │                        │ metric-based
  └─────┬──────────────────┘                        │ health rules
        │ confirmed verdict                         │ (CPU>95% 5m,
        ▼                                           │  svc down, etc)
  ┌──────────────────────────────────────────────────────────────┐
  │  Health-State Service   (per-node + per-service verdict,      │
  │   in-memory, replicated; source of "is X healthy")           │
  └───────┬───────────────────────────────────┬──────────────────┘
          │ threshold crossings                │ health snapshots
          ▼                                    ▼
  ┌───────────────────────┐          ┌──────────────────────────┐
  │  Alerting             │          │  Query API / Dashboards  │
  │  (dedup, group,       │          └──────────────────────────┘
  │   route to paging)    │
  └───────────────────────┘
          │ eligible signatures
          ▼
  ┌───────────────────────────────────────────────┐
  │  Remediation Controller (guarded state machine)│
  │   rate limit + blast-radius cap + circuit break│
  │   -> Orchestrator API (restart/drain/reboot/   │
  │      reprovision), idempotent, audited         │
  └───────────────────────────────────────────────┘

Liveness flow (fast, must be sub-second):

  1. The agent sends a heartbeat every 250ms to its assigned failure-detector shard (assignment via consistent hashing on node_id), carrying {node_id, seq, ts, summary_flags}. summary_flags is a one-byte roll-up (any critical service down? disk full? OOM imminent?) so a hard signal rides the fast path for free.
  2. Each detector shard runs a phi-accrual estimator per node: it models the inter-arrival distribution of that node’s heartbeats and outputs a suspicion level phi that rises smoothly as a heartbeat is overdue, rather than a hard “missed N in a row.” When phi crosses a threshold, the node is suspect.
  3. To avoid a single detector’s blind spot (its own network path is bad), each node’s heartbeat is also mirrored to a second (and third) detector. A node is declared down only when a quorum of its observers independently reach suspect - this separates “the node died” from “one detector cannot see it.”
  4. The confirmed verdict updates the Health-State Service, which fires alerting and hands eligible failures to the remediation controller.

Metric flow (bulk, latency-tolerant):

  1. The agent batches ~200 metrics every 10s and ships one compressed payload to the Metric Ingest Tier (behind a load balancer, producing into Kafka partitioned by node_id).
  2. Stateless writers consume, apply delta-of-delta + XOR (Gorilla) compression, and append to the Time-Series DB, sharded by series.
  3. A rules engine evaluates metric-based health conditions (CPU > 95% for 5m, disk > 90%, service restart-count spiking, OOMKills) against recent windows and feeds derived health into the Health-State Service, joining the liveness verdict.

The structural key: liveness and metrics never share a queue. A backed-up metric pipeline (a storage hiccup, a slow writer) cannot delay a heartbeat, because they are physically different paths with different SLAs. That separation is what makes sub-second detection survivable while still collecting a million datapoints a second.

Component Deep Dive

The hard parts, naive-first then evolved: (1) sub-second failure detection at 50K scale, (2) suppressing false positives and correlated-failure storms, (3) ingesting and storing 1M metric datapoints/sec, (4) safe auto-remediation.

1. Sub-second failure detection across 50K nodes

The job: within about a second, know that a node stopped responding, across 50,000 of them, without drowning in heartbeats and without a fixed timeout that is either too twitchy or too slow.

Naive approach: one monitor polls every node on a timer

The obvious first cut: a central monitor loops over the node list and pings each one; if a ping fails, mark it down.

every 1s:
  for node in all_nodes:            # 50,000 of them
      if not ping(node, timeout=200ms): mark_down(node)

Where it breaks:

  • You cannot finish a sweep in a second. 50K sequential pings at up to 200ms each is hours; even massively parallel, one box issuing 50K probes/sec plus tracking 50K timeouts is a hot spot, and the sweep interval is your detection latency. Sub-second is impossible this way.
  • Pull inverts the load. The monitor initiates every probe, so its outbound connection count and its own health gate the entire fleet’s visibility. It is a single point of failure and a single point of overload.
  • A fixed timeout is wrong at both ends. 200ms flags every GC pause and network hiccup as “down” (false positives); 5s to be safe blows the latency budget (false negatives). One threshold cannot fit nodes with different, time-varying latency.
  • No health, just reachability. A node that answers ping while its disk is full and its service is crash-looping reads as “up.”

Central polling with a fixed timeout scales neither in latency nor in load, and conflates “reachable” with “healthy.”

Evolved approach: push heartbeats + phi-accrual + sharded detectors + timer wheels

Three changes, together.

Push, do not pull. Each agent sends a heartbeat every 250ms to its detector. Now the fleet’s load is distributed (each node emits its own beat), the monitor does no outbound probing, and the natural cadence (250ms) sets the detection floor, not a serial sweep. A missing beat is the signal.

Replace the fixed timeout with phi-accrual. Instead of “missed 3 beats = down,” each detector maintains a running estimate of the node’s heartbeat inter-arrival distribution (mean and variance over a sliding window) and computes a suspicion value phi:

phi(now) = -log10( P(no heartbeat since last one | learned distribution) )

phi rises smoothly the longer a beat is overdue, scaled by how regular that node normally is. A node that beats like clockwork trips a given phi fast; a node that is naturally jittery is given more slack automatically. You act on thresholds: phi >= 1 (about 90% confidence) -> suspect; phi >= 3 (about 99.9%) plus quorum -> down. This adapts per node and per time-of-day instead of guessing one universal timeout, which is exactly what turns “sub-second but not twitchy” from a contradiction into a tunable knob.

Shard the detectors, use timer wheels. Assign nodes to detector shards by consistent hashing on node_id; at 50 shards each tracks ~1,000 nodes at 4K beats/sec, trivial per box, and one shard failing blinds only its slice (and only briefly, because of mirroring in deep dive 2). Each shard tracks “when is each node next expected” in a hierarchical timer wheel, so checking 1,000 timers is O(expired), not an O(n) scan every tick. The wheel wakes only for nodes whose beat is actually overdue.

Heartbeat in -> detector shard (hash(node_id))
   update per-node arrival stats (mean, var, last_ts)
   reschedule node in timer wheel for expected-next + slack
Timer wheel tick -> for each overdue node:
   compute phi; if phi>=suspect_threshold: emit SUSPECT(node)

This gives sub-second, self-tuning detection whose cost scales with fleet size linearly across shards and whose latency is set by the heartbeat cadence, not a sweep.

Why 250ms and not 50ms. Cadence trades detection speed for message rate and false-positive rate. 250ms gives a ~500ms-1s detection with room for one or two lost packets before suspicion; 50ms would 5x the message rate (1M beats/sec) and make normal jitter look like loss. 250ms hits the sub-second target with margin. For a subset of ultra-critical nodes you can dial cadence down further at their own cost.

2. False positives and correlated-failure storms

The job: be sure a “down” is real before anyone gets paged or anything gets rebooted, and when 500 nodes fail at once (a switch, an AZ), produce one incident and not a remediation stampede.

Naive approach: one detector decides, one alert per node, remediate immediately

Whichever detector owns the node declares it down; that declaration pages someone and kicks remediation, per node.

Where it breaks:

  • The detector’s own network is a blind spot. If the link between the detector shard and a rack goes bad, the detector sees every node on that rack go silent and declares 40 healthy nodes dead. The failure is in the observer, not the observed, and a single observer cannot tell the difference.
  • Correlated failures become alert floods. A top-of-rack switch dies and 40 nodes go suspect in the same 300ms. Naively that is 40 pages and 40 reboot attempts for what is one physical fault - and rebooting nodes whose switch is down does nothing but add churn.
  • A network partition looks identical to mass death. From the monitoring side of a partition, the far side “disappeared.” Treating partition-as-death and remediating the far side is how you turn a transient partition into a real outage (you reboot nodes that were fine).
  • Flapping. A node oscillating up/suspect/up (a marginal NIC) fires a page on every transition.

A single observer with immediate per-node action cannot distinguish “node died,” “I cannot see the node,” and “a shared dependency died,” and it amplifies correlated faults.

Evolved approach: quorum of observers, monitor leases, failure-domain grouping, hysteresis

Quorum of independent observers. Each node’s heartbeat is mirrored to 3 detector shards placed in different failure domains (different racks/AZs from each other and, where possible, from the node). A node is down only when >= 2 of its 3 observers independently reach suspect. If only one observer suspects it, the fault is that observer’s path - we mark that observer degraded, not the node. This directly removes the single-observer blind spot: it takes independent agreement across domains to condemn a node.

Monitor liveness leases. Detectors themselves heartbeat to a coordination service (etcd/ZooKeeper) and hold a lease over their node slice. If a detector shard dies or is partitioned, its lease expires and its slice is reassigned - and critically, a verdict from a detector whose lease is stale is ignored. This is how “the monitor is down” stops masquerading as “the monitored nodes are down.”

Failure-domain correlation and grouping. Before any node-level page or remediation, the confirmed-verdict stream passes through a correlator that knows the topology (node -> rack -> switch -> AZ). When many nodes in one domain go suspect within a short window, it hypothesizes a shared-dependency failure and collapses them into a single incident against the domain:

if >= K nodes under the same switch go suspect within T ms:
    open ONE incident: "switch S / rack R suspected"
    suppress the N per-node alerts under it
    remediation target = the domain (or: do nothing, escalate to human)
    (never mass-reboot nodes whose shared dependency is the fault)

This turns 500 node-down events into one “AZ-2 unreachable” incident, and - crucially - it withholds destructive per-node remediation when the real fault is upstream, because rebooting a node whose switch is dead is worthless and rebooting it when the switch comes back is a synchronized storm.

Partition awareness. If a detector loses a large, contiguous slice all at once while the rest of the fleet still beats fine, that pattern is treated as a suspected partition, not mass death: raise a partition incident, keep the far side’s last-known state, and do not remediate across the partition. Only isolated, uncorrelated node-downs get automatic node-level action.

Hysteresis / flap damping. State transitions require the condition to hold: up -> suspect at phi>=1, but suspect -> down only after quorum holds for a debounce window, and down -> up only after N consecutive good beats. A node must earn its way back. This kills flapping-driven page storms from a marginal NIC.

        phi>=1 (quorum)         holds T + quorum
  UP ───────────────▶ SUSPECT ───────────────▶ DOWN
   ▲                     │                       │
   └──── N good beats ───┴──── N good beats ─────┘   (hysteresis)

The combination - multi-domain quorum, monitor leases, topology-aware grouping, and hysteresis - is what lets the system be twitchy enough for sub-second detection without being so twitchy that it pages on jitter or reboots a healthy rack because a switch hiccuped.

3. Ingesting and storing 1M metric datapoints/sec

The job: absorb a million datapoints a second of resource/service telemetry, store it cheaply with tiered retention, and answer range queries fast - all off the liveness critical path.

Naive approach: agents write each metric to a relational table

Every sample is a row: INSERT INTO metrics(node_id, metric, ts, value).

Where it breaks:

  • Write amplification kills the DB. 1M inserts/sec into a B-tree with indexes is orders of magnitude past what a relational store sustains; index maintenance and per-row overhead dominate.
  • Storage explodes. 24 bytes of payload becomes 100+ bytes per row with row headers and indexes; 63 TB/year of data becomes hundreds of TB. Timestamps and near-constant values are stored raw and uncompressed.
  • Range queries scan. “CPU for node X over 6h” without a time-series-aware layout is a filtered scan over a giant table.
  • One sample per row ignores the shape of the data. Metrics are regular, append-only, mostly-slowly-changing numeric series - the exact case relational storage is worst at.

A general-purpose relational table is the wrong tool for high-rate, append-only, regular time series.

Evolved approach: batched ingest -> Kafka -> TSDB with columnar compression and tiering

Batch at the agent, buffer with Kafka. The agent sends one compressed payload of ~200 metrics every 10s, not 200 messages. Ingest is a stateless tier behind a load balancer that validates and produces into Kafka partitioned by node_id, which absorbs bursts, decouples producers from storage, and lets writers scale independently. A storage stall backs up in Kafka, it does not drop data and it never touches liveness.

Store in a purpose-built TSDB with Gorilla compression. Writers consume Kafka and append to a time-series store organized by series_id = hash(node_id, metric_name, tags). Two compression tricks do the heavy lifting:

  • Delta-of-delta timestamps. Samples arrive at a near-fixed 10s cadence, so the delta between deltas is almost always 0, stored in a bit or two instead of 8 bytes.
  • XOR float compression. Consecutive values (CPU 41.2 -> 41.3) differ in few bits; XOR them and store only the changed bits.

Together these push ~24 raw bytes to ~1.5-2 bytes/point, which is what makes 1M points/sec a ~2 MB/sec storage problem instead of 24 MB/sec.

Agent(10s batch) -> Ingest(LB, stateless) -> Kafka(part by node_id)
   -> Writer -> in-memory head block (recent 2h, per series)
   -> flush to immutable compressed chunks on disk/object store
Query: head block (hot) + chunks (warm/cold), merged by time range

Tiered retention and downsampling. A background compactor rolls raw data into coarser resolutions and expires it:

Tier Resolution Retention Store
Hot raw (10s) 7 days local SSD / head blocks
Warm 1-min rollup 30 days compressed chunks
Cold 5-min rollup 1 year object storage (S3)

Dashboards over a month read the 1-min tier (60x less data); a live debug reads raw. This bounds cost (the 63 TB/year raw figure collapses to a few TB once aged) while keeping recent data at full fidelity.

Metric-derived health rules run on the head block. The rules engine (CPU > 95% for 5m, restart-count spike, disk > 90%, OOMKills) evaluates over the in-memory recent window per series, so health signals are fresh (10s) without querying cold storage. Those derived signals join the liveness verdict in the Health-State Service. The service status per daemon rides here too: the agent reports each critical service’s state (running/degraded/crash-looping with restart count), which is both a metric and, when critical, a bit on the summary_flags fast path.

Purpose-built time-series storage with columnar compression and tiering turns a million-datapoint firehose into a cheap, queryable, off-critical-path store.

4. Safe auto-remediation

The job: when a failure matches a known signature, take the corrective action automatically, but never let the remediation logic cause a bigger failure than the one it is fixing.

Naive approach: on “down,” immediately reboot / restart

The detector says node down; the controller calls reboot(node) (or restart(service)) right then.

Where it breaks:

  • It acts on false positives. If the “down” was a partition or an observer blind spot (deep dive 2), you just rebooted a healthy node, causing the outage you were monitoring for.
  • Remediation storms. A bad deploy makes 2,000 nodes crash-loop; naive logic reboots all 2,000 at once, saturating the reprovisioning pipeline and the orchestrator, and taking down capacity faster than it recovers.
  • Not idempotent, no memory. If the controller crashes mid-remediation and restarts, it may issue a second reboot on a node already rebooting; two controllers may both grab the same node.
  • Fighting a persistent fault. A node with dead hardware fails, gets rebooted, fails, gets rebooted - an infinite loop of useless actions and churn.
  • No audit. When it does the wrong thing at 3am, there is no record of what it did or why.

Reflexive, stateless, unbounded remediation amplifies faults and cannot be trusted to run unattended.

Evolved approach: guarded state machine, rate limit, blast-radius cap, circuit breaker, idempotent + audited

Only confirmed signatures, only after grouping. The controller consumes confirmed verdicts (post-quorum, post-correlation from deep dive 2), never raw suspicions. If the correlator opened a shared-dependency incident, node-level remediation is suppressed and it escalates to a human. Only isolated, well-understood signatures (single node OOM, single service crash-loop, single node unreachable-with-healthy-neighbors) are eligible for automatic action.

A per-node remediation state machine with idempotency. Each remediation is a durable, serialized workflow keyed by node_id, holding a lock so only one controller acts on a node at a time:

HEALTHY -> DIAGNOSING -> ACTING(restart) -> VERIFYING
   -> if recovered: HEALTHY
   -> if not, escalate: ACTING(drain+reboot) -> VERIFYING
   -> if still not: ACTING(cordon+reprovision) -> VERIFYING
   -> if repeated failures: QUARANTINE (stop acting, page human)

It escalates gently (least-destructive action first: restart service, then drain and reboot, then reprovision) and each action is idempotent (tagged with an action-id so a retried or duplicated call is a no-op). State is persisted so a controller crash resumes, not restarts.

Rate limit + blast-radius cap. Global and per-domain governors bound how much can be in-flight:

- at most X remediations/min globally (token bucket)
- at most Y% of any single failure domain (rack/AZ) acted on at once
- never drain a node if it would push a service below min healthy replicas

So even if 2,000 nodes are eligible, actions drip out under a cap, and a fault confined to one rack can never take down more than Y% of that rack at a time. This is the single most important safety property: the remediation rate is bounded independent of the failure rate.

Circuit breaker. If remediations are failing to restore health above some rate (say >30% of recent actions did not recover the node), the controller trips: it stops all automatic action and pages a human. A rising failure-of-remediation rate means the diagnosis is wrong or the fault is systemic (bad deploy, upstream dependency), and the correct response is to stop, not to keep swinging.

Quarantine for repeat offenders. A node that fails remediation N times enters quarantine: no further automatic action, flagged for hardware inspection. This breaks the reboot-fail-reboot loop.

Everything is audited. Every decision writes an immutable record: {node_id, signature, action, action_id, decided_by, verdict_snapshot, outcome, ts}. This is what lets you trust the thing at 3am - you can always answer “what did it do and why,” and you can replay a bad night.

The combination - confirmed-only input, an idempotent escalating state machine, hard rate and blast-radius caps, a circuit breaker, quarantine, and a full audit trail - is what makes automatic remediation safe to leave on. The design assumption is that the remediator will sometimes be wrong, so every mechanism bounds the damage of a wrong action rather than assuming actions are always right.

API Design & Data Schema

API

Agent -> monitoring plane (high volume, internal):

POST /v1/heartbeat            (fast path, tiny, every 250ms)
  { node_id, seq, ts_ms, summary_flags }   // flags: svc_down|disk_full|oom
  -> 204 No Content            // ack optional; fire-and-forget is fine

POST /v1/metrics              (bulk path, every 10s, compressed batch)
  { node_id, ts_ms,
    samples: [ { metric, tags{...}, value }, ... ~200 ],
    services: [ { name, state, restart_count }, ... ] }
  -> 202 Accepted

Query / control plane (dashboards, operators, orchestrator):

GET /v1/nodes/{node_id}/health
  -> { node_id, liveness: "up|suspect|down",
       phi, confirmed_by: 2, services: {...},
       resources: { cpu, mem, disk, ... }, updated_at }

GET /v1/nodes?status=unhealthy&domain=az-2        -> paginated list
GET /v1/nodes/{node_id}/metrics?metric=cpu&from=..&to=..&step=1m
  -> time series (served from the appropriate tier)

GET /v1/incidents?open=true                       -> grouped incidents
POST /v1/nodes/{node_id}/remediate                -> manual trigger (audited)
  { action: "restart|drain|reboot|reprovision", reason }
POST /v1/remediation/pause                         -> global kill switch
GET  /v1/remediation/audit?node_id=..&from=..      -> audit trail

Errors:
  409  remediation already in-flight for this node
  423  remediation globally paused (circuit breaker / kill switch)
  429  rate limit / blast-radius cap hit

Data stores

1. Live health state - in-memory, replicated (the “is X healthy” source). Per-node and per-service current verdict. Rebuildable from the heartbeat stream in seconds, so it is a fast KV/actor store (in-process sharded map + Raft-replicated, or Redis) rather than a durable DB. Sharded by node_id.

HealthState (key: node_id)
  node_id        STRING   PK / shard key
  liveness       ENUM     up | suspect | down
  phi            FLOAT
  observers      MAP      detector_id -> {suspect: bool, lease_ok: bool}
  services       MAP      name -> {state, restart_count, last_change}
  resources      STRUCT   {cpu, mem, disk, net, fd} last snapshot
  domain         STRUCT   {rack, switch, az}         // for correlation
  updated_at     TS

2. Metrics - time-series DB (NOT relational). High-rate, append-only, regular numeric series with columnar Gorilla compression and tiered retention. Relational storage is the wrong shape (write amplification, no delta/XOR compression, scan-based range reads); a TSDB is purpose-built for exactly this access pattern.

Series (key: series_id = hash(node_id, metric_name, sorted_tags))
  series_id      BYTES    -- shard key
  labels         MAP      node_id, metric_name, service, dc, ...
Chunks (per series, immutable, time-ordered)
  series_id, start_ts, end_ts,
  data BLOB     -- delta-of-delta timestamps + XOR-compressed values
  resolution    ENUM  raw | 1m | 5m
  Index: (series_id, start_ts)   -- range scan by time
Inverted index: label -> [series_id]   -- for "all cpu series in az-2"

3. Remediation / audit - durable relational (SQL). Low volume, needs transactions, serialization (one controller per node), and an immutable audit trail - relational fits. Shard by node_id.

Table: remediations
  remediation_id  UUID     PK
  node_id         STRING   -- shard key; UNIQUE(node_id) WHERE state active
  signature       STRING   -- oom | svc_crashloop | unreachable ...
  state           ENUM     diagnosing|acting|verifying|quarantine|done
  action          STRING   restart|drain|reboot|reprovision
  action_id       UUID     -- idempotency key for the orchestrator call
  attempt         INT
  decided_by      STRING   controller id
  verdict_snapshot JSONB   -- what the health looked like at decision
  outcome         ENUM     recovered|failed|escalated
  created_at, updated_at  TS
  INDEX (state), INDEX (node_id, created_at)   -- active work + audit

4. Topology store - KV/graph, cached everywhere. node_id -> {rack, switch, az, service_group} plus the domain hierarchy, used by the correlator and blast-radius caps. Small, slow-changing, replicated read-only into every detector and the controller.

5. Detector coordination - etcd/ZooKeeper. Detector leases, node-to-shard assignment, and the remediation kill switch. Strongly consistent, small.

Why this split: liveness state is hot, ephemeral, and rebuildable, so it lives in memory; metrics are a big-data time-series problem, so they live in a compressed TSDB; remediation needs transactions and an audit trail, so it lives in SQL; topology and leases are small consistent metadata, so they live in a KV/coordination store. No single store is asked to be fast, huge, transactional, and durable at once - each access pattern gets the store it deserves, and the latency-critical liveness path never touches durable storage.

Bottlenecks & Scaling

Where it breaks first, in order, and the fix for each.

1. Detection latency vs message rate (the defining tension). Sub-second detection wants a fast heartbeat cadence; a fast cadence multiplies message rate and false positives. Fix: 250ms push heartbeats on a dedicated path, phi-accrual adaptive suspicion instead of fixed timeouts, sharded detectors (hash by node_id) with timer wheels so per-node checking is O(expired). Detection latency is set by cadence, cost scales linearly across shards.

2. Single-observer blind spots and correlated storms. One detector’s bad link condemns healthy nodes; a switch failure floods 500 alerts and 500 reboots. Fix: 3-observer quorum across failure domains (>= 2 must agree), monitor leases so a partitioned detector’s verdict is ignored, topology-aware correlation that collapses a domain’s failures into one incident and suppresses per-node remediation when a shared dependency is the fault, and hysteresis to damp flapping.

3. Metric ingest and storage. 1M datapoints/sec destroys a relational store. Fix: batch at the agent, buffer in Kafka (partition by node_id), store in a TSDB with delta-of-delta + XOR compression (~2 bytes/point), and tier/downsample (raw 7d, 1m 30d, 5m 1y). The firehose becomes ~2 MB/sec, off the liveness path entirely.

4. Remediation storms. 2,000 eligible failures would trigger 2,000 simultaneous destructive actions. Fix: rate limit (token bucket, X/min global), blast-radius cap (<= Y% of any domain at once, never below min healthy replicas), circuit breaker (stop if remediation success rate drops), and quarantine for repeat offenders. Remediation rate is bounded independent of failure rate.

5. Hot detector shards and rebalancing. A shard owning a busy slice or a failed shard’s reassignment. Fix: consistent hashing spreads nodes evenly; a failed shard’s lease expires and its slice reassigns to standbys, which rebuild per-node arrival stats from the next few beats (state is a few seconds of heartbeats, not durable data). Mirroring to 3 observers means a shard loss never blinds a node.

6. TSDB query load for dashboards. “Unhealthy nodes in AZ-2” and month-long charts. Fix: serve from the coarsest sufficient tier (1m/5m rollups), an inverted label index for series selection, and cache the current-health rollups (unhealthy counts per domain) in the Health-State Service so the common dashboard query never hits the TSDB at all.

7. The monitoring plane as a SPOF. A monitor that is down misses the outage it exists to catch. Fix: every tier is replicated across failure domains - detectors mirror 3x, Health-State is Raft-replicated, Kafka/TSDB/SQL replicate, and detector leases live in a consistent coordination store. The plane is deliberately deployed to be more available (and in different domains) than the fleet it watches.

8. Agent overhead and clock skew. 50K agents must be cheap and their timestamps trustworthy. Fix: the agent is a tiny static binary with strict CPU/mem caps; heartbeats carry a monotonic seq so out-of-order/duplicate beats are handled by sequence, not wall clock, and phi-accrual works on arrival times at the detector, not on the node’s clock - so node clock skew cannot forge liveness.

Shard keys, stated plainly: Heartbeats and Health-State -> node_id (co-locates a node’s liveness on its detector shard, mirrored to 3 domains). Metrics -> Kafka partitioned by node_id, TSDB sharded by series_id (spreads a node’s many series, keeps a single series contiguous for range scans). Remediation -> node_id (one serialized workflow per node, UNIQUE active row prevents double action). Never shard detectors by time or randomly - you would scatter a node’s observers and break the co-located, quorum-based verdict that keeps false positives out.

Wrap-Up

The trade-offs that define this design:

  • Split liveness from metrics at the first hop. Sub-second detection and a million-datapoint firehose are different problems; sharing a path forces one to ruin the other. A tiny dedicated heartbeat path plus a separate batched telemetry path is the structural decision everything else rests on.
  • Statistical suspicion, not fixed timeouts. Phi-accrual makes “fast but not twitchy” a tunable knob per node, and 250ms push heartbeats set detection latency by cadence instead of by a serial sweep.
  • Condemn a node only by cross-domain quorum. Three observers in different failure domains, monitor leases, and topology-aware correlation separate “node died,” “I cannot see it,” and “its switch died” - which is what stops a healthy rack from being rebooted over a network blip.
  • Store time series in a TSDB, not a table. Delta-of-delta plus XOR compression and tiered retention turn 1M points/sec into a few MB/sec and a few TB/year, cheaply queryable and off the critical path.
  • Bound the damage of remediation, do not assume it is right. An idempotent escalating state machine gated by rate limits, blast-radius caps, a circuit breaker, quarantine, and a full audit trail - so leaving auto-remediation on is safe because every wrong action is bounded, not because actions are always correct.

One-line summary: lightweight agents push 250ms heartbeats to sharded phi-accrual detectors that reach a cross-domain quorum verdict for sub-second, false-positive-resistant failure detection, while a separate batched pipeline stores a million resource datapoints/sec in a compressed tiered TSDB; a topology-aware correlator collapses correlated failures into single incidents, and a guarded, rate-limited, circuit-broken remediation controller acts on confirmed signatures - so 50K nodes stay observed, verdicts stay correct, and automation heals faults without ever amplifying them.