diff --git a/shepherd_broker/redis.conf b/shepherd_broker/redis.conf index 83bd9a8..1f399f9 100644 --- a/shepherd_broker/redis.conf +++ b/shepherd_broker/redis.conf @@ -8,6 +8,10 @@ maxmemory 4gb # a TTL, so it can never evict live task streams (which carry no TTL) or, # in practice, in-flight queries -- those were written most recently and thus # have the most TTL remaining, making them the last candidates, not the first. +# Worker heartbeat keys are likewise written WITHOUT a TTL for this reason: a +# short-lived heartbeat would be the first thing evicted under pressure, which +# would make live workers look crashed. They stay persistent and the monitor +# reaps stale ones by last_seen age instead (see shepherd_utils/heartbeat.py). maxmemory-policy volatile-ttl loglevel notice # If we want to log to a file diff --git a/shepherd_utils/config.py b/shepherd_utils/config.py index b08cafc..b491573 100644 --- a/shepherd_utils/config.py +++ b/shepherd_utils/config.py @@ -240,6 +240,32 @@ class Settings(BaseSettings): # flood into just the ``broker_down``/``broker_recovered`` pair; a worker # that is *genuinely* still down once the window elapses alerts as normal. monitor_broker_recovery_grace_sec: int = 30 + # When the Redis broker hits its maxmemory cap and starts evicting keys, the + # short-TTL worker heartbeat keys are prime eviction targets (the deployment + # uses volatile-ttl, which sheds nearest-expiry keys first). Evicting a live + # worker's heartbeat makes it momentarily read as zero-alive even though it + # never disconnected -- key eviction doesn't close client connections. To + # stop that from firing a flood of bogus worker-down/crash alerts, worker-down + # alerts are suppressed while eviction is active and for this long after the + # last observed eviction, giving heartbeats (interval 5s, TTL 15s) time to be + # re-registered. A worker that is genuinely still down once the window + # elapses alerts as normal. + monitor_redis_eviction_grace_sec: int = 30 + # Redis broker memory-pressure alerting. Modeled on the broker/postgres + # health alerts (an automatic engine state machine, not a YAML rule): a + # single alert when usage crosses a threshold, at most one re-notify per + # cooldown while it stays elevated, and a recovery alert when it falls back + # below the warning threshold. The critical alert supersedes the warning + # one, so crossing the critical line never also sends the warning alert. + # All no-ops when the broker runs uncapped (maxmemory 0). + monitor_redis_memory_warning_pct: float = 85.0 + monitor_redis_memory_critical_pct: float = 95.0 + # An elevated level must hold this long before it alerts (and a drop must + # hold this long before it clears), so a brief spike/dip doesn't trip it. + monitor_redis_memory_grace_sec: float = 120.0 + # Re-notify backoff while memory stays elevated -- a full day by default, so + # a sustained condition reminds once a day rather than every few minutes. + monitor_redis_memory_cooldown_sec: float = 86400.0 # Postgres availability alerting. Same debounce idea as the broker: Postgres # must stay unreachable this long before the single ``postgres_down`` alert # fires, so a transient query timeout doesn't trip it. No recovery-grace diff --git a/shepherd_utils/heartbeat.py b/shepherd_utils/heartbeat.py index 82934e0..22d0cb0 100644 --- a/shepherd_utils/heartbeat.py +++ b/shepherd_utils/heartbeat.py @@ -7,9 +7,23 @@ Kubernetes -- this works identically under either, and handles autoscaling. Key format: ``worker:heartbeat:{stream}:{consumer}`` -TTL: refreshed every ``HEARTBEAT_INTERVAL_SEC``, expires after -``HEARTBEAT_TTL_SEC``. If a worker crashes or is killed, the key disappears -within the TTL window and the monitor will surface it as a worker-loss event. + +The key is written *without* a Redis TTL, deliberately. The broker runs a +``volatile-ttl`` maxmemory policy (see ``shepherd_broker/redis.conf``), which +evicts the keys nearest to expiry first when memory is tight. A short-TTL +heartbeat would be the very first thing shed under pressure, so a live worker +would vanish from the keyspace and the monitor would misread it as crashed -- +even though key eviction never actually disconnects the worker. Making the key +persistent exempts it from eviction entirely (the policy only touches keys that +carry a TTL, which is also why the no-TTL task streams are safe). + +Liveness is therefore judged from the payload's ``last_seen`` timestamp, not +from the key's existence: a heartbeat is "fresh" (the worker is alive) if it was +refreshed within ``HEARTBEAT_TTL_SEC``. Every reader agrees on this via +``is_heartbeat_fresh``. Because a persistent key can't expire itself away when a +worker crashes, the monitor reaps heartbeats left unrefreshed for longer than +``HEARTBEAT_REAP_SEC`` so dead consumers don't accumulate. A worker deletes its +own key on clean shutdown (see ``stop``/``_mark_shutdown_sync``). On SIGTERM/SIGINT we additionally write a *shutdown marker* ``worker:shutdown:{stream}:{consumer}`` synchronously before exiting. The @@ -34,7 +48,16 @@ HEARTBEAT_PREFIX = "worker:heartbeat" HEARTBEAT_SCAN_PATTERN = f"{HEARTBEAT_PREFIX}:*" HEARTBEAT_INTERVAL_SEC = 5 +# A worker whose heartbeat hasn't been refreshed within this window is considered +# not-alive (``is_heartbeat_fresh`` returns False). Sized comfortably above the +# ping interval so a single missed tick doesn't flap liveness. HEARTBEAT_TTL_SEC = 15 +# Heartbeat keys are persistent (see the module docstring), so a crashed worker's +# key can't expire on its own. The monitor deletes any heartbeat left unrefreshed +# for longer than this so dead consumers don't pile up in the keyspace. Well above +# HEARTBEAT_TTL_SEC so a briefly-paused worker (or a broker blip) isn't reaped and +# then forced to re-register; a genuinely dead worker is cleaned within ~2 min. +HEARTBEAT_REAP_SEC = 120 SHUTDOWN_PREFIX = "worker:shutdown" SHUTDOWN_SCAN_PATTERN = f"{SHUTDOWN_PREFIX}:*" @@ -51,6 +74,29 @@ def shutdown_key(stream: str, consumer: str) -> str: return f"{SHUTDOWN_PREFIX}:{stream}:{consumer}" +def is_heartbeat_fresh(raw: "str | bytes | None", now: "float | None" = None) -> bool: + """Whether a heartbeat payload was refreshed within ``HEARTBEAT_TTL_SEC``. + + Heartbeat keys are persistent (no Redis TTL) so they survive maxmemory + eviction; liveness is therefore decided from the payload's ``last_seen`` + age rather than from the key merely existing. This is the single definition + every "is this worker alive?" reader shares -- the monitor poller, the task + reclaimer, and the janitor -- so they can't drift apart. A missing/malformed + payload or absent ``last_seen`` reads as not-fresh (fail safe toward "dead"). + """ + if not raw: + return False + try: + last_seen = float(json.loads(raw).get("last_seen", 0) or 0) + except (TypeError, ValueError, json.JSONDecodeError): + return False + if last_seen <= 0: + return False + if now is None: + now = time.time() + return (now - last_seen) <= HEARTBEAT_TTL_SEC + + def _read_rss_bytes() -> "int | None": """Resident set size of this process in bytes, read from ``/proc/self``. @@ -66,6 +112,42 @@ def _read_rss_bytes() -> "int | None": return None +def _read_cgroup_memory_limit() -> "int | None": + """The container's memory limit in bytes, or ``None`` if uncapped/unknown. + + Read from the same ``/sys/fs/cgroup`` tree the CPU accounting uses, handling + both cgroup v2 (``memory.max``) and v1 (``memory/memory.limit_in_bytes``). + Both use a sentinel for "no limit": v2 writes the literal ``"max"``, v1 an + enormous page-counter maximum. We return ``None`` for either so the monitor + treats the worker as uncapped (and shows raw RSS without a percentage), + exactly like the Redis panel does when ``maxmemory`` is 0. Only meaningful + inside a container -- on a bare host the cgroup is the whole machine's. + """ + if not _in_container(): + return None + # cgroup v2: a single value or the literal "max". + try: + with open("/sys/fs/cgroup/memory.max", encoding="ascii") as f: + raw = f.read().strip() + if raw and raw != "max": + limit = int(raw) + # Guard against a v2 kernel also exposing a giant sentinel. + if 0 < limit < (1 << 62): + return limit + return None + except (OSError, ValueError): + pass + # cgroup v1: a byte count; "unlimited" is a near-INT64_MAX page-counter max. + try: + with open("/sys/fs/cgroup/memory/memory.limit_in_bytes", encoding="ascii") as f: + limit = int(f.read().strip()) + if 0 < limit < (1 << 62): + return limit + except (OSError, ValueError): + pass + return None + + def _read_proc_self_cpu_seconds() -> "float | None": """Cumulative CPU time (user + system) of *this process* in seconds. @@ -175,6 +257,10 @@ def __init__( # top-style "% of one core" reading is interpretable against the pod's # allocation. self._cpu_count = available_cpu_count() + # Container memory limit (cgroup cap), sampled once since it's static for + # the process's lifetime. Reported alongside rss_bytes so the monitor can + # show RSS as a percentage of the cap; None when uncapped. + self._mem_limit_bytes = _read_cgroup_memory_limit() # Previous CPU sample, so each ping can report utilization over the # interval since the last one rather than since process start. self._last_cpu_sec: float | None = None @@ -237,15 +323,19 @@ async def _ping(self) -> None: "task_limit": self.task_limit, "in_flight": self._in_flight(), "rss_bytes": _read_rss_bytes(), + "mem_limit_bytes": self._mem_limit_bytes, "cpu_pct": self._cpu_pct(), "cpu_count": self._cpu_count, } ) try: + # No TTL: a persistent key is exempt from the broker's volatile-ttl + # eviction, so memory pressure can't shed a live worker's heartbeat. + # Liveness rides on the payload's last_seen instead; the monitor + # reaps keys left unrefreshed past HEARTBEAT_REAP_SEC. await broker_client.set( heartbeat_key(self.stream, self.consumer), payload, - ex=HEARTBEAT_TTL_SEC, ) except Exception as e: self._logger.debug(f"Heartbeat ping failed: {e}") diff --git a/shepherd_utils/reclaim.py b/shepherd_utils/reclaim.py index cea04e3..54121c6 100644 --- a/shepherd_utils/reclaim.py +++ b/shepherd_utils/reclaim.py @@ -5,7 +5,10 @@ stream's PEL and uses ``XCLAIM`` to take over messages whose owner is no longer alive so the work gets retried instead of dropped. -Aliveness is determined by the heartbeat keys (TTL ~15s presence pings). +Aliveness is determined by the heartbeat keys, which are persistent (no Redis +TTL, so eviction can't shed them) and carry a ``last_seen`` timestamp refreshed +every ~5s; a consumer counts as alive only while that timestamp is fresh (see +``is_heartbeat_fresh``), so a stale leftover key doesn't shield a dead consumer. Two independent gates protect a busy-but-alive consumer from being robbed: 1. **Heartbeat filter**: a message owned by a consumer with a live heartbeat @@ -27,7 +30,7 @@ from .broker import broker_client from .config import settings -from .heartbeat import HEARTBEAT_PREFIX +from .heartbeat import HEARTBEAT_PREFIX, is_heartbeat_fresh # Per-stream idle floor in seconds. A reclaim only triggers once a pending # message has been idle longer than this, which gates against yanking work @@ -69,13 +72,27 @@ def min_idle_sec_for(stream: str, override: Optional[int] = None) -> int: async def _alive_consumers_on_stream(stream: str) -> set: - """Return the set of consumer names with a live heartbeat for ``stream``.""" + """Return the set of consumer names with a *fresh* heartbeat for ``stream``. + + Heartbeat keys are persistent, so key existence alone no longer proves + liveness -- a crashed consumer's key lingers until the monitor reaps it. + We read each payload and keep only consumers whose ``last_seen`` is fresh, + so a dead consumer's stale key can't wrongly shield its orphaned messages + from reclaim. This freshness check is intrinsic here (it reads last_seen + directly), so reclaim never depends on the monitor being up to reap. + """ pattern = f"{HEARTBEAT_PREFIX}:{stream}:*" - alive: set = set() + keys: List[str] = [] async for key in broker_client.scan_iter(match=pattern, count=200): + keys.append(key) + if not keys: + return set() + values = await broker_client.mget(keys) + alive: set = set() + for key, raw in zip(keys, values): # key format: worker:heartbeat:{stream}:{consumer} parts = key.split(":", 3) - if len(parts) == 4: + if len(parts) == 4 and is_heartbeat_fresh(raw): alive.add(parts[3]) return alive diff --git a/tests/unit/monitor/test_alerts.py b/tests/unit/monitor/test_alerts.py index ebe4b9e..a05b6b0 100644 --- a/tests/unit/monitor/test_alerts.py +++ b/tests/unit/monitor/test_alerts.py @@ -181,14 +181,83 @@ def _redis_snap(used, maxmem, evicted_delta=0): } -def test_redis_memory_rule_fires_over_threshold(): - rule = Rule({"name": "mem", "type": "redis_memory", "threshold": 85}) - # 9/10 GB = 90% -> breach. - assert rule.evaluate(_redis_snap(9_000_000_000, 10_000_000_000)) is not None - # 8/10 GB = 80% -> clear. - assert rule.evaluate(_redis_snap(8_000_000_000, 10_000_000_000)) is None - # Uncapped broker (maxmemory 0) -> no-op regardless of usage. - assert rule.evaluate(_redis_snap(9_000_000_000, 0)) is None +async def _settle(engine, snap, clock, start, *, grace=130): + """Drive handle_redis_memory across the debounce grace so a level commits. + + First tick arms the debounce; a second tick past the grace window commits + it (and fires any transition alert). Returns the post-grace timestamp. + """ + clock.return_value = start + await engine.handle_redis_memory(snap) + clock.return_value = start + grace + await engine.handle_redis_memory(snap) + return start + grace + + +async def test_redis_memory_warning_fires_after_grace(patched, clock): + engine = _engine() + warn = _redis_snap(88, 100) # 88% -> warning + # First observation only arms the debounce -- no alert yet. + clock.return_value = 1_000.0 + await engine.handle_redis_memory(warn) + patched["dispatch"].assert_not_called() + # Still inside the grace window: silent. + clock.return_value = 1_060.0 + await engine.handle_redis_memory(warn) + patched["dispatch"].assert_not_called() + # Past the grace window: the warning fires once. + clock.return_value = 1_130.0 + await engine.handle_redis_memory(warn) + patched["dispatch"].assert_called_once() + ev = patched["dispatch"].call_args.args[0] + assert ev["rule"] == "redis_memory_high" and ev["severity"] == "warning" + + +async def test_redis_memory_critical_does_not_also_send_warning(patched, clock): + engine = _engine() + # Straight to 97% -> critical only; the warning alert must not fire too. + await _settle(engine, _redis_snap(97, 100), clock, 2_000.0) + assert patched["dispatch"].call_count == 1 + ev = patched["dispatch"].call_args.args[0] + assert ev["rule"] == "redis_memory_critical" and ev["severity"] == "critical" + + +async def test_redis_memory_escalates_warning_to_critical(patched, clock): + engine = _engine() + await _settle(engine, _redis_snap(88, 100), clock, 3_000.0) # warning + assert patched["dispatch"].call_args.args[0]["rule"] == "redis_memory_high" + await _settle(engine, _redis_snap(97, 100), clock, 3_300.0) # escalate + assert patched["dispatch"].call_count == 2 + assert patched["dispatch"].call_args.args[0]["rule"] == "redis_memory_critical" + + +async def test_redis_memory_recovery_fires_when_back_below_warning(patched, clock): + engine = _engine() + await _settle(engine, _redis_snap(88, 100), clock, 4_000.0) # warning + await _settle(engine, _redis_snap(70, 100), clock, 4_300.0) # back to ok + assert patched["dispatch"].call_count == 2 + ev = patched["dispatch"].call_args.args[0] + assert ev["rule"] == "redis_memory_recovered" and ev["severity"] == "info" + + +async def test_redis_memory_renotifies_once_per_cooldown(patched, clock): + engine = _engine() + fired_at = await _settle(engine, _redis_snap(88, 100), clock, 5_000.0) + assert patched["dispatch"].call_count == 1 + # Sustained, but only an hour later: no re-notify (cooldown is a full day). + clock.return_value = fired_at + 3_600 + await engine.handle_redis_memory(_redis_snap(88, 100)) + assert patched["dispatch"].call_count == 1 + # A day later, still elevated: one re-notify. + clock.return_value = fired_at + 86_400 + 1 + await engine.handle_redis_memory(_redis_snap(88, 100)) + assert patched["dispatch"].call_count == 2 + + +async def test_redis_memory_uncapped_is_noop(patched, clock): + engine = _engine() + await _settle(engine, _redis_snap(999, 0), clock, 6_000.0) # maxmemory 0 + patched["dispatch"].assert_not_called() def test_redis_eviction_rule_fires_on_any_delta(): @@ -246,6 +315,69 @@ async def test_heartbeat_lost_suppressed_in_grace_then_fires_after(patched, cloc assert patched["dispatch"].call_args.args[0]["rule"] == "arax_down" +def _worker_snapshot_with_eviction(ts, evicted_delta, worker="arax", alive=0): + snap = _worker_snapshot(ts, worker=worker, alive=alive) + snap["redis"] = {"evicted_keys_delta": evicted_delta} + return snap + + +async def test_heartbeat_lost_suppressed_while_redis_evicting(patched, clock, mocker): + """A Redis eviction spell shouldn't surface as a worker-crash flood. + + Eviction can shed a live worker's short-TTL heartbeat key, making it read as + zero-alive though it never disconnected. The engine should suppress the + worker-down alert while eviction is active and for the grace window after. + """ + mocker.patch.object( + alerts.settings, "monitor_redis_eviction_grace_sec", 30, create=True + ) + rule = Rule( + {"name": "arax_down", "type": "heartbeat_lost", "worker": "arax", "duration": 0} + ) + engine = _engine([rule]) + + # Tick observes an eviction AND a zero-alive worker: suppressed, and no + # cooldown armed, so it can still fire later if genuinely down. + clock.return_value = 2_000.0 + await engine.evaluate(_worker_snapshot_with_eviction(2_000.0, evicted_delta=7)) + patched["dispatch"].assert_not_called() + + # Still inside the 30s grace after the last eviction (no new evictions): + # remains suppressed. + clock.return_value = 2_020.0 + await engine.evaluate(_worker_snapshot_with_eviction(2_020.0, evicted_delta=0)) + patched["dispatch"].assert_not_called() + + # Past the grace window with the worker still down: it now buffers a + # down-alert that flushes past the debounce window. + clock.return_value = 2_040.0 + await engine.evaluate(_worker_snapshot_with_eviction(2_040.0, evicted_delta=0)) + clock.return_value = 2_047.0 + await engine.evaluate(_worker_snapshot_with_eviction(2_047.0, evicted_delta=0)) + patched["dispatch"].assert_called_once() + assert patched["dispatch"].call_args.args[0]["rule"] == "arax_down" + + +async def test_scale_down_suppressed_while_redis_evicting(patched, clock, mocker): + mocker.patch.object( + alerts.settings, "monitor_redis_eviction_grace_sec", 30, create=True + ) + engine = _engine() + clock.return_value = 3_000.0 + snap = { + "ts": 3_000.0, + "events": [_scale_down("arax"), _scale_down("bte")], + "redis": {"evicted_keys_delta": 3}, + } + await engine.evaluate(snap) + clock.return_value = 3_010.0 + await engine.evaluate( + {"ts": 3_010.0, "events": [], "redis": {"evicted_keys_delta": 0}} + ) + patched["dispatch"].assert_not_called() + patched["dispatch_batch"].assert_not_called() + + async def test_multiple_downed_workers_coalesce_into_one_batch(patched): engine = _engine() events = [_scale_down("aragorn.lookup"), _scale_down("arax"), _scale_down("bte")] diff --git a/tests/unit/monitor/test_poller_rollup.py b/tests/unit/monitor/test_poller_rollup.py index d746078..e10598c 100644 --- a/tests/unit/monitor/test_poller_rollup.py +++ b/tests/unit/monitor/test_poller_rollup.py @@ -3,6 +3,10 @@ through into the snapshot the dashboard's per-replica modal renders from. """ +import json +import time + +from shepherd_utils.heartbeat import HEARTBEAT_REAP_SEC from workers.monitor import poller @@ -70,3 +74,36 @@ def test_rollup_stale_heartbeats_counted_separately(): assert bucket["stale"] == 1 # Both replicas remain visible in the table so the modal can flag the stale one. assert len(bucket["consumers"]) == 2 + + +async def test_collect_heartbeats_reaps_dead_persistent_keys(redis_mock, monkeypatch): + """Persistent heartbeat keys left unrefreshed past the reap window are deleted. + + A fresh worker survives and is returned; a long-dead one (last_seen far + older than HEARTBEAT_REAP_SEC) is both excluded from the result and removed + from Redis, so crashed consumers don't accumulate as phantom stale rows. + """ + broker = redis_mock["broker"] + monkeypatch.setattr(poller, "broker_client", broker) + now = time.time() + await broker.set( + "worker:heartbeat:merge_message:live", + json.dumps({"stream": "merge_message", "consumer": "live", "last_seen": now}), + ) + dead_key = "worker:heartbeat:merge_message:dead" + await broker.set( + dead_key, + json.dumps( + { + "stream": "merge_message", + "consumer": "dead", + "last_seen": now - HEARTBEAT_REAP_SEC - 10, + } + ), + ) + + workers = await poller._collect_heartbeats() + + assert [w["consumer"] for w in workers] == ["live"] + # The dead worker's key was reaped from the keyspace. + assert await broker.get(dead_key) is None diff --git a/tests/unit/test_reclaim.py b/tests/unit/test_reclaim.py index 6da5f24..b970fb3 100644 --- a/tests/unit/test_reclaim.py +++ b/tests/unit/test_reclaim.py @@ -15,7 +15,9 @@ """ import asyncio +import json import logging +import time import pytest @@ -24,6 +26,13 @@ logger = logging.getLogger(__name__) +def _fresh_heartbeat(last_seen: "float | None" = None) -> str: + """A heartbeat payload that reads as alive (recent ``last_seen``).""" + return json.dumps( + {"last_seen": last_seen if last_seen is not None else time.time()} + ) + + @pytest.fixture() def tiny_idle_floor(monkeypatch): """Force a ~1ms idle floor so the fakeredis IDLE filter behaves.""" @@ -70,8 +79,8 @@ async def test_reclaim_orphaned_skips_live_owner( stream, group = "aragorn.score", "consumer" await _seed_pending(broker, stream, group, "busyworker") - # busyworker has a live heartbeat -> protected from reclaim. - await broker.set(f"worker:heartbeat:{stream}:busyworker", "{}") + # busyworker has a fresh heartbeat -> protected from reclaim. + await broker.set(f"worker:heartbeat:{stream}:busyworker", _fresh_heartbeat()) reclaimed = await reclaim.reclaim_orphaned(stream, group, "liveworker", logger) @@ -80,6 +89,31 @@ async def test_reclaim_orphaned_skips_live_owner( assert [c["name"] for c in summary["consumers"]] == ["busyworker"] +@pytest.mark.asyncio +async def test_reclaim_orphaned_claims_from_stale_heartbeat_owner( + redis_mock, tiny_idle_floor, monkeypatch +): + """A lingering-but-stale heartbeat key does NOT shield a dead consumer. + + Heartbeat keys are persistent, so a crashed consumer's key can outlive it. + Reclaim must judge liveness by last_seen freshness, not key existence, or + orphaned work would never be recovered while the stale key sits there. + """ + broker = redis_mock["broker"] + monkeypatch.setattr(reclaim, "broker_client", broker) + + stream, group = "aragorn.score", "consumer" + msg_id = await _seed_pending(broker, stream, group, "deadworker") + # Key still present, but last_seen is far older than HEARTBEAT_TTL_SEC. + await broker.set( + f"worker:heartbeat:{stream}:deadworker", _fresh_heartbeat(time.time() - 3600) + ) + + reclaimed = await reclaim.reclaim_orphaned(stream, group, "liveworker", logger) + + assert [m[0] for m in reclaimed] == [msg_id] + + @pytest.mark.asyncio async def test_reclaim_orphaned_skips_own_pending( redis_mock, tiny_idle_floor, monkeypatch diff --git a/tests/unit/test_worker_lifecycle.py b/tests/unit/test_worker_lifecycle.py index 7229151..7aeaeeb 100644 --- a/tests/unit/test_worker_lifecycle.py +++ b/tests/unit/test_worker_lifecycle.py @@ -302,6 +302,38 @@ async def test_heartbeat_ping_payload_includes_resources(redis_mock, monkeypatch assert payload["cpu_count"] >= 1 +async def test_heartbeat_key_is_persistent_no_ttl(redis_mock, monkeypatch): + """The heartbeat key carries no TTL so volatile-ttl eviction can't shed it.""" + from shepherd_utils.heartbeat import heartbeat_key + + monkeypatch.setattr(heartbeat_module, "broker_client", redis_mock["broker"]) + hb = Heartbeat("merge_message", "abc", 8, manage_signals=False) + await hb._ping() + # -1 == key exists with no expiry set (persistent). + ttl = await redis_mock["broker"].ttl(heartbeat_key("merge_message", "abc")) + assert ttl == -1 + + +def test_is_heartbeat_fresh_by_last_seen(): + """Freshness rides on last_seen age, not on the key existing.""" + import json + + from shepherd_utils.heartbeat import HEARTBEAT_TTL_SEC, is_heartbeat_fresh + + now = 10_000.0 + assert is_heartbeat_fresh(json.dumps({"last_seen": now - 1}), now=now) is True + assert ( + is_heartbeat_fresh( + json.dumps({"last_seen": now - HEARTBEAT_TTL_SEC - 1}), now=now + ) + is False + ) + # Missing/blank/malformed payloads fail safe toward "not alive". + assert is_heartbeat_fresh(None, now=now) is False + assert is_heartbeat_fresh("{}", now=now) is False + assert is_heartbeat_fresh("not json", now=now) is False + + # --- CPU accounting: cgroup-wide vs. this process --------------------------- diff --git a/workers/merge_message/worker.py b/workers/merge_message/worker.py index d60c474..6b52901 100644 --- a/workers/merge_message/worker.py +++ b/workers/merge_message/worker.py @@ -902,8 +902,8 @@ async def process_query(task, parent_ctx, logger, limiter): return span.set_attribute("drained_callbacks", drained) - logger.debug( - f"[{callback_id}] Drained {drained} callback(s) in " + logger.info( + f"[{callback_id}] Merged {drained} callback(s) in " f"{time.time() - lock_time:.2f}s" ) await remove_lock(response_id, CONSUMER, logger) diff --git a/workers/monitor/alerts.py b/workers/monitor/alerts.py index 1514597..4c1f179 100644 --- a/workers/monitor/alerts.py +++ b/workers/monitor/alerts.py @@ -92,8 +92,6 @@ def evaluate(self, snapshot: Dict[str, Any]) -> Optional[str]: return None if self.kind == "db_capacity": return self._eval_db_capacity(snapshot) - if self.kind == "redis_memory": - return self._eval_redis_memory(snapshot) if self.kind == "redis_eviction": return self._eval_redis_eviction(snapshot) if self.kind == "stuck_pending": @@ -140,29 +138,6 @@ def _eval_db_capacity(self, snapshot: Dict[str, Any]) -> Optional[str]: ) return None - def _eval_redis_memory(self, snapshot: Dict[str, Any]) -> Optional[str]: - """Fire when the broker's used memory crosses ``threshold`` percent of - its ``maxmemory`` cap. - - This is the early warning the OOM-kill incident lacked: eviction (and, - if the cap sits at the container limit, an OOM-kill) only bites once - usage nears the cap, so crossing e.g. 85% is the actionable signal. A - no-op when the broker runs uncapped (``maxmemory`` 0) since there's no - denominator. - """ - redis = snapshot.get("redis", {}) - maxmem = redis.get("maxmemory_bytes", 0) - used = redis.get("used_memory_bytes", 0) - if not maxmem or self.threshold is None: - return None - pct = 100.0 * used / maxmem - if pct >= float(self.threshold): - return ( - f"Redis memory {pct:.1f}% of maxmemory " - f"({used / 1e9:.1f}GB of {maxmem / 1e9:.1f}GB) >= {self.threshold}%" - ) - return None - def _eval_redis_eviction(self, snapshot: Dict[str, Any]) -> Optional[str]: """Fire when the broker evicted keys in the last interval. @@ -273,12 +248,31 @@ def __init__(self, rules: List[Rule]): self._broker_down_since: float = 0.0 self._broker_down_alerted: bool = False self._broker_recovered_at: float = 0.0 + # Redis-eviction grace window. When the broker evicts keys under memory + # pressure it can shed live workers' short-TTL heartbeat keys, making + # them read as zero-alive though they never disconnected. This holds the + # wall-clock time until which worker-down alerts stay suppressed; it's + # pushed forward on every tick that observes an eviction, then lets the + # window elapse so heartbeats can re-register before we trust "down". + self._redis_eviction_grace_until: float = 0.0 # Postgres-availability state machine. Same debounce/fire-once shape as # the broker's, minus the recovery-grace window (a PG outage doesn't # zero out heartbeats, so there's no worker-down flood to suppress). self._pg_up: bool = True self._pg_down_since: float = 0.0 self._pg_down_alerted: bool = False + # Redis memory-pressure state machine (see handle_redis_memory). Like the + # broker/postgres machines it's driven once per poll tick and tracked + # in-process. It tiers usage into ok/warning/critical so only the current + # level alerts (crossing the critical line doesn't also fire the warning), + # re-notifies at most once per cooldown while elevated, and emits a + # recovery alert when usage returns below the warning threshold. + # ``_candidate``/``_since`` debounce a level change so a single-tick spike + # (or dip) doesn't trip it. + self._redis_mem_level: str = "ok" + self._redis_mem_candidate: str = "ok" + self._redis_mem_candidate_since: float = 0.0 + self._redis_mem_last_alerted_at: float = 0.0 @property def in_startup_grace(self) -> bool: @@ -301,6 +295,31 @@ def in_broker_recovery_grace(self) -> bool: time.time() - self._broker_recovered_at ) < settings.monitor_broker_recovery_grace_sec + @property + def in_redis_eviction_grace(self) -> bool: + """True while (or shortly after) the broker is evicting keys. + + Redis key eviction removes keys to free memory; it does *not* close + client connections, so the workers themselves are fine. But the monitor + infers worker liveness from short-TTL heartbeat keys, which eviction can + shed -- so during and just after an eviction spell a live worker can read + as zero-alive. Suppressing worker-down alerts across this window keeps a + Redis memory spike from spamming bogus worker-crash notifications. + """ + return time.time() < self._redis_eviction_grace_until + + def _note_redis_eviction(self, snapshot: Dict[str, Any]) -> None: + """Open/extend the eviction grace window when this tick saw evictions.""" + redis = snapshot.get("redis", {}) or {} + try: + delta = int(redis.get("evicted_keys_delta", 0) or 0) + except (TypeError, ValueError): + delta = 0 + if delta > 0: + self._redis_eviction_grace_until = ( + time.time() + settings.monitor_redis_eviction_grace_sec + ) + async def handle_broker_health(self, is_up: bool) -> None: """Drive the broker up/down state machine and emit broker alerts. @@ -416,6 +435,119 @@ async def handle_postgres_health(self, is_up: bool) -> None: await _record_alert(event) await dispatch(event) + async def handle_redis_memory(self, snapshot: Dict[str, Any]) -> None: + """Drive the Redis memory-pressure state machine and emit its alerts. + + Modeled on the broker/postgres health machines rather than run as two + independent threshold rules, so three behaviors compose cleanly: + + * **Tiering without duplicates**: usage is classified into a single + level -- ok (< warning%), warning (>= warning%), or critical + (>= critical%). Only the current level ever alerts, so crossing the + critical line does NOT also send the warning alert. + * **Day-long backoff**: while usage stays elevated we re-notify at most + once per ``monitor_redis_memory_cooldown_sec`` (a full day by default) + instead of every few minutes. + * **Recovery**: when usage falls back below the warning threshold we send + a one-off "recovered" alert, just like the broker/postgres machines. + + A no-op when the broker runs uncapped (``maxmemory`` 0) -- without a + denominator there's nothing to measure. + """ + now = time.time() + redis = snapshot.get("redis", {}) or {} + maxmem = int(redis.get("maxmemory_bytes", 0) or 0) + used = int(redis.get("used_memory_bytes", 0) or 0) + if maxmem <= 0: + # Uncapped: reset to ok silently. Losing the denominator isn't a real + # recovery, so don't announce one. + self._redis_mem_level = "ok" + self._redis_mem_candidate = "ok" + return + pct = 100.0 * used / maxmem + if pct >= settings.monitor_redis_memory_critical_pct: + target = "critical" + elif pct >= settings.monitor_redis_memory_warning_pct: + target = "warning" + else: + target = "ok" + + # Debounce: an observed level must hold for the grace window before we + # act on it, so a single-tick spike (or dip) doesn't alert or clear. + if target != self._redis_mem_candidate: + self._redis_mem_candidate = target + self._redis_mem_candidate_since = now + return + if target == self._redis_mem_level: + # Sustained at the committed level. Re-notify an elevated level at + # most once per cooldown; ``ok`` needs nothing. + if ( + target != "ok" + and (now - self._redis_mem_last_alerted_at) + >= settings.monitor_redis_memory_cooldown_sec + ): + await self._fire_redis_memory(target, pct, now) + return + # A different level has held steady long enough: commit the transition. + if ( + now - self._redis_mem_candidate_since + ) < settings.monitor_redis_memory_grace_sec: + return + prev = self._redis_mem_level + self._redis_mem_level = target + if target == "ok": + # Recovery: an elevated committed level is only ever reached by + # alerting, so a return to ok is always worth an "it's better" note. + await self._fire_redis_memory_recovery(pct, now) + elif target == "critical" or prev == "ok": + # Fire when entering an elevated level from ok, or escalating to + # critical. A critical->warning de-escalation stays quiet (still + # elevated and already alerted) until it either clears or re-escalates. + await self._fire_redis_memory(target, pct, now) + + async def _fire_redis_memory(self, level: str, pct: float, now: float) -> None: + """Dispatch (and record) a Redis memory warning/critical alert.""" + self._redis_mem_last_alerted_at = now + if level == "critical": + rule = "redis_memory_critical" + severity = "critical" + threshold = settings.monitor_redis_memory_critical_pct + tail = "Eviction of live data is imminent" + else: + rule = "redis_memory_high" + severity = "warning" + threshold = settings.monitor_redis_memory_warning_pct + tail = "Eviction and OOM-kill risk are near" + event = { + "ts": now, + "rule": rule, + "severity": severity, + "detail": f"Redis memory {pct:.1f}% of maxmemory (>= {threshold:.0f}%)", + "message": ( + f"Redis broker memory is at {pct:.1f}% of its maxmemory cap " + f"(>= {threshold:.0f}%). {tail} -- shed load, raise the cap, or " + "lower redis_ttl." + ), + } + await _record_alert(event) + await dispatch(event) + + async def _fire_redis_memory_recovery(self, pct: float, now: float) -> None: + """Dispatch (and record) the Redis memory 'recovered' alert.""" + warn = settings.monitor_redis_memory_warning_pct + event = { + "ts": now, + "rule": "redis_memory_recovered", + "severity": "info", + "detail": f"Redis memory back to {pct:.1f}% of maxmemory", + "message": ( + f"Redis broker memory has fallen back below {warn:.0f}% of its " + f"maxmemory cap (now {pct:.1f}%). Memory pressure has cleared." + ), + } + await _record_alert(event) + await dispatch(event) + async def _in_cooldown(self, rule: Rule) -> bool: try: return bool(await broker_client.exists(f"alert:cooldown:{rule.name}")) @@ -463,13 +595,22 @@ async def evaluate(self, snapshot: Dict[str, Any]) -> List[Dict[str, Any]]: """Return the list of alerts that fired on this snapshot.""" now = snapshot["ts"] fired: List[Dict[str, Any]] = [] - # Worker-down alerts are suppressed both at boot (startup grace) and - # around a broker outage (recovery grace): in either case a worker type - # reads as zero-alive not because it died but because heartbeats haven't - # (re-)registered yet. A broker restart otherwise turns into ~one - # worker-crash alert per worker type -- the flood the broker_down alert - # is meant to replace. - suppress_worker_down = self.in_startup_grace or self.in_broker_recovery_grace + # Refresh the Redis-eviction grace window from this snapshot before we + # decide whether to trust "worker down" readings below. + self._note_redis_eviction(snapshot) + # Worker-down alerts are suppressed at boot (startup grace), around a + # broker outage (recovery grace), and while Redis is evicting keys + # (eviction grace): in every case a worker type reads as zero-alive not + # because it died but because its heartbeat key is missing or hasn't + # (re-)registered yet. A broker restart or a memory-pressure eviction + # spell would otherwise turn into ~one worker-crash alert per worker + # type -- exactly the inaccurate flood these grace windows exist to + # replace with a single, accurate signal. + suppress_worker_down = ( + self.in_startup_grace + or self.in_broker_recovery_grace + or self.in_redis_eviction_grace + ) for rule in self.rules: detail = rule.evaluate(snapshot) if detail is None: @@ -481,8 +622,9 @@ async def evaluate(self, snapshot: Dict[str, Any]) -> List[Dict[str, Any]]: if duration_in_breach < rule.duration: continue if rule.kind == "heartbeat_lost" and rule.worker and suppress_worker_down: - # Boot or broker-recovery window: a zero-alive reading here is an - # artifact of heartbeats not being (re-)registered, not a real + # Boot, broker-recovery, or Redis-eviction window: a zero-alive + # reading here is an artifact of heartbeats not being + # (re-)registered or their keys having been evicted, not a real # loss. Skip WITHOUT arming the cooldown or recording an alert, # and leave the breach streak intact -- so a worker that is # genuinely still down once the window elapses fires promptly. @@ -514,14 +656,15 @@ async def evaluate(self, snapshot: Dict[str, Any]) -> List[Dict[str, Any]]: if not (ev.get("type") == "scale_down" and ev.get("to") == 0): continue if suppress_worker_down: - # Either the whole stack just came up (startup grace) or the - # broker just restarted (recovery grace): persistent worker - # state looks "alive" but current heartbeats haven't arrived - # yet, so every worker type spuriously reads as crashed. Stay - # silent until workers have had a chance to (re-)register. + # The whole stack just came up (startup grace), the broker just + # restarted (recovery grace), or Redis is evicting heartbeat keys + # under memory pressure (eviction grace): persistent worker state + # looks "alive" but current heartbeats are missing or haven't + # arrived yet, so every worker type spuriously reads as crashed. + # Stay silent until workers have had a chance to (re-)register. logger.debug( f"Suppressing worker-down alert for {ev.get('worker')} " - "during startup/broker-recovery grace" + "during startup/broker-recovery/redis-eviction grace" ) continue kind = ev.get("kind", "unknown") diff --git a/workers/monitor/janitor.py b/workers/monitor/janitor.py index 355b19d..06e8624 100644 --- a/workers/monitor/janitor.py +++ b/workers/monitor/janitor.py @@ -26,7 +26,7 @@ from shepherd_utils.db import purge_old_queries from shepherd_utils.db import reap_abandoned_queries as reap_abandoned_queries_db from shepherd_utils.db import reap_completed_callbacks -from shepherd_utils.heartbeat import HEARTBEAT_SCAN_PATTERN +from shepherd_utils.heartbeat import HEARTBEAT_SCAN_PATTERN, is_heartbeat_fresh from . import alerts, storage from .poller import SEED_STREAMS @@ -225,12 +225,24 @@ async def _alert_abandoned(abandoned: List[Dict[str, Any]]) -> None: async def _list_alive_consumers() -> set: - """Return ``{(stream, consumer)}`` for every worker with a live heartbeat.""" - alive = set() + """Return ``{(stream, consumer)}`` for every worker with a *fresh* heartbeat. + + Heartbeat keys are persistent now, so a key existing no longer proves the + worker is alive -- only a recently refreshed ``last_seen`` does. Filter on + freshness so a crashed consumer's lingering key doesn't keep it in the + "alive" set and block its stale consumer-group entry from being cleaned up. + """ + keys = [] async for key in broker_client.scan_iter(match=HEARTBEAT_SCAN_PATTERN, count=200): + keys.append(key) + if not keys: + return set() + values = await broker_client.mget(keys) + alive = set() + for key, raw in zip(keys, values): # worker:heartbeat:{stream}:{consumer} parts = key.split(":", 3) - if len(parts) == 4: + if len(parts) == 4 and is_heartbeat_fresh(raw): alive.add((parts[2], parts[3])) return alive diff --git a/workers/monitor/monitor_alerts.yaml b/workers/monitor/monitor_alerts.yaml index 36e937b..7fd2b41 100644 --- a/workers/monitor/monitor_alerts.yaml +++ b/workers/monitor/monitor_alerts.yaml @@ -6,14 +6,13 @@ # Rule schema: # name: unique identifier (used as cooldown key) # type: threshold | heartbeat_lost | queue_pending | oldest_callback_age -# | db_capacity | redis_memory | redis_eviction | stuck_pending +# | db_capacity | redis_eviction | stuck_pending # metric: for type=threshold: xlen | callbacks_pending | pg_connection_count # stream: redis stream name (for xlen / queue_pending / stuck_pending; # stuck_pending scans all streams when omitted) # worker: worker type name (for heartbeat_lost) -# threshold: number to compare against (redis_memory: percent of maxmemory; -# redis_eviction: keys-per-interval floor; stuck_pending: min -# pending tasks per consumer) +# threshold: number to compare against (redis_eviction: keys-per-interval +# floor; stuck_pending: min pending tasks per consumer) # idle_ms: for stuck_pending: how long a consumer's held tasks must sit # idle before it counts as wedged # duration: how long the condition must hold before firing (e.g. 60s, 5m) @@ -101,20 +100,12 @@ rules: duration: 2m severity: warning - # Redis broker memory: warn before the maxmemory cap starts evicting / risks - # an OOM-kill. No-op when the broker runs uncapped (maxmemory 0). - - name: redis_memory_high - type: redis_memory - threshold: 85 - duration: 2m - severity: warning - message: "Redis broker is over 85% of its maxmemory cap. Eviction and OOM-kill risk are near -- shed load, raise the cap, or lower redis_ttl." - - name: redis_memory_critical - type: redis_memory - threshold: 95 - duration: 1m - severity: critical - message: "Redis broker is over 95% of its maxmemory cap. Eviction of live data is imminent." + # Redis broker memory pressure is monitored automatically by the alert engine + # (like broker/postgres availability) -- there is no rule to configure here. + # It tiers usage into warning (>= 85%) and critical (>= 95%) so only one alert + # fires at a time, re-notifies at most once a day while elevated, and emits a + # redis_memory_recovered alert when usage falls back below the warning line. + # Tune via the monitor_redis_memory_* settings (thresholds, grace, cooldown). # Redis eviction: with volatile-ttl, any eviction means the cap was reached # and blobs are being shed. threshold 0 => alert on the first eviction. diff --git a/workers/monitor/poller.py b/workers/monitor/poller.py index d3b0c5f..24b8a57 100644 --- a/workers/monitor/poller.py +++ b/workers/monitor/poller.py @@ -17,6 +17,7 @@ from shepherd_utils.config import settings from shepherd_utils.db import pool as pg_pool from shepherd_utils.heartbeat import ( + HEARTBEAT_REAP_SEC, HEARTBEAT_SCAN_PATTERN, HEARTBEAT_TTL_SEC, SHUTDOWN_SCAN_PATTERN, @@ -97,16 +98,31 @@ async def _collect_heartbeats() -> List[Dict[str, Any]]: raw_values = await broker_client.mget(keys) now = time.time() workers: List[Dict[str, Any]] = [] - for raw in raw_values: + # Heartbeat keys are persistent (exempt from broker eviction), so a crashed + # worker's key never expires on its own. Delete any left unrefreshed past the + # reap window here -- the poller already has every payload in hand, so this + # is the cheapest place to keep dead consumers from piling up in the keyspace + # and lingering as phantom "stale" rows on the dashboard. + dead_keys: List[str] = [] + for key, raw in zip(keys, raw_values): if not raw: continue try: hb = json.loads(raw) except json.JSONDecodeError: continue - hb["age_sec"] = max(0.0, now - hb.get("last_seen", now)) - hb["stale"] = hb["age_sec"] > HEARTBEAT_TTL_SEC + age = max(0.0, now - hb.get("last_seen", now)) + if age > HEARTBEAT_REAP_SEC: + dead_keys.append(key) + continue + hb["age_sec"] = age + hb["stale"] = age > HEARTBEAT_TTL_SEC workers.append(hb) + if dead_keys: + try: + await broker_client.delete(*dead_keys) + except Exception as e: + logger.debug(f"Failed to reap {len(dead_keys)} dead heartbeat(s): {e}") return workers @@ -390,6 +406,7 @@ def _rollup_workers(workers: List[Dict[str, Any]]) -> Dict[str, Dict[str, Any]]: "task_limit": hb.get("task_limit"), "in_flight": hb.get("in_flight"), "rss_bytes": hb.get("rss_bytes"), + "mem_limit_bytes": hb.get("mem_limit_bytes"), "cpu_pct": hb.get("cpu_pct"), "cpu_count": hb.get("cpu_count"), "stale": hb.get("stale", False), diff --git a/workers/monitor/static/app.js b/workers/monitor/static/app.js index 27baaab..0208910 100644 --- a/workers/monitor/static/app.js +++ b/workers/monitor/static/app.js @@ -249,7 +249,16 @@ const rowClass = c.stale ? ' class="stale-row"' : ""; const running = c.in_flight === null || c.in_flight === undefined ? "-" : fmt(c.in_flight); const limit = fmt(c.task_limit); - const mem = fmtBytes(c.rss_bytes); + // RSS, plus a colorized % of the container's memory cap when one is + // reported (cgroup limit). Uncapped workers just show the raw RSS, + // mirroring how the Redis panel omits the % when maxmemory is 0. + let mem = fmtBytes(c.rss_bytes); + if (c.rss_bytes != null && c.mem_limit_bytes) { + const memPct = 100 * c.rss_bytes / c.mem_limit_bytes; + const memColor = memPct >= 90 ? "var(--bad)" : memPct >= 80 ? "var(--warn)" : null; + const memText = `${mem} (${memPct.toFixed(0)}%)`; + mem = memColor ? `${memText}` : memText; + } // CPU is a top-style "% of a single core" (can exceed 100 on // multi-core work), so show the core allocation alongside it to keep // the number interpretable. @@ -399,7 +408,12 @@ const usedH = redis.used_memory_human || (redis.used_memory_bytes ? fmtBytes(redis.used_memory_bytes) : "-"); const maxBytes = redis.maxmemory_bytes || 0; - const memValue = maxBytes ? `${usedH} / ${fmtBytes(maxBytes)}` : usedH; + // Inline the % of the cap alongside used/cap so the headroom reads at a + // glance without cross-referencing the separate "Redis mem %" row below. + let memValue = maxBytes ? `${usedH} / ${fmtBytes(maxBytes)}` : usedH; + if (maxBytes && redis.maxmemory_pct != null) { + memValue += ` (${redis.maxmemory_pct.toFixed(1)}%)`; + } let memPct = "-"; if (redis.maxmemory_pct != null) { const p = redis.maxmemory_pct; diff --git a/workers/monitor/worker.py b/workers/monitor/worker.py index 78e63a2..7f407e0 100644 --- a/workers/monitor/worker.py +++ b/workers/monitor/worker.py @@ -118,6 +118,9 @@ async def _poll_loop(engine: alerts.AlertEngine) -> None: # PG up/down alert off that, same cadence as the broker check. pg_up = "error" not in (snapshot.get("postgres") or {}) await engine.handle_postgres_health(pg_up) + # Redis memory pressure is an automatic engine state machine (like + # broker/postgres health), not a YAML rule -- drive it each tick. + await engine.handle_redis_memory(snapshot) _latest_snapshot = snapshot # Persist history at a slower cadence than the live UI tick to # keep Redis memory bounded.