healthcheck-container/monitors.py

import ipaddress
import logging
import socket
import threading
import time
import urllib.request

import wifi

log = logging.getLogger(__name__)


class State:
    def __init__(self):
        self._lock = threading.Lock()
        self._data = {}
        self._listeners = []

    def on_change(self, listener):
        self._listeners.append(listener)

    def update(self, **fields):
        with self._lock:
            self._data.update(fields)
            snapshot = dict(self._data)
        for listener in self._listeners:
            try:
                listener(snapshot)
            except Exception:
                log.exception("state listener failed")

    def snapshot(self):
        with self._lock:
            return dict(self._data)


class Monitor(threading.Thread):
    def __init__(self, name, interval, state, storage):
        super().__init__(name=name, daemon=True)
        self.interval = interval
        self.state = state
        self.storage = storage

    def run(self):
        while True:
            started = time.monotonic()
            try:
                self.tick()
            except Exception as exc:
                log.warning("%s: %s", self.name, exc)
                self.state.update(**{f"{self.name}_error": str(exc)})
            elapsed = time.monotonic() - started
            time.sleep(max(1.0, self.interval - elapsed))

    def tick(self):
        raise NotImplementedError


def tcp_latency_ms(target, timeout):
    if ":" in target:
        host, port = target.rsplit(":", 1)
    else:
        host, port = target, "443"
    started = time.monotonic()
    try:
        with socket.create_connection((host, int(port)), timeout=timeout):
            pass
    except (OSError, ValueError):
        return None
    return round((time.monotonic() - started) * 1000, 1)


class Reachability(Monitor):
    def __init__(self, cfg, state, storage):
        super().__init__("internet", cfg.reach_interval, state, storage)
        self.targets = cfg.reach_targets
        self.timeout = cfg.reach_timeout
        self.up = None

    def tick(self):
        ts = time.time()
        results = {target: tcp_latency_ms(target, self.timeout) for target in self.targets}
        reachable = [ms for ms in results.values() if ms is not None]
        up = bool(reachable)

        samples = [(f"reach.{target}", ms) for target, ms in results.items()]
        samples.append(("internet.up", 1 if up else 0))
        self.storage.add_samples(samples, ts)

        if up != self.up:
            if not up:
                self.storage.add_event("internet_down", "Internet unreachable: all targets failed")
            elif self.up is False:
                self.storage.add_event("internet_up", "Internet reachable again")
            self.up = up

        self.state.update(
            internet_up=up,
            internet_latency_ms=min(reachable) if reachable else None,
            reach=results,
            internet_checked_at=ts,
            internet_error=None,
        )


class Dns(Monitor):
    def __init__(self, cfg, state, storage):
        super().__init__("dns", cfg.dns_interval, state, storage)
        self.name_to_resolve = cfg.dns_name
        self.up = None

    def tick(self):
        ts = time.time()
        started = time.monotonic()
        try:
            socket.getaddrinfo(self.name_to_resolve, 443, proto=socket.IPPROTO_TCP)
            ms = round((time.monotonic() - started) * 1000, 1)
        except socket.gaierror:
            ms = None
        up = ms is not None

        self.storage.add_sample("dns.latency_ms", ms, ts)
        if up != self.up:
            if not up:
                self.storage.add_event("dns_down", f"DNS resolution of {self.name_to_resolve} failed")
            elif self.up is False:
                self.storage.add_event("dns_up", "DNS resolution working again")
            self.up = up

        self.state.update(dns_up=up, dns_latency_ms=ms, dns_checked_at=ts, dns_error=None)


class PublicIp(Monitor):
    def __init__(self, cfg, state, storage):
        super().__init__("public_ip", cfg.public_ip_interval, state, storage)
        self.urls = cfg.public_ip_urls
        self.timeout = cfg.public_ip_timeout
        previous = storage.get("public_ip")
        since = storage.get("public_ip_since")
        if previous:
            state.update(public_ip=previous, public_ip_since=float(since) if since else None)

    def tick(self):
        ts = time.time()
        ip, source, error = None, None, "no lookup services configured"
        for url in self.urls:
            try:
                request = urllib.request.Request(url, headers={"User-Agent": "healthcheck-container"})
                with urllib.request.urlopen(request, timeout=self.timeout) as response:
                    body = response.read(200).decode("utf-8", errors="replace").strip()
                ip = str(ipaddress.ip_address(body))
                source = url
                break
            except (OSError, ValueError) as exc:
                error = f"{url}: {exc}"

        if ip is None:
            self.state.update(public_ip_checked_at=ts, public_ip_error=error)
            return

        previous = self.storage.get("public_ip")
        if previous != ip:
            if previous:
                self.storage.add_event("public_ip_changed", f"Public IP changed from {previous} to {ip}")
            else:
                self.storage.add_event("public_ip", f"Public IP is {ip}")
            self.storage.set("public_ip", ip)
            self.storage.set("public_ip_since", str(ts))

        since = self.storage.get("public_ip_since")
        self.state.update(
            public_ip=ip,
            public_ip_source=source,
            public_ip_since=float(since) if since else ts,
            public_ip_checked_at=ts,
            public_ip_error=None,
        )


