Christian Manivong
209537a793
parse_power_supply also reads items of the legacy Power.PowerSupplies array, parse_ethernet_interface exposes the raw LinkStatus, and parse_network_protocol reports protocols that carry a Port without a ProtocolEnabled flag.
310 lines
13 KiB
Python
310 lines
13 KiB
Python
"""Chassis environment: legacy Thermal/Power, or ThermalSubsystem/PowerSubsystem/Sensors.
|
|
|
|
Services from Redfish 2020.4 on link ``ThermalSubsystem``, ``PowerSubsystem`` and
|
|
``Sensors`` from the Chassis and deprecate ``Thermal`` and ``Power``; older ones (such as
|
|
Supermicro X10) have only the latter. ``environment_sources`` picks per part, and both
|
|
paths end in the same shape, the parts of NAPALM's ``get_environment``:
|
|
|
|
- ``fans``: ``{name: {"status": bool}}``
|
|
- ``temperature``: ``{name: {"temperature": float, "is_alert": bool, "is_critical": bool}}``
|
|
- ``power``: ``{name: {"status": bool, "capacity": float, "output": float}}`` (-1.0 unknown)
|
|
|
|
plus ``voltages`` (``{name: {"voltage", "is_alert", "is_critical"}}``) and
|
|
``consumed_watts`` for the whole chassis. Absent parts and sensors without a reading are
|
|
left out.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Iterable, Mapping
|
|
|
|
from .common import link, number, present, status, text, unique_name
|
|
|
|
_SENSOR_TYPES_BY_UNITS = {"Cel": "Temperature", "V": "Voltage", "W": "Power", "A": "Current",
|
|
"RPM": "Rotational", "%": "Percent", "Hz": "Frequency"}
|
|
|
|
|
|
def environment_sources(chassis: Mapping[str, Any]) -> dict[str, Any]:
|
|
"""Which resources to read for a Chassis, preferring the newer ones per part.
|
|
|
|
Returns ``{"thermal": {"schema", "path"} | None, "power": ... | None,
|
|
"sensors": path | None}``.
|
|
"""
|
|
|
|
def pick(new: str, old: str) -> dict[str, str] | None:
|
|
for schema in (new, old):
|
|
path = link(chassis, schema)
|
|
if path is not None:
|
|
return {"schema": schema, "path": path}
|
|
return None
|
|
|
|
return {
|
|
"thermal": pick("ThermalSubsystem", "Thermal"),
|
|
"power": pick("PowerSubsystem", "Power"),
|
|
"sensors": link(chassis, "Sensors"),
|
|
}
|
|
|
|
|
|
def _ok(resource: Mapping[str, Any]) -> bool:
|
|
"""Working: enabled (or no state given) and healthy (or no health given)."""
|
|
state, health = status(resource)
|
|
return state in (None, "Enabled") and health in (None, "OK")
|
|
|
|
|
|
def _health_flags(resource: Mapping[str, Any]) -> tuple[bool, bool]:
|
|
_, health = status(resource)
|
|
return health in ("Warning", "Critical"), health == "Critical"
|
|
|
|
|
|
def _legacy_flags(item: Mapping[str, Any], reading: float) -> tuple[bool, bool]:
|
|
"""Alert/critical from Thermal/Power v1 threshold properties and health."""
|
|
alert, critical = _health_flags(item)
|
|
|
|
def beyond(suffix: str) -> bool:
|
|
upper = number(item.get(f"UpperThreshold{suffix}"))
|
|
lower = number(item.get(f"LowerThreshold{suffix}"))
|
|
return (upper is not None and reading >= upper) or (lower is not None and reading <= lower)
|
|
|
|
if beyond("Critical") or beyond("Fatal"):
|
|
return True, True
|
|
if beyond("NonCritical"):
|
|
alert = True
|
|
return alert, critical
|
|
|
|
|
|
def _sensor_flags(sensor: Mapping[str, Any], reading: float) -> tuple[bool, bool]:
|
|
"""Alert/critical from Sensor ``Thresholds`` and health."""
|
|
alert, critical = _health_flags(sensor)
|
|
thresholds = sensor.get("Thresholds") if isinstance(sensor.get("Thresholds"), Mapping) else {}
|
|
|
|
def beyond(level: str) -> bool:
|
|
def limit(name: str) -> float | None:
|
|
t = thresholds.get(name)
|
|
return number(t.get("Reading")) if isinstance(t, Mapping) else None
|
|
|
|
upper, lower = limit(f"Upper{level}"), limit(f"Lower{level}")
|
|
return (upper is not None and reading >= upper) or (lower is not None and reading <= lower)
|
|
|
|
if beyond("Critical") or beyond("Fatal"):
|
|
return True, True
|
|
if beyond("Caution"):
|
|
alert = True
|
|
return alert, critical
|
|
|
|
|
|
def _name(item: Mapping[str, Any], *keys: str) -> str:
|
|
for key in keys:
|
|
name = text(item.get(key))
|
|
if name:
|
|
return name
|
|
return text(item.get("MemberId")) or text(item.get("Id")) or "unknown"
|
|
|
|
|
|
def _members(resource: Mapping[str, Any], key: str) -> Iterable[tuple[int, Mapping[str, Any]]]:
|
|
items = resource.get(key)
|
|
for index, item in enumerate(items if isinstance(items, list) else []):
|
|
if isinstance(item, Mapping):
|
|
yield index, item
|
|
|
|
|
|
# --- legacy -------------------------------------------------------------------------------
|
|
|
|
|
|
def parse_thermal(thermal: Mapping[str, Any]) -> dict[str, Any]:
|
|
"""``Chassis/{id}/Thermal``: ``fans`` and ``temperature``."""
|
|
temperature: dict[str, Any] = {}
|
|
for index, item in _members(thermal, "Temperatures"):
|
|
reading = number(item.get("ReadingCelsius"))
|
|
if reading is None or not present(item):
|
|
continue
|
|
alert, critical = _legacy_flags(item, reading)
|
|
name = unique_name(temperature, _name(item, "Name"), item.get("MemberId", index))
|
|
temperature[name] = {"temperature": reading, "is_alert": alert, "is_critical": critical}
|
|
|
|
fans: dict[str, Any] = {}
|
|
for index, item in _members(thermal, "Fans"):
|
|
if not present(item):
|
|
continue
|
|
name = unique_name(fans, _name(item, "Name", "FanName"), item.get("MemberId", index))
|
|
fans[name] = {"status": _ok(item)}
|
|
return {"fans": fans, "temperature": temperature}
|
|
|
|
|
|
def parse_power(power: Mapping[str, Any]) -> dict[str, Any]:
|
|
"""``Chassis/{id}/Power``: ``power`` (supplies), ``voltages``, ``consumed_watts``."""
|
|
supplies: dict[str, Any] = {}
|
|
for index, item in _members(power, "PowerSupplies"):
|
|
psu = parse_power_supply(item)
|
|
if not psu["present"]:
|
|
continue
|
|
name = unique_name(supplies, psu["name"], item.get("MemberId", index))
|
|
supplies[name] = {"status": psu["status"], "capacity": psu["capacity"],
|
|
"output": psu["output"]}
|
|
|
|
voltages: dict[str, Any] = {}
|
|
for index, item in _members(power, "Voltages"):
|
|
reading = number(item.get("ReadingVolts"))
|
|
if reading is None or not present(item):
|
|
continue
|
|
alert, critical = _legacy_flags(item, reading)
|
|
name = unique_name(voltages, _name(item, "Name"), item.get("MemberId", index))
|
|
voltages[name] = {"voltage": reading, "is_alert": alert, "is_critical": critical}
|
|
|
|
consumed = None
|
|
for _, item in _members(power, "PowerControl"):
|
|
watts = number(item.get("PowerConsumedWatts"))
|
|
if watts is not None:
|
|
consumed = (consumed or 0.0) + watts
|
|
return {"power": supplies, "voltages": voltages, "consumed_watts": consumed}
|
|
|
|
|
|
# --- ThermalSubsystem, PowerSubsystem, Sensors --------------------------------------------
|
|
|
|
|
|
def parse_thermal_subsystem(subsystem: Mapping[str, Any]) -> dict[str, Any]:
|
|
"""``Chassis/{id}/ThermalSubsystem``: where the fans and metrics are."""
|
|
state, health = status(subsystem)
|
|
return {
|
|
"fans_path": link(subsystem, "Fans"),
|
|
"thermal_metrics_path": link(subsystem, "ThermalMetrics"),
|
|
"state": state,
|
|
"health": health,
|
|
}
|
|
|
|
|
|
def parse_power_subsystem(subsystem: Mapping[str, Any]) -> dict[str, Any]:
|
|
"""``Chassis/{id}/PowerSubsystem``: where the supplies are, total capacity."""
|
|
state, health = status(subsystem)
|
|
return {
|
|
"power_supplies_path": link(subsystem, "PowerSupplies"),
|
|
"capacity_watts": number(subsystem.get("CapacityWatts")),
|
|
"state": state,
|
|
"health": health,
|
|
}
|
|
|
|
|
|
def _excerpt(resource: Mapping[str, Any], key: str, field: str = "Reading") -> float | None:
|
|
value = resource.get(key)
|
|
return number(value.get(field)) if isinstance(value, Mapping) else None
|
|
|
|
|
|
def parse_fan(fan: Mapping[str, Any]) -> dict[str, Any]:
|
|
"""One member of ``ThermalSubsystem/Fans``."""
|
|
state, health = status(fan)
|
|
rpm = _excerpt(fan, "SpeedPercent", "SpeedRPM")
|
|
return {
|
|
"id": text(fan.get("Id")),
|
|
"name": _name(fan, "Name"),
|
|
"present": present(fan),
|
|
"status": _ok(fan),
|
|
"speed_rpm": int(rpm) if rpm is not None else None,
|
|
"speed_percent": _excerpt(fan, "SpeedPercent"),
|
|
"state": state,
|
|
"health": health,
|
|
}
|
|
|
|
|
|
def parse_power_supply(supply: Mapping[str, Any],
|
|
metrics: Mapping[str, Any] | None = None) -> dict[str, Any]:
|
|
"""One member of ``PowerSubsystem/PowerSupplies``, with its ``Metrics`` if read.
|
|
|
|
An item of the legacy ``Power.PowerSupplies`` array parses too (``MemberId`` as id,
|
|
``LastPowerOutputWatts`` as output).
|
|
"""
|
|
state, health = status(supply)
|
|
capacity = number(supply.get("PowerCapacityWatts"))
|
|
output = _excerpt(metrics, "OutputPowerWatts") if metrics else None
|
|
if output is None:
|
|
output = number(supply.get("LastPowerOutputWatts"))
|
|
return {
|
|
"id": text(supply.get("Id")) or text(supply.get("MemberId")),
|
|
"name": _name(supply, "Name"),
|
|
"present": present(supply),
|
|
"status": _ok(supply),
|
|
"capacity": capacity if capacity is not None else -1.0,
|
|
"output": output if output is not None else -1.0,
|
|
"manufacturer": text(supply.get("Manufacturer")),
|
|
"model": text(supply.get("Model")),
|
|
"serial_number": text(supply.get("SerialNumber")),
|
|
"part_number": text(supply.get("PartNumber")),
|
|
"firmware_version": text(supply.get("FirmwareVersion")),
|
|
"state": state,
|
|
"health": health,
|
|
}
|
|
|
|
|
|
def parse_sensor(sensor: Mapping[str, Any]) -> dict[str, Any]:
|
|
"""One member of ``Chassis/{id}/Sensors``.
|
|
|
|
``reading_type`` falls back to one derived from ``ReadingUnits`` (some sensors, such as
|
|
temperatures in the DMTF mockups, carry only units).
|
|
"""
|
|
state, health = status(sensor)
|
|
units = text(sensor.get("ReadingUnits"))
|
|
reading = number(sensor.get("Reading"))
|
|
alert, critical = _sensor_flags(sensor, reading) if reading is not None else (False, False)
|
|
return {
|
|
"id": text(sensor.get("Id")),
|
|
"name": _name(sensor, "Name"),
|
|
"reading_type": text(sensor.get("ReadingType")) or _SENSOR_TYPES_BY_UNITS.get(units or ""),
|
|
"reading": reading,
|
|
"units": units,
|
|
"physical_context": text(sensor.get("PhysicalContext")),
|
|
"electrical_context": text(sensor.get("ElectricalContext")),
|
|
"state": state,
|
|
"health": health,
|
|
"is_alert": alert,
|
|
"is_critical": critical,
|
|
}
|
|
|
|
|
|
def parse_thermal_metrics(metrics: Mapping[str, Any]) -> dict[str, Any]:
|
|
"""``ThermalSubsystem/ThermalMetrics``: temperatures without thresholds.
|
|
|
|
A fallback for services with a ThermalSubsystem but no Sensors collection.
|
|
"""
|
|
temperature: dict[str, Any] = {}
|
|
for index, item in _members(metrics, "TemperatureReadingsCelsius"):
|
|
reading = number(item.get("Reading"))
|
|
if reading is None:
|
|
continue
|
|
name = unique_name(temperature, _name(item, "DeviceName"), index)
|
|
temperature[name] = {"temperature": reading, "is_alert": False, "is_critical": False}
|
|
return {"temperature": temperature}
|
|
|
|
|
|
def build_environment(*, fans: Iterable[Mapping[str, Any]] = (),
|
|
power_supplies: Iterable[Mapping[str, Any]] = (),
|
|
sensors: Iterable[Mapping[str, Any]] = ()) -> dict[str, Any]:
|
|
"""Combine parsed fans, power supplies and sensors into the shape of the legacy parsers.
|
|
|
|
``consumed_watts`` comes from the chassis total power sensor
|
|
(``ReadingType`` Power, ``ElectricalContext`` Total, ``PhysicalContext`` Chassis).
|
|
"""
|
|
env: dict[str, Any] = {"fans": {}, "temperature": {}, "power": {}, "voltages": {},
|
|
"consumed_watts": None}
|
|
for fan in fans:
|
|
if fan["present"]:
|
|
name = unique_name(env["fans"], fan["name"], fan["id"])
|
|
env["fans"][name] = {"status": fan["status"]}
|
|
for psu in power_supplies:
|
|
if psu["present"]:
|
|
name = unique_name(env["power"], psu["name"], psu["id"])
|
|
env["power"][name] = {"status": psu["status"], "capacity": psu["capacity"],
|
|
"output": psu["output"]}
|
|
for sensor in sensors:
|
|
reading = sensor["reading"]
|
|
if reading is None:
|
|
continue
|
|
flags = {"is_alert": sensor["is_alert"], "is_critical": sensor["is_critical"]}
|
|
if sensor["reading_type"] == "Temperature":
|
|
name = unique_name(env["temperature"], sensor["name"], sensor["id"])
|
|
env["temperature"][name] = {"temperature": reading, **flags}
|
|
elif sensor["reading_type"] == "Voltage":
|
|
name = unique_name(env["voltages"], sensor["name"], sensor["id"])
|
|
env["voltages"][name] = {"voltage": reading, **flags}
|
|
elif (sensor["reading_type"] == "Power" and sensor["electrical_context"] == "Total"
|
|
and sensor["physical_context"] == "Chassis" and env["consumed_watts"] is None):
|
|
env["consumed_watts"] = reading
|
|
return env
|