bvandeusenandClaude Opus 5.5
2df8431de9
The same Holding as the SNMP poll, for the rest of Steward's own collectors: - Monitors: the list is read within 10 s or the last one read stands in; a start the database doesn't take is kept in memory, so a monitor isn't checked twice; each result is held and stored at the time it was checked. A stall shows on the monitors as the checks it saw, not a gap. - UniFi and Traefik: each poll's writes are held (hold_write) and run oldest first in one transaction; every reading is stamped with its scrape. - A held round the database refuses (as opposed to can't be reached) is stored alone and let go, so it can't hold up every round after it. The Prometheus scrape is left as it is: its one target, Fabled Archives, keeps its own database on the same NFS. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
79 lines
2.6 KiB
Python
79 lines
2.6 KiB
Python
"""Collected rounds held in memory until the database takes them (#5995):
|
|
stored oldest first, one writer at a time, bounded, and kept through a
|
|
store that fails or never answers."""
|
|
import asyncio
|
|
from types import SimpleNamespace as NS
|
|
|
|
from steward.core.holding import Holding, holding
|
|
|
|
|
|
def test_rounds_held_while_the_store_hangs_are_stored_in_order_once_it_answers():
|
|
async def main():
|
|
gate, stored = asyncio.Event(), []
|
|
|
|
async def store(rounds):
|
|
await gate.wait()
|
|
stored.append(list(rounds))
|
|
|
|
held = Holding("t", store, limit=10)
|
|
held.hold("a")
|
|
assert await held.flush(0.05) is False # the store hangs: the caller goes on
|
|
held.hold("b")
|
|
assert await held.flush(0.05) is False and len(held) == 2
|
|
gate.set()
|
|
assert await held.flush(1.0) is True and len(held) == 0
|
|
return stored
|
|
|
|
assert asyncio.run(main()) == [["a"], ["b"]] # "b" waited for the store in flight
|
|
|
|
|
|
def test_past_its_limit_the_oldest_round_goes_and_is_counted():
|
|
held = Holding("t", store=None, limit=2)
|
|
for r in ("a", "b", "c"):
|
|
held.hold(r)
|
|
assert len(held) == 2 and held.dropped == 1 and [i for _, i in held._held] == ["b", "c"]
|
|
|
|
|
|
def test_a_store_that_cant_reach_the_database_or_overruns_its_deadline_keeps_the_rounds_held():
|
|
async def main():
|
|
async def unreachable(rounds):
|
|
raise ConnectionRefusedError("database down")
|
|
|
|
held = Holding("t", unreachable, limit=10)
|
|
held.hold("a")
|
|
assert await held.flush(1.0) is False and len(held) == 1
|
|
|
|
async def never(rounds):
|
|
await asyncio.sleep(3600)
|
|
|
|
stuck = Holding("t", never, limit=10, deadline_s=0.05)
|
|
stuck.hold("a")
|
|
assert await stuck.flush(1.0) is False and len(stuck) == 1 # given up, not lost
|
|
|
|
asyncio.run(main())
|
|
|
|
|
|
def test_a_round_the_database_refuses_is_dropped_and_holds_up_nobody_after_it():
|
|
async def main():
|
|
stored = []
|
|
|
|
async def store(rounds):
|
|
if "bad" in rounds:
|
|
raise ValueError("value out of range")
|
|
stored.extend(rounds)
|
|
|
|
held = Holding("t", store, limit=10)
|
|
for r in ("a", "bad", "c"):
|
|
held.hold(r)
|
|
assert await held.flush(1.0) is True
|
|
return stored
|
|
|
|
assert asyncio.run(main()) == ["a", "c"]
|
|
|
|
|
|
def test_an_app_has_one_holding_per_collector():
|
|
app = NS()
|
|
first = holding(app, "SNMP poll", store=None, limit=5)
|
|
assert holding(app, "SNMP poll", store=None, limit=5) is first
|
|
assert holding(app, "UniFi poll", store=None, limit=5) is not first
|