ServerMonitor/checks.py

"""Health check helpers for ServerMonitor.

All check functions return a tuple ``(ok, message)`` and never raise.
"""

import json
import logging
import socket
import subprocess
import time
import urllib.parse
import urllib.request
from urllib.error import HTTPError, URLError

logger = logging.getLogger("servermonitor.checks")

# Tracks whether the ping unavailability warning has been logged already.
_ping_unavailable_logged = False


def _safe_message(text, max_len=250):
    """Return a single-line string truncated to ``max_len`` characters."""
    text = text.replace("\n", " ")
    if len(text) > max_len:
        text = text[: max_len - 1] + "…"
    return text


def ping_check(host, timeout):
    """Ping a host once and return ``(ok, message)``.

    On busybox (Alpine) ``-W`` is seconds. If ICMP is unavailable or disallowed
    the check degrades gracefully: it returns ``ok=True`` with a message that
    says ping is unavailable, so the service does not fail purely because the
    container cannot ping.
    """
    global _ping_unavailable_logged

    try:
        start = time.monotonic()
        result = subprocess.run(
            ["ping", "-c", "1", "-W", str(timeout), host],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            timeout=timeout + 2,
        )
        latency = int((time.monotonic() - start) * 1000)

        if result.returncode == 0:
            return True, f"ping ok, {latency}ms"

        # Failure: host unreachable or no reply
        return False, f"ping failed (code {result.returncode})"

    except FileNotFoundError:
        if not _ping_unavailable_logged:
            logger.warning("ping binary not available, degrading ping check")
            _ping_unavailable_logged = True
        return True, "ping unavailable"

    except PermissionError:
        if not _ping_unavailable_logged:
            logger.warning("ping permission denied, degrading ping check")
            _ping_unavailable_logged = True
        return True, "ping unavailable"

    except subprocess.TimeoutExpired:
        return False, f"ping timeout after {timeout + 2}s"

    except Exception as exc:  # noqa: BLE001
        return False, f"ping error: {exc}"


def http_check(name, url, expected_codes, timeout):
    """Check that ``url`` returns a status code in ``expected_codes``.

    Follows no redirects. Returns ``(ok, message)`` where the message includes
    the status code and latency.
    """
    expected_codes = set(expected_codes)
    start = time.monotonic()

    try:
        req = urllib.request.Request(
            url,
            method="GET",
            headers={"User-Agent": "ServerMonitor/1.0"},
        )

        # Custom redirect handler that does not follow redirects, so the
        # original status code (e.g. 301/302) is preserved.
        class NoRedirectHandler(urllib.request.HTTPRedirectHandler):
            def http_error_302(self, req, fp, code, msg, headers):
                return fp

            http_error_301 = http_error_303 = http_error_307 = http_error_302

        opener = urllib.request.build_opener(NoRedirectHandler)
        with opener.open(req, timeout=timeout) as response:
            status = response.status
            latency = int((time.monotonic() - start) * 1000)
            if status in expected_codes:
                return True, f"HTTP {status}, {latency}ms"
            return False, f"HTTP {status} (expected {sorted(expected_codes)}), {latency}ms"

    except HTTPError as exc:
        latency = int((time.monotonic() - start) * 1000)
        status = exc.code
        if status in expected_codes:
            return True, f"HTTP {status}, {latency}ms"
        return False, f"HTTP {status} (expected {sorted(expected_codes)}), {latency}ms"

    except URLError as exc:
        reason = str(exc.reason)
        return False, f"connection error: {_safe_message(reason)}"

    except TimeoutError:
        return False, f"timeout after {timeout}s"

    except Exception as exc:  # noqa: BLE001
        return False, f"HTTP error: {_safe_message(str(exc))}"


def tcp_check(name, host, port, timeout):
    """Check that a TCP connection to ``host:port`` succeeds.

    Returns ``(ok, message)`` with the latency in milliseconds.
    """
    start = time.monotonic()
    sock = None
    try:
        sock = socket.create_connection((host, port), timeout=timeout)
        latency = int((time.monotonic() - start) * 1000)
        return True, f"TCP ok, {latency}ms"
    except socket.timeout:
        return False, f"TCP timeout after {timeout}s"
    except OSError as exc:
        return False, f"TCP error: {_safe_message(str(exc))}"
    except Exception as exc:  # noqa: BLE001
        return False, f"TCP error: {_safe_message(str(exc))}"
    finally:
        if sock is not None:
            try:
                sock.close()
            except Exception:  # noqa: BLE001
                pass


def _netdata_json(url, timeout):
    """Fetch and parse a Netdata JSON endpoint. Returns dict or None."""
    req = urllib.request.Request(url, headers={"User-Agent": "ServerMonitor/1.0"})
    with urllib.request.urlopen(req, timeout=timeout) as response:
        data = response.read()
        return json.loads(data.decode("utf-8"))


def netdata_alarms(base_url, timeout):
    """Check Netdata active alarms.

    ``status: false`` in the response or any non-empty alarms object means the
    check fails. Returns ``(ok, message)``.
    """
    url = f"{base_url}/api/v1/alarms?active"
    try:
        payload = _netdata_json(url, timeout)
    except Exception as exc:  # noqa: BLE001
        return False, f"Netdata alarms API error: {_safe_message(str(exc))}"

    if not isinstance(payload, dict):
        return False, "Netdata alarms invalid response format"

    net_status = payload.get("status", True)
    alarms = payload.get("alarms", {})

    if not net_status:
        return False, "Netdata status false"

    if not alarms:
        return True, "Netdata alarms empty"

    parts = []
    for name, details in alarms.items():
        if not isinstance(details, dict):
            details = {}
        status = details.get("status", "?")
        value = details.get("value", "?")
        info = details.get("info", "")
        parts.append(f"{name} [{status}]: {value} — {info}")

    return False, "Netdata alarms: " + "; ".join(parts)


