Christian Manivong
09029ca0a4
Mechanical: ruff format, and ruff check --fix with the rule set every NAPALM driver uses (E, F, I, UP, B at line length 100, Python 3.10). Mapping and Iterable come from collections.abc, imports are sorted. What ruff cannot fix comes in the next commit.
180 lines
6.9 KiB
Python
180 lines
6.9 KiB
Python
"""NAPALM driver for a BMC that takes the best of Redfish and IPMI, chosen by itself.
|
|
|
|
``redfish`` and ``ipmi`` make the operator choose, and the right choice changes with the
|
|
BMC's license: a Supermicro X10 refuses Redfish without one, answers most of it with
|
|
SFT-OOB-LIC, and all of it with DCMS. This driver asks Redfish first and IPMI where Redfish
|
|
does not answer:
|
|
|
|
* A BMC that refuses Redfish altogether (no license) is read over IPMI entirely.
|
|
* A getter that fails over Redfish (an error, a resource the BMC withholds) is asked over
|
|
IPMI, where IPMI has it. IPMI is opened only then.
|
|
* Power control goes over Redfish when it answers, else over IPMI. A refusal by the BMC is
|
|
an answer, not a failure, and is not retried over IPMI.
|
|
* The BMC itself is reset over Redfish only (``Manager.Reset``); IPMI's cold reset is not
|
|
offered, as in the ``ipmi`` driver.
|
|
|
|
Both protocols use the same account. ``optional_args``: everything ``RedfishDriver``
|
|
takes, plus ``ipmi_port`` (623).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Any
|
|
|
|
from napalm.base.exceptions import ConnectionException
|
|
from napalm_device_types import HealthMetricsMixin, LicenseMixin, ManagementControllerDriver
|
|
|
|
from napalm_redfish.driver import BMC_HTTP_FINGERPRINT, BMC_PORT_SPECS, RedfishDriver
|
|
from napalm_redfish.ipmi import IpmiDriver
|
|
from napalm_redfish.transport import RedfishError
|
|
|
|
LOG = logging.getLogger(__name__)
|
|
|
|
_JsonDict = dict[str, Any]
|
|
_IPMI_PORT = 623
|
|
|
|
|
|
class BmcDriver(HealthMetricsMixin, LicenseMixin, ManagementControllerDriver):
|
|
"""One BMC over Redfish and IPMI: Redfish first, IPMI where Redfish does not answer."""
|
|
|
|
platform = "bmc"
|
|
DRIVER_NAME = "bmc"
|
|
default_port = 443
|
|
#: Discovery suggests this driver for a BMC it finds, not ``redfish``.
|
|
PORT_SPECS = BMC_PORT_SPECS
|
|
HTTP_FINGERPRINT = BMC_HTTP_FINGERPRINT
|
|
#: A BMC restart takes a few minutes before Redfish answers again.
|
|
REBOOT_SETTLE_SECONDS = RedfishDriver.REBOOT_SETTLE_SECONDS
|
|
|
|
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 = dict(optional_args or {})
|
|
ipmi_port = int(self.optional_args.pop("ipmi_port", None) or _IPMI_PORT)
|
|
self.ipmi_port = ipmi_port
|
|
self._redfish = RedfishDriver(hostname, username, password, timeout, self.optional_args)
|
|
self._ipmi = IpmiDriver(hostname, username, password, timeout, {"port": ipmi_port})
|
|
self._redfish_ok = False
|
|
self._ipmi_open = False
|
|
|
|
# --- connection ----------------------------------------------------------------------
|
|
|
|
def open(self) -> None:
|
|
"""Connect over Redfish; a BMC that refuses it or does not answer it is opened over
|
|
IPMI instead. Fails only when neither answers."""
|
|
try:
|
|
self._redfish.open()
|
|
except ConnectionException as exc:
|
|
LOG.info("%s: no Redfish (%s); reading it over IPMI", self.hostname, exc)
|
|
self._redfish_ok = False
|
|
else:
|
|
self._redfish_ok = self._redfish._license_refused is None
|
|
if not self._redfish_ok:
|
|
self._ipmi_conn()
|
|
|
|
def close(self) -> None:
|
|
self._redfish.close()
|
|
if self._ipmi_open:
|
|
self._ipmi.close()
|
|
self._ipmi_open = False
|
|
|
|
def is_alive(self) -> _JsonDict:
|
|
if self._redfish_ok:
|
|
return self._redfish.is_alive()
|
|
return self._ipmi.is_alive()
|
|
|
|
def _ipmi_conn(self) -> IpmiDriver:
|
|
if not self._ipmi_open:
|
|
self._ipmi.open()
|
|
self._ipmi_open = True
|
|
return self._ipmi
|
|
|
|
def _call(self, name: str, *args: Any) -> Any:
|
|
"""``name`` over Redfish; over IPMI when Redfish is out or fails and IPMI has it."""
|
|
has_ipmi = callable(getattr(IpmiDriver, name, None))
|
|
if self._redfish_ok:
|
|
try:
|
|
return getattr(self._redfish, name)(*args)
|
|
except (RedfishError, ConnectionException) as exc:
|
|
if not has_ipmi:
|
|
raise
|
|
LOG.info("%s: %s failed over Redfish (%s); asking IPMI", self.hostname, name, exc)
|
|
if not has_ipmi:
|
|
raise NotImplementedError(f"{self.hostname}: {name} needs Redfish")
|
|
return getattr(self._ipmi_conn(), name)(*args)
|
|
|
|
# --- NAPALM getters ------------------------------------------------------------------
|
|
|
|
def get_facts(self) -> _JsonDict:
|
|
return self._call("get_facts")
|
|
|
|
def get_interfaces(self) -> _JsonDict:
|
|
return self._call("get_interfaces")
|
|
|
|
def get_interfaces_ip(self) -> _JsonDict:
|
|
return self._call("get_interfaces_ip")
|
|
|
|
def get_environment(self) -> _JsonDict:
|
|
return self._call("get_environment")
|
|
|
|
def get_users(self) -> _JsonDict:
|
|
return self._call("get_users")
|
|
|
|
def get_ntp_servers(self) -> _JsonDict:
|
|
return self._call("get_ntp_servers")
|
|
|
|
def get_snmp_information(self) -> _JsonDict:
|
|
return self._call("get_snmp_information")
|
|
|
|
def get_licenses(self) -> list[_JsonDict]:
|
|
"""Redfish's license facts, also while IPMI reads the BMC: IPMI knows none."""
|
|
return self._redfish.get_licenses()
|
|
|
|
def get_device_warnings(self) -> list[_JsonDict]:
|
|
"""Redfish's signals; for a BMC that refuses Redfish, with ``ipmi_fallback`` set, as
|
|
this driver reads it over IPMI meanwhile."""
|
|
warnings = self._redfish.get_device_warnings()
|
|
if self._redfish_ok:
|
|
return warnings
|
|
return [{**w, "meta": {**w.get("meta", {}), "ipmi_fallback": True}} for w in warnings]
|
|
|
|
# --- role methods --------------------------------------------------------------------
|
|
|
|
def get_power_state(self) -> _JsonDict:
|
|
return self._call("get_power_state")
|
|
|
|
def set_power_state(self, action: str) -> None:
|
|
self._call("set_power_state", action)
|
|
|
|
def reboot_host(self) -> None:
|
|
"""Restart the BMC (``Manager.Reset``), never the server. Needs Redfish."""
|
|
if not self._redfish_ok:
|
|
raise RuntimeError(f"{self.hostname}: resetting the BMC needs Redfish")
|
|
self._redfish.reboot_host()
|
|
|
|
def get_hardware_inventory(self) -> _JsonDict:
|
|
return self._call("get_hardware_inventory")
|
|
|
|
def get_event_log(self, limit: int = 500) -> list[_JsonDict]:
|
|
return self._call("get_event_log", limit)
|
|
|
|
def get_firmware_inventory(self) -> list[_JsonDict]:
|
|
return self._call("get_firmware_inventory")
|
|
|
|
def get_management_protocols(self) -> _JsonDict:
|
|
"""Over Redfish, what the BMC reports. Over IPMI only one thing is known: IPMI over
|
|
LAN is on, since this driver is talking it."""
|
|
if self._redfish_ok:
|
|
return self._redfish.get_management_protocols()
|
|
return {"ipmi": {"enabled": True, "port": self.ipmi_port}}
|