Change interval
This commit is contained in:
+13
-9
@@ -75,18 +75,22 @@ def check_endpoint(ep, default_timeout: float, timestamp_ns: int) -> CheckResult
|
||||
)
|
||||
|
||||
|
||||
def log_result(result: CheckResult) -> None:
|
||||
level = logging.INFO if result.success else logging.WARNING
|
||||
logger.log(
|
||||
level,
|
||||
"endpoint=%s success=%s status=%s expected=%s ip=%s time_ms=%s error=%s",
|
||||
result.endpoint_name, result.success, result.status_code,
|
||||
result.expected_status, result.resolved_ip,
|
||||
None if result.response_time_ms is None else round(result.response_time_ms, 1),
|
||||
result.error,
|
||||
)
|
||||
|
||||
|
||||
def check_all(endpoints, default_timeout: float, timestamp_ns: int) -> List[CheckResult]:
|
||||
results = []
|
||||
for ep in endpoints:
|
||||
result = check_endpoint(ep, default_timeout, timestamp_ns)
|
||||
level = logging.INFO if result.success else logging.WARNING
|
||||
logger.log(
|
||||
level,
|
||||
"endpoint=%s success=%s status=%s expected=%s ip=%s time_ms=%s error=%s",
|
||||
result.endpoint_name, result.success, result.status_code,
|
||||
result.expected_status, result.resolved_ip,
|
||||
None if result.response_time_ms is None else round(result.response_time_ms, 1),
|
||||
result.error,
|
||||
)
|
||||
log_result(result)
|
||||
results.append(result)
|
||||
return results
|
||||
|
||||
+2
-23
@@ -3,11 +3,9 @@ from __future__ import annotations
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
|
||||
from app.checker import check_all
|
||||
from app.config import ConfigError, load_config
|
||||
from app.influx_writer import write_results
|
||||
from app.scheduler import run_forever
|
||||
|
||||
logger = logging.getLogger("uptime_monitor")
|
||||
|
||||
@@ -20,13 +18,6 @@ def setup_logging() -> None:
|
||||
)
|
||||
|
||||
|
||||
def run_cycle(cfg) -> list:
|
||||
timestamp_ns = time.time_ns()
|
||||
results = check_all(cfg.endpoints, cfg.check.timeout_seconds, timestamp_ns)
|
||||
write_results(cfg.influx, results)
|
||||
return results
|
||||
|
||||
|
||||
def main() -> None:
|
||||
setup_logging()
|
||||
config_path = os.environ.get("CONFIG_PATH", "/config/config.yaml")
|
||||
@@ -42,19 +33,7 @@ def main() -> None:
|
||||
len(cfg.endpoints), config_path, cfg.check.interval_seconds,
|
||||
)
|
||||
|
||||
while True:
|
||||
results = run_cycle(cfg)
|
||||
|
||||
if any(not r.success for r in results):
|
||||
logger.warning(
|
||||
"at least one endpoint failed this cycle; "
|
||||
"re-checking all endpoints immediately in %.0fs",
|
||||
cfg.check.immediate_recheck_delay_seconds,
|
||||
)
|
||||
time.sleep(cfg.check.immediate_recheck_delay_seconds)
|
||||
run_cycle(cfg)
|
||||
|
||||
time.sleep(cfg.check.interval_seconds)
|
||||
run_forever(cfg)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
@@ -0,0 +1,67 @@
|
||||
"""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)]
|
||||
Reference in New Issue
Block a user