def _row_values(labels, row):
    """Convert a Netdata data row into a ``{label: float or None}`` dict.

    The ``time`` dimension is skipped. Missing values (``null``) become
    ``None``.
    """
    values = {}
    for i, label in enumerate(labels):
        if label == "time":
            continue
        try:
            raw = row[i]
            values[label] = float(raw) if raw is not None else None
        except (TypeError, ValueError, IndexError):
            values[label] = None
    return values


def netdata_metrics(base_url, timeout):
    """Collect Netdata metrics for threshold checks.

    Returns a dict with keys ``cpu_percent``, ``ram_used_mb``, ``ram_total_mb``,
    ``disk_used_percent``. Missing values are represented as ``None``.
    """
    result = {
        "cpu_percent": None,
        "ram_used_mb": None,
        "ram_total_mb": None,
        "disk_used_percent": None,
    }

    try:
        cpu_url = (
            f"{base_url}/api/v1/data?chart=system.cpu&after=-1&points=1&format=json"
        )
        cpu_data = _netdata_json(cpu_url, timeout)
        if isinstance(cpu_data, dict):
            labels = cpu_data.get("labels", [])
            data_rows = cpu_data.get("data", [])
            if data_rows and labels:
                values = _row_values(labels, data_rows[0])
                if values.get("idle") is not None:
                    # Older Netdata: idle dimension present.
                    result["cpu_percent"] = max(0.0, 100.0 - values["idle"])
                else:
                    # Netdata v2: only busy dimensions, their sum is ~utilization.
                    total = sum(v for v in values.values() if v is not None)
                    result["cpu_percent"] = max(0.0, min(100.0, total))
    except Exception as exc:  # noqa: BLE001
        logger.warning("Netdata CPU metrics fetch failed: %s", exc)

    try:
        ram_url = (
            f"{base_url}/api/v1/data?chart=system.ram&after=-1&points=1&format=json"
        )
        ram_data = _netdata_json(ram_url, timeout)
        if isinstance(ram_data, dict):
            labels = ram_data.get("labels", [])
            data_rows = ram_data.get("data", [])
            if data_rows and labels:
                values = _row_values(labels, data_rows[0])
                total = 0.0
                for label in ("used", "free", "cached", "buffers"):
                    value = values.get(label)
                    if value is not None:
                        total += value
                result["ram_used_mb"] = values.get("used")
                result["ram_total_mb"] = total if total > 0 else None
    except Exception as exc:  # noqa: BLE001
        logger.warning("Netdata RAM metrics fetch failed: %s", exc)

    try:
        # Discover per-mount disk charts (named disk_space.<mount> on Netdata v2)
        # and take the maximum used-percent across them.
        charts_data = _netdata_json(f"{base_url}/api/v1/charts", timeout)
        disk_charts = []
        if isinstance(charts_data, dict):
            disk_charts = [
                key
                for key in charts_data.get("charts", {})
                if key.startswith("disk_space.")
            ][:8]

        max_used = None
        for chart in disk_charts:
            try:
                disk_url = (
                    f"{base_url}/api/v1/data"
                    f"?chart={urllib.parse.quote(chart)}&after=-1&points=1&format=json"
                )
                disk_data = _netdata_json(disk_url, timeout)
                if not isinstance(disk_data, dict):
                    continue
                labels = disk_data.get("labels", [])
                data_rows = disk_data.get("data", [])
                if not data_rows or not labels:
                    continue
                values = _row_values(labels, data_rows[0])
                used = values.get("used")
                avail = values.get("avail")
                if used is not None and avail is not None:
                    # Charts may report absolute values (GB/MB) or percents;
                    # compute the used-percent ourselves so units do not matter.
                    reserved = values.get("reserved for root") or 0.0
                    denominator = used + avail + reserved
                    candidate = (used / denominator * 100.0) if denominator > 0 else None
                else:
                    candidates = [
                        v
                        for k, v in values.items()
                        if v is not None and k != "avail"
                    ]
                    candidate = max(candidates) if candidates else None
                if candidate is not None:
                    max_used = (
                        candidate if max_used is None else max(max_used, candidate)
                    )
            except Exception as exc:  # noqa: BLE001
                logger.warning("Netdata disk chart %s fetch failed: %s", chart, exc)

        result["disk_used_percent"] = max_used
    except Exception as exc:  # noqa: BLE001
        logger.warning("Netdata disk metrics fetch failed: %s", exc)

    return result


def netdata_info(base_url, timeout):
    """Check that the Netdata info endpoint is reachable and contains version."""
    url = f"{base_url}/api/v1/info"
    try:
        payload = _netdata_json(url, timeout)
        if isinstance(payload, dict) and "version" in payload:
            return True, f"Netdata {payload.get('version')}"
        return False, "Netdata info missing version"
    except Exception as exc:  # noqa: BLE001
        return False, f"Netdata info error: {_safe_message(str(exc))}"