jbod-monitor/services/host_sensors.py

"""Host/chassis temperature collectors: IPMI (iDRAC) + CPU coretemp.

Used by the host-poller (separate process from the SAS poller). In-band IPMI
via ipmitool (/dev/ipmi0) gives environmental sensors (Inlet/Exhaust/…); the
kernel coretemp hwmon gives CPU package temps. IPMI is gated for serialization;
coretemp is a cheap sysfs read.
"""
import asyncio
import logging
import os
import re
from collections import Counter
from pathlib import Path

from services.hwgate import gate

logger = logging.getLogger(__name__)

HWMON = Path("/sys/class/hwmon")


async def read_ipmi_temps() -> list[dict]:
    """`ipmitool sdr type temperature` → [{name, temperature_c, status, kind}]."""
    try:
        async with gate():
            proc = await asyncio.create_subprocess_exec(
                "ipmitool", "sdr", "type", "temperature",
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
            )
            out, err = await proc.communicate()
    except FileNotFoundError:
        logger.warning("ipmitool not found")
        return []
    if proc.returncode != 0:
        logger.warning("ipmitool failed: %s", err.decode(errors="replace").strip())
        return []
    return _parse_ipmi(out.decode(errors="replace"))


def _parse_ipmi(text: str) -> list[dict]:
    # e.g.  "Inlet Temp       | 04h | ok  |  7.1 | 23 degrees C"
    #       "Temp             | 0Eh | ok  |  3.1 | 40 degrees C"  (CPU1)
    sensors: list[dict] = []
    for line in text.splitlines():
        parts = [p.strip() for p in line.split("|")]
        if len(parts) < 5:
            continue
        name, _sid, status, entity, reading = parts[:5]
        # Skip relative/derived sensors (e.g. "P1 Therm Margin") — they read in
        # "degrees C" but are a distance-to-throttle, not an absolute temp.
        if "margin" in name.lower():
            continue
        m = re.search(r"(-?\d+(?:\.\d+)?)\s*degrees?\s*C", reading, re.IGNORECASE)
        if not m:  # "no reading", "disabled", "ns", etc.
            continue
        temp = float(m.group(1))
        kind = "env"
        # Dell reports CPU temps as "Temp" under entity 3.x — relabel them.
        if name.lower() == "temp" and entity.startswith("3."):
            name = f"CPU {entity.split('.')[-1]}"
            kind = "cpu"
        sensors.append({
            "name": name,
            "temperature_c": temp,
            "status": status or "ok",
            "kind": kind,
        })
    return sensors


def read_hwmon_env() -> list[dict]:
    """Read labelled temps from extra hwmon chips as env sensors.

    For boxes without IPMI (e.g. Dell workstations), the BIOS SMM interface
    (`dell_smm`) exposes CPU/ambient/SODIMM/board temps. Chips are configurable
    via HOST_HWMON_CHIPS (comma list; default "dell_smm"); set to "" to disable.
    Names are "<chip> <label>", with the temp index appended when a label
    repeats within a chip (e.g. dell_smm's multiple "Other" sensors).
    """
    chips = [c.strip() for c in os.environ.get("HOST_HWMON_CHIPS", "dell_smm").split(",") if c.strip()]
    if not chips or not HWMON.is_dir():
        return []
    out: list[dict] = []
    for hw in sorted(HWMON.iterdir()):
        try:
            chip = (hw / "name").read_text().strip()
        except OSError:
            continue
        if chip not in chips:
            continue
        entries: list[tuple[str, str, int]] = []  # (idx, label, temp)
        for label_f in sorted(hw.glob("temp*_label")):
            try:
                label = label_f.read_text().strip()
            except OSError:
                continue
            input_f = label_f.with_name(label_f.name.replace("_label", "_input"))
            try:
                temp = round(int(input_f.read_text().strip()) / 1000)
            except (OSError, ValueError):
                continue
            idx = label_f.name[len("temp"):-len("_label")]
            entries.append((idx, label, temp))
        dup = {lbl for lbl, n in Counter(l for _, l, _ in entries).items() if n > 1}
        for idx, label, temp in entries:
            name = f"{chip} {label}" + (f" {idx}" if label in dup else "")
            out.append({"name": name, "temperature_c": temp, "status": "ok", "kind": "env"})
    return out


def read_coretemp() -> list[dict]:
    """CPU package temps from coretemp hwmon → [{name, temperature_c, kind}]."""
    cpus: list[dict] = []
    if not HWMON.is_dir():
        return cpus
    for hw in sorted(HWMON.iterdir()):
        try:
            if (hw / "name").read_text().strip() != "coretemp":
                continue
        except OSError:
            continue
        packages: list[dict] = []
        cores: list[int] = []
        for label_f in sorted(hw.glob("temp*_label")):
            try:
                label = label_f.read_text().strip()
            except OSError:
                continue
            input_f = label_f.with_name(label_f.name.replace("_label", "_input"))
            try:
                temp = round(int(input_f.read_text().strip()) / 1000)
            except (OSError, ValueError):
                continue
            low = label.lower()
            if low.startswith("package id"):
                pkg = label.split("id")[-1].strip()
                packages.append({"name": f"CPU {pkg}", "temperature_c": temp, "kind": "cpu"})
            elif low.startswith("core"):
                cores.append(temp)
        # Prefer the per-package sensor; older CPUs (no package sensor) fall back
        # to the hottest core as the CPU temp.
        if packages:
            cpus.extend(packages)
        elif cores:
            cpus.append({"name": "CPU", "temperature_c": max(cores), "kind": "cpu"})
    return cpus