68 lines
2.4 KiB
Python
68 lines
2.4 KiB
Python
"""Rolling scheduler: spreads endpoint checks evenly across the interval.
|
|
|
|
With N endpoints and `interval_seconds` = T, each endpoint gets its own
|
|
slot, T/N seconds apart, and is checked once every T seconds — e.g. 6
|
|
endpoints with interval_seconds: 3600 means one endpoint is checked every
|
|
10 minutes, rotating through all six over the course of an hour. This is
|
|
NOT "check everything, then sleep T seconds": individual endpoint checks
|
|
are spread out, not bunched together.
|
|
|
|
If any single check fails, ALL endpoints are checked once immediately
|
|
(out of band) as an extra confirmation pass, and the rotation is then
|
|
re-staggered from that point so it stays evenly spread afterward.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import time
|
|
from typing import List
|
|
|
|
from .checker import check_all, check_endpoint, log_result
|
|
from .influx_writer import write_results
|
|
|
|
logger = logging.getLogger("uptime_monitor.scheduler")
|
|
|
|
|
|
def run_forever(cfg) -> None:
|
|
endpoints = cfg.endpoints
|
|
interval = cfg.check.interval_seconds
|
|
n = len(endpoints)
|
|
stagger = interval / n
|
|
|
|
now = time.monotonic()
|
|
next_at: List[float] = [now + i * stagger for i in range(n)]
|
|
|
|
logger.info(
|
|
"rolling schedule: %d endpoint(s) spread across %ss (~%.1fs apart)",
|
|
n, interval, stagger,
|
|
)
|
|
|
|
while True:
|
|
idx = min(range(n), key=lambda i: next_at[i])
|
|
wait = next_at[idx] - time.monotonic()
|
|
if wait > 0:
|
|
time.sleep(wait)
|
|
|
|
ep = endpoints[idx]
|
|
result = check_endpoint(ep, cfg.check.timeout_seconds, time.time_ns())
|
|
log_result(result)
|
|
write_results(cfg.influx, [result])
|
|
|
|
# schedule this endpoint's next regular slot
|
|
next_at[idx] += interval
|
|
|
|
if not result.success:
|
|
logger.warning(
|
|
"endpoint=%s failed; re-checking all endpoints immediately in %.0fs",
|
|
ep.name, cfg.check.immediate_recheck_delay_seconds,
|
|
)
|
|
time.sleep(cfg.check.immediate_recheck_delay_seconds)
|
|
results = check_all(endpoints, cfg.check.timeout_seconds, time.time_ns())
|
|
write_results(cfg.influx, results)
|
|
|
|
# Re-stagger everyone from now so the rotation stays evenly
|
|
# spread out afterward, instead of bunching back up.
|
|
now = time.monotonic()
|
|
next_at = [now + (i + 1) * stagger for i in range(n)]
|