135 lines
4.1 KiB
Python
135 lines
4.1 KiB
Python
from __future__ import annotations
|
|
|
|
import dataclasses
|
|
from typing import List, Optional
|
|
|
|
import yaml
|
|
|
|
from . import dns_resolver
|
|
|
|
|
|
class ConfigError(Exception):
|
|
"""Raised for any problem with the config file's contents."""
|
|
|
|
|
|
@dataclasses.dataclass
|
|
class Endpoint:
|
|
name: str
|
|
host: str
|
|
scheme: str # "http" or "https"
|
|
port: int
|
|
expected_status: int
|
|
# DNS server used to resolve `host` — the system/router resolver is
|
|
# never consulted, see app/dns_resolver.py. Optional (and unused) when
|
|
# `host` is already a literal IP address, since there's nothing to
|
|
# resolve in that case.
|
|
dns_server: Optional[str] = None
|
|
path: str = "/"
|
|
timeout_seconds: Optional[float] = None
|
|
verify_tls: bool = True # set False for endpoints with self-signed certs
|
|
|
|
|
|
@dataclasses.dataclass
|
|
class InfluxConfig:
|
|
url: str
|
|
token: str
|
|
org: str
|
|
bucket: str
|
|
verify_ssl: bool = False # default off: influx uses a self-signed cert
|
|
|
|
|
|
@dataclasses.dataclass
|
|
class CheckConfig:
|
|
interval_seconds: int = 600
|
|
timeout_seconds: float = 10.0
|
|
immediate_recheck_delay_seconds: float = 5.0
|
|
|
|
|
|
@dataclasses.dataclass
|
|
class AppConfig:
|
|
influx: InfluxConfig
|
|
check: CheckConfig
|
|
endpoints: List[Endpoint]
|
|
|
|
|
|
def _require(d: dict, key: str, ctx: str):
|
|
value = d.get(key)
|
|
if value in (None, ""):
|
|
raise ConfigError(f"Missing required field '{key}' in {ctx}")
|
|
return value
|
|
|
|
|
|
def load_config(path: str) -> AppConfig:
|
|
try:
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
raw = yaml.safe_load(f)
|
|
except FileNotFoundError:
|
|
raise ConfigError(f"Config file not found: {path}")
|
|
except yaml.YAMLError as e:
|
|
raise ConfigError(f"Invalid YAML in config file '{path}': {e}")
|
|
|
|
if not raw:
|
|
raise ConfigError(f"Config file '{path}' is empty")
|
|
|
|
influx_raw = raw.get("influx") or {}
|
|
influx = InfluxConfig(
|
|
url=_require(influx_raw, "url", "influx"),
|
|
token=_require(influx_raw, "token", "influx"),
|
|
org=_require(influx_raw, "org", "influx"),
|
|
bucket=_require(influx_raw, "bucket", "influx"),
|
|
verify_ssl=bool(influx_raw.get("verify_ssl", False)),
|
|
)
|
|
|
|
check_raw = raw.get("check") or {}
|
|
check = CheckConfig(
|
|
interval_seconds=int(check_raw.get("interval_seconds", 600)),
|
|
timeout_seconds=float(check_raw.get("timeout_seconds", 10.0)),
|
|
immediate_recheck_delay_seconds=float(
|
|
check_raw.get("immediate_recheck_delay_seconds", 5.0)
|
|
),
|
|
)
|
|
|
|
endpoints_raw = raw.get("endpoints") or []
|
|
if not endpoints_raw:
|
|
raise ConfigError("No endpoints configured under 'endpoints'")
|
|
|
|
endpoints: List[Endpoint] = []
|
|
seen_names = set()
|
|
for i, ep in enumerate(endpoints_raw):
|
|
ctx = f"endpoints[{i}]"
|
|
name = _require(ep, "name", ctx)
|
|
if name in seen_names:
|
|
raise ConfigError(f"Duplicate endpoint name: '{name}'")
|
|
seen_names.add(name)
|
|
|
|
scheme = str(_require(ep, "scheme", ctx)).lower()
|
|
if scheme not in ("http", "https"):
|
|
raise ConfigError(
|
|
f"{ctx} ('{name}'): scheme must be 'http' or 'https', got '{scheme}'"
|
|
)
|
|
|
|
host = _require(ep, "host", ctx)
|
|
dns_server = ep.get("dns_server") or None
|
|
if not dns_resolver.is_ip_literal(host) and not dns_server:
|
|
raise ConfigError(
|
|
f"{ctx} ('{name}'): 'dns_server' is required when 'host' is not "
|
|
"a literal IP address"
|
|
)
|
|
|
|
timeout = ep.get("timeout_seconds")
|
|
endpoints.append(
|
|
Endpoint(
|
|
name=name,
|
|
host=host,
|
|
scheme=scheme,
|
|
port=int(_require(ep, "port", ctx)),
|
|
dns_server=dns_server,
|
|
expected_status=int(_require(ep, "expected_status", ctx)),
|
|
path=ep.get("path", "/"),
|
|
timeout_seconds=float(timeout) if timeout else None,
|
|
verify_tls=bool(ep.get("verify_tls", True)),
|
|
)
|
|
)
|
|
|
|
return AppConfig(influx=influx, check=check, endpoints=endpoints)
|