Christian Manivong
adf152ee12
First run against real hardware over Redfish: an X10DRL-i (BMC 3.91) with
SFT-OOB-LIC. Its Redfish tree is recorded under
tests/fixtures/supermicro-x10drl-i-3.91/redfish, scrubbed.
- A resource refused for want of a further license (403 OemLicenseNotPassed)
no longer fails the getter. Storage then falls back to SimpleStorage. The
firmware list falls back to the BMC and BIOS versions when UpdateService
is withheld.
- get_device_warnings reports bmc_license_partial with the licenses and the
withheld parts (drives, firmware inventory).
- A placeholder system model ("Super Server") is replaced by the board
(PartNumber, else the chassis model).
663 lines
28 KiB
Python
663 lines
28 KiB
Python
"""NAPALM driver for server BMCs over DMTF Redfish.
|
|
|
|
The device is the BMC. NAPALM's getters describe the BMC (its ports, users, services);
|
|
the ``ManagementControllerDriver`` methods describe and control the server it manages.
|
|
``reboot_host`` restarts the BMC (``Manager.Reset``), never the server.
|
|
|
|
Optional args
|
|
-------------
|
|
port : int
|
|
HTTPS port (default 443).
|
|
verify_ssl : bool or str
|
|
``True`` (default) verifies the certificate against the CA bundle of ``requests``,
|
|
a path verifies against that CA file, ``False`` turns verification off (self-signed
|
|
BMC certificates are common).
|
|
auth : str
|
|
``auto`` (default: session, basic auth if the service cannot create sessions),
|
|
``session`` or ``basic``.
|
|
retries : int
|
|
Extra attempts after a connection error (default 1).
|
|
|
|
The managed server is the first member of ``Systems``; its BMC is the first manager in
|
|
its ``Links.ManagedBy`` and its chassis the first in its ``Links.Chassis`` (the first
|
|
member of ``Managers``/``Chassis`` when the links are missing).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from datetime import datetime, timezone
|
|
from typing import Any
|
|
|
|
from napalm.base.exceptions import ConnectionException
|
|
from napalm_device_types import HealthMetricsMixin, ManagementControllerDriver
|
|
from napalm_device_types.base import FingerprintRule, PortSpec
|
|
from napalm_device_types.models import POWER_ACTIONS
|
|
|
|
from napalm_redfish.parsers import (
|
|
build_environment,
|
|
environment_sources,
|
|
parse_account,
|
|
parse_action_info,
|
|
parse_drive,
|
|
parse_ethernet_interface,
|
|
parse_fan,
|
|
parse_firmware,
|
|
parse_log_entry,
|
|
parse_log_service,
|
|
parse_manager,
|
|
parse_memory,
|
|
parse_network_protocol,
|
|
parse_power,
|
|
parse_power_state,
|
|
parse_power_subsystem,
|
|
parse_power_supply,
|
|
parse_processor,
|
|
parse_reset_action,
|
|
parse_sensor,
|
|
parse_service_root,
|
|
parse_simple_storage,
|
|
parse_storage,
|
|
parse_system,
|
|
parse_thermal,
|
|
parse_thermal_metrics,
|
|
parse_thermal_subsystem,
|
|
to_napalm_users,
|
|
)
|
|
from napalm_redfish.parsers.common import link, link_list, odata_id
|
|
from napalm_redfish.transport import (
|
|
RedfishError,
|
|
RedfishFeatureLicenseError,
|
|
RedfishLicenseError,
|
|
RedfishNotFound,
|
|
RedfishTransport,
|
|
)
|
|
|
|
LOG = logging.getLogger(__name__)
|
|
|
|
_JsonDict = dict[str, Any]
|
|
|
|
#: ``get_management_protocols`` key -> ManagerNetworkProtocol property.
|
|
_PROTOCOLS = {
|
|
"ipmi": "IPMI",
|
|
"ssh": "SSH",
|
|
"http": "HTTP",
|
|
"https": "HTTPS",
|
|
"snmp": "SNMP",
|
|
"kvm": "KVMIP",
|
|
"virtual_media": "VirtualMedia",
|
|
}
|
|
_SEVERITIES = ("OK", "Warning", "Critical")
|
|
#: The parts a BMC may withhold for want of a license, in the order they are named.
|
|
_WITHHELD_ORDER = ("drives", "firmware inventory")
|
|
#: Models a system reports when it has none of its own. Supermicro's Redfish calls every
|
|
#: system "Super Server"; the board is the system's PartNumber and the chassis model.
|
|
_PLACEHOLDER_MODELS = frozenset({"super server", "default string", "to be filled by o.e.m."})
|
|
|
|
|
|
def _timestamp(created: str | None) -> float | None:
|
|
"""Unix time of a Redfish ``Created`` value; a time without offset counts as UTC."""
|
|
if not created:
|
|
return None
|
|
try:
|
|
when = datetime.fromisoformat(created.replace("Z", "+00:00"))
|
|
except ValueError:
|
|
return None
|
|
if when.tzinfo is None:
|
|
when = when.replace(tzinfo=timezone.utc)
|
|
return when.timestamp()
|
|
|
|
|
|
def _status(item: _JsonDict) -> str:
|
|
"""Role ``status``: ``Absent`` for an empty slot, else the Redfish health."""
|
|
if not item.get("present", True):
|
|
return "Absent"
|
|
return item.get("health") or ""
|
|
|
|
|
|
class RedfishDriver(HealthMetricsMixin, ManagementControllerDriver):
|
|
"""NAPALM driver for Redfish BMCs (Supermicro, iDRAC, iLO, XCC and others).
|
|
|
|
``HealthMetricsMixin``: the BMC's own load, memory and interfaces over SNMP (UCD-MIB),
|
|
where the BMC runs net-snmp. The managed server's sensors come from ``get_environment``.
|
|
"""
|
|
|
|
platform = "redfish"
|
|
DRIVER_NAME = "redfish"
|
|
#: Every Redfish service answers ``GET /redfish/v1/`` without credentials. The port
|
|
#: itself is worth nothing: every switch with a web UI answers on 443.
|
|
PORT_SPECS = [PortSpec("https", 443, ("/redfish/v1/",), weight=0.0)]
|
|
#: Rules in one group are alternatives and count once towards the maximum
|
|
#: (napalm-device-types' ``rules_max_weight``).
|
|
HTTP_FINGERPRINT = [
|
|
# The service root, or the error body a service refuses it with (an unlicensed
|
|
# Supermicro X10 answers 403 with a Redfish error).
|
|
FingerprintRule("redfishversion", 10.0, group="redfish_answer"),
|
|
FingerprintRule("@message.extendedinfo", 8.0, group="redfish_answer"),
|
|
FingerprintRule("#serviceroot.", 5.0),
|
|
# Landing pages of the BMC web UIs: Supermicro X9-X11 (ATEN firmware), iDRAC, iLO.
|
|
FingerprintRule("aten international", 6.0, group="landing_page"),
|
|
FingerprintRule("idrac", 6.0, group="landing_page"),
|
|
FingerprintRule("integrated lights-out", 6.0, group="landing_page"),
|
|
]
|
|
#: A BMC restart takes a few minutes before Redfish answers again.
|
|
REBOOT_SETTLE_SECONDS = 180
|
|
|
|
def __init__(
|
|
self,
|
|
hostname: str,
|
|
username: str,
|
|
password: str,
|
|
timeout: int = 60,
|
|
optional_args: _JsonDict | None = None,
|
|
) -> None:
|
|
self.hostname = hostname
|
|
self.username = username
|
|
self.password = password
|
|
self.timeout = timeout
|
|
self.optional_args: _JsonDict = optional_args or {}
|
|
self.transport = RedfishTransport(
|
|
hostname,
|
|
username,
|
|
password,
|
|
port=self.optional_args.get("port"),
|
|
verify=self.optional_args.get("verify_ssl", True),
|
|
auth=self.optional_args.get("auth", "auto"),
|
|
timeout=timeout,
|
|
retries=self.optional_args.get("retries", 1),
|
|
)
|
|
self._cache: _JsonDict = {}
|
|
self._license_refused: RedfishLicenseError | None = None
|
|
#: Parts the BMC withholds for want of a license, with the licenses it names.
|
|
self._withheld: dict[str, str] = {}
|
|
|
|
# --- connection ----------------------------------------------------------------------
|
|
|
|
def open(self) -> None:
|
|
"""Connect. A BMC that refuses Redfish for want of a license is reachable all the
|
|
same: ``open`` succeeds, ``get_facts`` returns what is known without Redfish, and
|
|
``get_device_warnings`` says which license is missing. Every other getter raises.
|
|
"""
|
|
self._license_refused = None
|
|
self._withheld = {}
|
|
try:
|
|
self.transport.open()
|
|
except RedfishLicenseError as exc:
|
|
self._license_refused = exc
|
|
except RedfishError as exc:
|
|
raise ConnectionException(f"Cannot connect to Redfish at {self.hostname}: {exc}") \
|
|
from exc
|
|
|
|
def close(self) -> None:
|
|
self.transport.close()
|
|
self._cache = {}
|
|
|
|
def is_alive(self) -> _JsonDict:
|
|
if self._license_refused is not None:
|
|
return {"is_alive": True}
|
|
try:
|
|
self.transport.get("/redfish/v1/")
|
|
except Exception:
|
|
return {"is_alive": False}
|
|
return {"is_alive": True}
|
|
|
|
# --- discovery (cached per connection) -------------------------------------------------
|
|
|
|
def _cached(self, key: str, load):
|
|
if key not in self._cache:
|
|
self._cache[key] = load()
|
|
return self._cache[key]
|
|
|
|
def _root(self) -> _JsonDict:
|
|
if self._license_refused is not None:
|
|
raise ConnectionException(f"{self.hostname}: {self._license_refused}")
|
|
return self._cached("root", lambda: parse_service_root(self.transport.root))
|
|
|
|
def _first(self, collection: str) -> _JsonDict:
|
|
path = self._root()["links"].get(collection)
|
|
if path is None:
|
|
raise RedfishError(f"the service root links no {collection}")
|
|
members = self.transport.members(path, limit=1)
|
|
if not members:
|
|
raise RedfishError(f"{path} has no members")
|
|
return members[0]
|
|
|
|
def _system_raw(self) -> _JsonDict:
|
|
return self._cached("system", lambda: self._first("Systems"))
|
|
|
|
def _system_path(self) -> str:
|
|
return odata_id(self._system_raw()) or ""
|
|
|
|
def _linked_or_first(self, key: str, links_key: str, collection: str) -> _JsonDict:
|
|
def load() -> _JsonDict:
|
|
linked = link_list(self._system_raw(), "Links", links_key)
|
|
if linked:
|
|
return self.transport.get(linked[0])
|
|
return self._first(collection)
|
|
|
|
return self._cached(key, load)
|
|
|
|
def _manager_raw(self) -> _JsonDict:
|
|
return self._linked_or_first("manager", "ManagedBy", "Managers")
|
|
|
|
def _chassis_raw(self) -> _JsonDict:
|
|
return self._linked_or_first("chassis", "Chassis", "Chassis")
|
|
|
|
def _server_model(self) -> str:
|
|
"""The system's model; for a placeholder, the board (PartNumber or chassis model)."""
|
|
raw = self._system_raw()
|
|
model = (parse_system(raw)["model"] or "").strip()
|
|
if model.lower() not in _PLACEHOLDER_MODELS:
|
|
return model
|
|
board = raw.get("PartNumber") or self._chassis_raw().get("Model") or ""
|
|
return str(board).strip() or model
|
|
|
|
def _members_of(self, resource: _JsonDict, key: str) -> list[_JsonDict]:
|
|
"""Members of the collection ``resource[key]`` links; [] if not linked."""
|
|
path = link(resource, key)
|
|
return self.transport.members(path) if path else []
|
|
|
|
def _service(self, name: str) -> _JsonDict | None:
|
|
path = self._root()["links"].get(name)
|
|
return self.transport.get_optional(path) if path else None
|
|
|
|
def _network_protocol(self) -> _JsonDict:
|
|
raw = self.transport.follow(self._manager_raw(), "NetworkProtocol")
|
|
return parse_network_protocol(raw or {})
|
|
|
|
def _manager_interfaces(self) -> list[_JsonDict]:
|
|
return [parse_ethernet_interface(raw)
|
|
for raw in self._members_of(self._manager_raw(), "EthernetInterfaces")]
|
|
|
|
# --- NAPALM getters ----------------------------------------------------------------
|
|
|
|
def get_facts(self) -> _JsonDict:
|
|
if self._license_refused is not None:
|
|
return {"uptime": -1.0, "vendor": "", "model": "", "hostname": self.hostname,
|
|
"fqdn": self.hostname, "os_version": "", "serial_number": "",
|
|
"interface_list": []}
|
|
system = parse_system(self._system_raw())
|
|
manager = parse_manager(self._manager_raw())
|
|
interfaces = self._manager_interfaces()
|
|
protocol = self._network_protocol()
|
|
hostname = (protocol["hostname"]
|
|
or next((i["hostname"] for i in interfaces if i["hostname"]), None)
|
|
or self.hostname)
|
|
fqdn = (protocol["fqdn"]
|
|
or next((i["fqdn"] for i in interfaces if i["fqdn"]), None)
|
|
or hostname)
|
|
return {
|
|
"uptime": -1.0, # Redfish has no uptime
|
|
"vendor": system["manufacturer"] or "",
|
|
"model": self._server_model(),
|
|
"hostname": hostname,
|
|
"fqdn": fqdn,
|
|
"os_version": manager["firmware_version"] or "",
|
|
"serial_number": system["serial_number"] or "",
|
|
"interface_list": sorted(i["name"] for i in interfaces if i["name"]),
|
|
}
|
|
|
|
def get_interfaces(self) -> _JsonDict:
|
|
return {i["name"]: i["interface"] for i in self._manager_interfaces() if i["name"]}
|
|
|
|
def get_interfaces_ip(self) -> _JsonDict:
|
|
result = {}
|
|
for iface in self._manager_interfaces():
|
|
families = {f: iface[f] for f in ("ipv4", "ipv6") if iface[f]}
|
|
if iface["name"] and families:
|
|
result[iface["name"]] = families
|
|
return result
|
|
|
|
def get_environment(self) -> _JsonDict:
|
|
"""Fans, temperatures and power supplies of the chassis; ``cpu`` and ``memory`` are
|
|
empty, since a BMC does not know the host OS's load."""
|
|
sources = environment_sources(self._chassis_raw())
|
|
fans: _JsonDict = {}
|
|
temperature: _JsonDict = {}
|
|
power: _JsonDict = {}
|
|
|
|
thermal = sources["thermal"]
|
|
if thermal and thermal["schema"] == "Thermal":
|
|
legacy = parse_thermal(self.transport.get(thermal["path"]))
|
|
fans, temperature = legacy["fans"], legacy["temperature"]
|
|
elif thermal:
|
|
subsystem = parse_thermal_subsystem(self.transport.get(thermal["path"]))
|
|
parsed_fans = [parse_fan(raw) for raw in self.transport.members(
|
|
subsystem["fans_path"])] if subsystem["fans_path"] else []
|
|
if sources["sensors"]:
|
|
sensors = [parse_sensor(raw)
|
|
for raw in self.transport.members(sources["sensors"])]
|
|
temperature = build_environment(sensors=sensors)["temperature"]
|
|
elif subsystem["thermal_metrics_path"]:
|
|
metrics = self.transport.get_optional(subsystem["thermal_metrics_path"])
|
|
temperature = parse_thermal_metrics(metrics or {})["temperature"]
|
|
fans = build_environment(fans=parsed_fans)["fans"]
|
|
|
|
supply = sources["power"]
|
|
if supply and supply["schema"] == "Power":
|
|
power = parse_power(self.transport.get(supply["path"]))["power"]
|
|
elif supply:
|
|
psus = [parse_power_supply(raw, self.transport.follow(raw, "Metrics"))
|
|
for raw in self._power_supplies_raw(supply["path"])]
|
|
power = build_environment(power_supplies=psus)["power"]
|
|
|
|
return {"fans": fans, "temperature": temperature, "power": power,
|
|
"cpu": {}, "memory": {}}
|
|
|
|
def _power_supplies_raw(self, subsystem_path: str) -> list[_JsonDict]:
|
|
subsystem = parse_power_subsystem(self.transport.get(subsystem_path))
|
|
path = subsystem["power_supplies_path"]
|
|
return self.transport.members(path) if path else []
|
|
|
|
def get_users(self) -> _JsonDict:
|
|
service = self._service("AccountService")
|
|
if service is None:
|
|
return {}
|
|
return to_napalm_users(parse_account(raw) for raw in self._members_of(service,
|
|
"Accounts"))
|
|
|
|
def get_ntp_servers(self) -> _JsonDict:
|
|
return {server: {} for server in self._network_protocol()["ntp"]["servers"]}
|
|
|
|
def get_snmp_information(self) -> _JsonDict:
|
|
"""SNMP state from ManagerNetworkProtocol. Redfish hides community strings, so the
|
|
``community`` keys are the communities' names; contact and location are not in
|
|
Redfish."""
|
|
snmp = self._network_protocol()["snmp"]
|
|
communities = {}
|
|
if snmp["enabled"]:
|
|
for community in snmp["communities"]:
|
|
if community["name"]:
|
|
mode = "rw" if community["access"] == "Full" else "ro"
|
|
communities[community["name"]] = {"acl": "", "mode": mode}
|
|
return {"chassis_id": "", "community": communities, "contact": "", "location": ""}
|
|
|
|
def get_device_warnings(self) -> list[_JsonDict]:
|
|
"""Raw signals for netOrk's warning catalog; presentation is netOrk's.
|
|
|
|
``bmc_license_required`` when the BMC refuses Redfish altogether;
|
|
``bmc_license_partial`` when it answers but withholds parts (drives, the firmware
|
|
inventory) for want of a license. The parts are probed here, so the warning does
|
|
not depend on which getters ran before.
|
|
"""
|
|
refused = self._license_refused
|
|
if refused is not None:
|
|
return [{"code": "bmc_license_required",
|
|
"meta": {"licenses": refused.licenses, "message": refused.detail}}]
|
|
self._probe_withheld()
|
|
if not self._withheld:
|
|
return []
|
|
licenses = sorted({name for names in self._withheld.values() for name in names.split()})
|
|
parts = [part for part in _WITHHELD_ORDER if part in self._withheld]
|
|
return [{"code": "bmc_license_partial",
|
|
"meta": {"licenses": " ".join(licenses), "parts": parts}}]
|
|
|
|
def _withhold(self, part: str, exc: RedfishFeatureLicenseError) -> None:
|
|
LOG.info("%s: %s withheld by the BMC: %s", self.hostname, part, exc)
|
|
self._withheld[part] = exc.licenses
|
|
|
|
def _probe_withheld(self) -> None:
|
|
"""Ask for the license-gated resources once, so ``_withheld`` is complete."""
|
|
storage = link(self._system_raw(), "Storage")
|
|
if storage and "drives" not in self._withheld:
|
|
try:
|
|
self.transport.get(storage)
|
|
except RedfishFeatureLicenseError as exc:
|
|
self._withhold("drives", exc)
|
|
except RedfishError:
|
|
pass
|
|
if "firmware inventory" not in self._withheld:
|
|
self._update_service()
|
|
|
|
# --- role: power -------------------------------------------------------------------
|
|
|
|
def _accepted_actions(self, system: _JsonDict) -> list[str]:
|
|
"""The PowerAction values the BMC accepts for ``system`` in its current state."""
|
|
reset = parse_reset_action(system)
|
|
if reset["target"] is None:
|
|
return []
|
|
allowed = reset["allowed"]
|
|
if allowed is None and reset["action_info"]:
|
|
info = self.transport.get_optional(reset["action_info"])
|
|
allowed = parse_action_info(info) if info else None
|
|
if allowed is None:
|
|
# Not advertised: assume the standard values; the BMC rejects what it lacks.
|
|
allowed = list(POWER_ACTIONS)
|
|
actions = [a for a in POWER_ACTIONS if a in allowed]
|
|
state = parse_power_state(system)
|
|
if state == "Off":
|
|
return [a for a in actions if a == "On"]
|
|
if state == "On":
|
|
return [a for a in actions if a != "On"]
|
|
return actions
|
|
|
|
def get_power_state(self) -> _JsonDict:
|
|
system = self.transport.get(self._system_path())
|
|
return {"state": parse_power_state(system) or "",
|
|
"actions": self._accepted_actions(system)}
|
|
|
|
def set_power_state(self, action: str) -> None:
|
|
if action not in POWER_ACTIONS:
|
|
raise ValueError(f"{action!r} is not one of {', '.join(POWER_ACTIONS)}")
|
|
system = self.transport.get(self._system_path())
|
|
target = parse_reset_action(system)["target"]
|
|
if target is None:
|
|
raise RuntimeError(f"{self.hostname}: the system offers no reset action")
|
|
accepted = self._accepted_actions(system)
|
|
if action not in accepted:
|
|
raise RuntimeError(
|
|
f"{self.hostname}: {action} not accepted while the server is "
|
|
f"{parse_power_state(system) or 'in an unknown state'} "
|
|
f"(accepted: {', '.join(accepted) or 'none'})")
|
|
try:
|
|
self.transport.post(target, {"ResetType": action})
|
|
except RedfishError as exc:
|
|
raise RuntimeError(f"{self.hostname}: {action} refused: {exc}") from exc
|
|
|
|
def reboot_host(self) -> None:
|
|
"""Restart the BMC (``Manager.Reset``, graceful when offered). The server keeps
|
|
running; the Redfish session ends with the restart."""
|
|
manager = self.transport.get(odata_id(self._manager_raw()) or "")
|
|
reset = parse_reset_action(manager, "#Manager.Reset")
|
|
if reset["target"] is None:
|
|
raise RuntimeError(f"{self.hostname}: the BMC offers no reset action")
|
|
body: _JsonDict = {}
|
|
if reset["allowed"] is not None:
|
|
reset_type = next((t for t in ("GracefulRestart", "ForceRestart")
|
|
if t in reset["allowed"]), None)
|
|
if reset_type is None:
|
|
raise RuntimeError(f"{self.hostname}: BMC reset types "
|
|
f"{reset['allowed']} include no restart")
|
|
body["ResetType"] = reset_type
|
|
try:
|
|
self.transport.post(reset["target"], body)
|
|
except RedfishError as exc:
|
|
raise RuntimeError(f"{self.hostname}: BMC reset refused: {exc}") from exc
|
|
|
|
# --- role: inventory -----------------------------------------------------------------
|
|
|
|
def get_hardware_inventory(self) -> _JsonDict:
|
|
raw = self._system_raw()
|
|
system = parse_system(raw)
|
|
return {
|
|
"manufacturer": system["manufacturer"] or "",
|
|
"model": self._server_model(),
|
|
"serial_number": system["serial_number"] or "",
|
|
"system_uuid": (system["uuid"] or "").lower(),
|
|
"cpus": self._cpus(raw),
|
|
"memory": self._dimms(raw),
|
|
"drives": self._drives(raw),
|
|
"power_supplies": self._psus(),
|
|
"nics": self._nics(raw),
|
|
}
|
|
|
|
def _cpus(self, system: _JsonDict) -> list[_JsonDict]:
|
|
cpus = []
|
|
for raw in self._members_of(system, "Processors"):
|
|
cpu = parse_processor(raw)
|
|
if cpu["processor_type"] not in (None, "CPU"):
|
|
continue # GPUs, FPGAs, accelerators
|
|
cpus.append({"socket": cpu["socket"] or cpu["id"] or "", "model": cpu["model"] or "",
|
|
"cores": cpu["cores"], "threads": cpu["threads"],
|
|
"status": _status(cpu)})
|
|
return cpus
|
|
|
|
def _dimms(self, system: _JsonDict) -> list[_JsonDict]:
|
|
dimms = []
|
|
for raw in self._members_of(system, "Memory"):
|
|
dimm = parse_memory(raw)
|
|
size = dimm["capacity_mib"]
|
|
dimms.append({
|
|
"slot": dimm["location"] or dimm["id"] or "",
|
|
"size_bytes": size * 1024 * 1024 if size else None,
|
|
"type": dimm["memory_device_type"] or "",
|
|
"speed_mts": dimm["speed_mhz"], # Redfish: MHz or MT/s "as appropriate"
|
|
"part_number": dimm["part_number"] or "",
|
|
"serial_number": dimm["serial_number"] or "",
|
|
"status": _status(dimm),
|
|
})
|
|
return dimms
|
|
|
|
def _storage_drives(self, system: _JsonDict) -> list[_JsonDict] | None:
|
|
"""Drives from ``Storage``; ``None`` when it is not linked or withheld."""
|
|
if not link(system, "Storage"):
|
|
return None
|
|
try:
|
|
storages = self._members_of(system, "Storage")
|
|
except RedfishFeatureLicenseError as exc:
|
|
self._withhold("drives", exc)
|
|
return None
|
|
parsed: list[_JsonDict] = []
|
|
seen: set[str] = set()
|
|
for storage in storages:
|
|
for path in parse_storage(storage)["drive_paths"]:
|
|
if path not in seen:
|
|
seen.add(path)
|
|
drive = self.transport.get_optional(path)
|
|
if drive is not None:
|
|
parsed.append(parse_drive(drive))
|
|
return parsed
|
|
|
|
def _drives(self, system: _JsonDict) -> list[_JsonDict]:
|
|
parsed = self._storage_drives(system)
|
|
if parsed is None:
|
|
# Older services, and a Supermicro X10 that withholds Storage without DCMS.
|
|
parsed = []
|
|
for storage in self._members_of(system, "SimpleStorage"):
|
|
parsed.extend(parse_simple_storage(storage)["drives"])
|
|
return [{
|
|
"slot": d["name"] or d["id"] or "",
|
|
"model": d["model"] or "",
|
|
"serial_number": d["serial_number"] or "",
|
|
"size_bytes": d["capacity_bytes"],
|
|
"media": d["media_type"] or "",
|
|
"protocol": d["protocol"] or "",
|
|
"status": _status(d),
|
|
} for d in parsed]
|
|
|
|
def _psus(self) -> list[_JsonDict]:
|
|
source = environment_sources(self._chassis_raw())["power"]
|
|
if source is None:
|
|
return []
|
|
if source["schema"] == "Power":
|
|
items = self.transport.get(source["path"]).get("PowerSupplies") or []
|
|
else:
|
|
items = self._power_supplies_raw(source["path"])
|
|
psus = []
|
|
for raw in items:
|
|
if not isinstance(raw, dict):
|
|
continue
|
|
psu = parse_power_supply(raw)
|
|
psus.append({
|
|
"slot": psu["name"],
|
|
"model": psu["model"] or "",
|
|
"serial_number": psu["serial_number"] or "",
|
|
"capacity_watts": int(psu["capacity"]) if psu["capacity"] >= 0 else None,
|
|
"status": _status(psu),
|
|
})
|
|
return psus
|
|
|
|
def _nics(self, system: _JsonDict) -> list[_JsonDict]:
|
|
nics = []
|
|
for raw in self._members_of(system, "EthernetInterfaces"):
|
|
nic = parse_ethernet_interface(raw)
|
|
speed = nic["interface"]["speed"]
|
|
link_up = {"LinkUp": True, "LinkDown": False, "NoLink": False}.get(
|
|
nic["link_status"] or "")
|
|
nics.append({
|
|
"name": nic["name"] or "",
|
|
"mac_address": nic["interface"]["mac_address"].lower(),
|
|
"speed_mbps": int(speed) if speed >= 0 else None,
|
|
"link_up": link_up,
|
|
})
|
|
return nics
|
|
|
|
# --- role: event log, firmware, protocols --------------------------------------------
|
|
|
|
def get_event_log(self, limit: int = 500) -> list[_JsonDict]:
|
|
"""Entries of every log service of the system and of the BMC, newest first.
|
|
|
|
Where the SEL lives varies by vendor, so all of them are read; entries without a
|
|
time sort last.
|
|
"""
|
|
entries: list[_JsonDict] = []
|
|
for owner in (self._system_raw(), self._manager_raw()):
|
|
try:
|
|
services = self._members_of(owner, "LogServices")
|
|
except RedfishNotFound:
|
|
continue
|
|
for raw_service in services:
|
|
service = parse_log_service(raw_service)
|
|
if not service["entries_path"]:
|
|
continue
|
|
try:
|
|
raw_entries = self.transport.members(service["entries_path"])
|
|
except RedfishNotFound:
|
|
LOG.warning("%s: log %s has no entries collection", self.hostname,
|
|
service["id"])
|
|
continue
|
|
for raw in raw_entries:
|
|
entry = parse_log_entry(raw)
|
|
severity = entry["severity"]
|
|
entries.append({
|
|
"id": odata_id(raw) or f"{service['id']}/{entry['id']}",
|
|
"timestamp": _timestamp(entry["created"]),
|
|
"severity": severity if severity in _SEVERITIES else "OK",
|
|
"message": entry["message"] or "",
|
|
})
|
|
entries.sort(key=lambda e: (e["timestamp"] is not None, e["timestamp"] or 0.0),
|
|
reverse=True)
|
|
return entries[:limit]
|
|
|
|
def _update_service(self) -> _JsonDict | None:
|
|
try:
|
|
return self._service("UpdateService")
|
|
except RedfishFeatureLicenseError as exc:
|
|
self._withhold("firmware inventory", exc)
|
|
return None
|
|
|
|
def get_firmware_inventory(self) -> list[_JsonDict]:
|
|
"""``UpdateService/FirmwareInventory``. Where the BMC withholds it for want of a
|
|
license, the BMC's own firmware and the BIOS, which the manager and the system
|
|
state themselves."""
|
|
service = self._update_service()
|
|
if service is None:
|
|
if "firmware inventory" not in self._withheld:
|
|
return []
|
|
bmc = parse_manager(self._manager_raw())["firmware_version"]
|
|
bios = self._system_raw().get("BiosVersion")
|
|
return [{"name": name, "version": version, "updateable": None}
|
|
for name, version in (("BMC", bmc), ("BIOS", bios)) if version]
|
|
items = [parse_firmware(raw) for raw in self._members_of(service, "FirmwareInventory")]
|
|
return [{"name": fw["name"] or fw["id"] or "", "version": fw["version"] or "",
|
|
"updateable": fw["updateable"]} for fw in items]
|
|
|
|
def get_management_protocols(self) -> _JsonDict:
|
|
"""What the BMC listens on. A protocol the BMC does not report counts as off."""
|
|
protocols = self._network_protocol()["protocols"]
|
|
result = {}
|
|
for key, name in _PROTOCOLS.items():
|
|
entry = protocols.get(name) or {}
|
|
result[key] = {"enabled": bool(entry.get("enabled")), "port": entry.get("port")}
|
|
return result
|