class Uptime(Monitor):
    def __init__(self, cfg, state, storage):
        super().__init__("host", cfg.uptime_interval, state, storage)

    def tick(self):
        with open("/proc/uptime") as handle:
            uptime = float(handle.read().split()[0])
        boot_time = time.time() - uptime
        previous = self.storage.get("boot_time")
        if previous is None or abs(float(previous) - boot_time) > 120:
            if previous is not None:
                self.storage.add_event("reboot", "Host rebooted")
            self.storage.set("boot_time", str(boot_time))
        self.state.update(uptime_s=int(uptime), boot_time=boot_time, host_error=None)


class Housekeeping(Monitor):
    def __init__(self, cfg, state, storage):
        super().__init__("housekeeping", 3600, state, storage)
        self.retention_days = cfg.sample_retention_days

    def tick(self):
        self.storage.prune_samples(self.retention_days)


def resolve_wifi_interface(cfg, state):
    found = wifi.interfaces()
    wireless = [item["name"] for item in found if item["wireless"]]
    fields = {"wifi_interfaces": found, "wifi_interface": cfg.wifi_interface}

    if not cfg.wifi_interface:
        if wireless:
            fields["wifi_status"] = "not_configured"
            fields["wifi_error"] = (
                "No interface selected. Set WIFI_INTERFACE to one of: " + ", ".join(wireless)
            )
        else:
            fields["wifi_status"] = "unavailable"
            fields["wifi_error"] = (
                "No wireless interfaces are visible from this container. "
                "Run it with network_mode: host on a device that has Wi-Fi."
            )
        state.update(**fields)
        return None

    if cfg.wifi_interface not in wireless:
        available = ", ".join(wireless) if wireless else "none"
        fields["wifi_status"] = "missing"
        fields["wifi_error"] = (
            f"Interface '{cfg.wifi_interface}' is not a visible wireless interface. "
            f"Wireless interfaces visible: {available}."
        )
        state.update(**fields)
        return None

    fields["wifi_status"] = "ok"
    fields["wifi_error"] = None
    state.update(**fields)
    return cfg.wifi_interface


class WifiLink(Monitor):
    def __init__(self, cfg, state, storage):
        super().__init__("wifi_link", cfg.wifi_link_interval, state, storage)
        self.cfg = cfg
        self.connected = None
        self.bssid = None
        self.counters = None

    def tick(self):
        iface = resolve_wifi_interface(self.cfg, self.state)
        if iface is None:
            return

        ts = time.time()
        info = wifi.link(iface)
        if info["connected"]:
            info.update(wifi.station(iface))

        self._record_events(info)
        self._record_samples(info, ts)
        self.state.update(wifi_link=info, wifi_link_checked_at=ts, wifi_link_error=None)

    def _record_events(self, info):
        connected = info["connected"]
        bssid = info.get("bssid")
        where = f"{info.get('ssid', '?')} ({bssid}, ch {info.get('channel', '?')}, {info.get('band', '?')})"
        if connected != self.connected:
            if connected:
                self.storage.add_event("wifi_connected", f"Wi-Fi connected to {where}")
            else:
                self.storage.add_event("wifi_disconnected", "Wi-Fi disconnected")
            self.connected = connected
            self.counters = None
        elif connected and bssid != self.bssid:
            self.storage.add_event("wifi_roamed", f"Wi-Fi roamed to {where}")
            self.counters = None
        self.bssid = bssid

    def _record_samples(self, info, ts):
        if not info["connected"]:
            self.storage.add_sample("wifi.signal_dbm", None, ts)
            self.counters = None
            return
        samples = [
            ("wifi.signal_dbm", info.get("signal_dbm")),
            ("wifi.tx_bitrate_mbps", info.get("tx_bitrate_mbps")),
            ("wifi.rx_bitrate_mbps", info.get("rx_bitrate_mbps")),
        ]
        counters = (info.get("tx_retries"), info.get("tx_failed"), info.get("beacon_loss"))
        if None not in counters:
            if self.counters and all(now >= before for now, before in zip(counters, self.counters)):
                deltas = [now - before for now, before in zip(counters, self.counters)]
                samples += [
                    ("wifi.tx_retries", deltas[0]),
                    ("wifi.tx_failed", deltas[1]),
                    ("wifi.beacon_loss", deltas[2]),
                ]
            self.counters = counters
        self.storage.add_samples(samples, ts)


class WifiScan(Monitor):
    def __init__(self, cfg, state, storage):
        super().__init__("wifi_scan", cfg.wifi_scan_interval, state, storage)
        self.cfg = cfg

    def tick(self):
        iface = resolve_wifi_interface(self.cfg, self.state)
        if iface is None:
            self.state.update(wifi_scan=None, wifi_scan_error=None)
            return
        aps = wifi.scan(iface)
        self.state.update(
            wifi_scan={"aps": aps, "count": len(aps), "ts": time.time()},
            wifi_scan_error=None,
        )


def build_monitors(cfg, state, storage):
    return [
        Reachability(cfg, state, storage),
        Dns(cfg, state, storage),
        PublicIp(cfg, state, storage),
        Uptime(cfg, state, storage),
        WifiLink(cfg, state, storage),
        WifiScan(cfg, state, storage),
        Housekeeping(cfg, state, storage),
    ]