Christian Manivong
cab1c84cba
Pure functions for ServiceRoot, ComputerSystem (facts, power state, reset action and ActionInfo), Processors, Memory, Storage/Drives/SimpleStorage, Chassis Thermal and Power as well as ThermalSubsystem/PowerSubsystem/Sensors (with environment_sources to pick whichever exists), Manager, EthernetInterfaces, ManagerNetworkProtocol, accounts, firmware inventory and log services/entries.
129 lines
4.8 KiB
Python
129 lines
4.8 KiB
Python
"""Hardware of the managed server: Processors, Memory, Storage, Drives, SimpleStorage."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Mapping
|
|
|
|
from .common import boolean, ident, integer, link, link_list, location_label, present, status, text
|
|
|
|
|
|
def parse_processor(cpu: Mapping[str, Any]) -> dict[str, Any]:
|
|
"""One member of ``Systems/{id}/Processors``."""
|
|
state, health = status(cpu)
|
|
return {
|
|
"id": text(cpu.get("Id")),
|
|
"name": text(cpu.get("Name")),
|
|
"socket": text(cpu.get("Socket")) or location_label(cpu),
|
|
"processor_type": text(cpu.get("ProcessorType")),
|
|
"manufacturer": ident(cpu.get("Manufacturer")),
|
|
"model": ident(cpu.get("Model")),
|
|
"architecture": text(cpu.get("ProcessorArchitecture")),
|
|
"instruction_set": text(cpu.get("InstructionSet")),
|
|
"max_speed_mhz": integer(cpu.get("MaxSpeedMHz")),
|
|
"cores": integer(cpu.get("TotalCores")),
|
|
"threads": integer(cpu.get("TotalThreads")),
|
|
"serial_number": ident(cpu.get("SerialNumber")),
|
|
"present": present(cpu),
|
|
"state": state,
|
|
"health": health,
|
|
}
|
|
|
|
|
|
def parse_memory(dimm: Mapping[str, Any]) -> dict[str, Any]:
|
|
"""One member of ``Systems/{id}/Memory``.
|
|
|
|
An enabled slot with a capacity of 0 counts as empty: some BMCs list empty slots so.
|
|
"""
|
|
state, health = status(dimm)
|
|
capacity = integer(dimm.get("CapacityMiB"))
|
|
return {
|
|
"id": text(dimm.get("Id")),
|
|
"name": text(dimm.get("Name")),
|
|
"location": (location_label(dimm) or text(dimm.get("DeviceLocator"))
|
|
or text(dimm.get("Name"))),
|
|
"capacity_mib": capacity,
|
|
"memory_device_type": text(dimm.get("MemoryDeviceType")),
|
|
"base_module_type": text(dimm.get("BaseModuleType")),
|
|
"speed_mhz": integer(dimm.get("OperatingSpeedMhz")),
|
|
"manufacturer": ident(dimm.get("Manufacturer")),
|
|
"part_number": ident(dimm.get("PartNumber")),
|
|
"serial_number": ident(dimm.get("SerialNumber")),
|
|
"present": present(dimm) and capacity != 0,
|
|
"state": state,
|
|
"health": health,
|
|
}
|
|
|
|
|
|
def _controller(ctrl: Mapping[str, Any]) -> dict[str, Any]:
|
|
state, health = status(ctrl)
|
|
return {
|
|
"name": text(ctrl.get("Name")),
|
|
"manufacturer": ident(ctrl.get("Manufacturer")),
|
|
"model": ident(ctrl.get("Model")),
|
|
"serial_number": ident(ctrl.get("SerialNumber")),
|
|
"firmware_version": text(ctrl.get("FirmwareVersion")),
|
|
"speed_gbps": ctrl.get("SpeedGbps") if isinstance(ctrl.get("SpeedGbps"), (int, float))
|
|
else None,
|
|
"state": state,
|
|
"health": health,
|
|
}
|
|
|
|
|
|
def parse_storage(storage: Mapping[str, Any]) -> dict[str, Any]:
|
|
"""One member of ``Systems/{id}/Storage``.
|
|
|
|
Drives are separate resources: ``drive_paths`` lists them for ``parse_drive``. Storage
|
|
1.15 and later move controllers into a collection at ``controllers_path``.
|
|
"""
|
|
state, health = status(storage)
|
|
controllers = storage.get("StorageControllers")
|
|
return {
|
|
"id": text(storage.get("Id")),
|
|
"name": text(storage.get("Name")),
|
|
"state": state,
|
|
"health": health,
|
|
"controllers": [_controller(c) for c in controllers if isinstance(c, Mapping)]
|
|
if isinstance(controllers, list) else [],
|
|
"controllers_path": link(storage, "Controllers"),
|
|
"drive_paths": link_list(storage, "Drives"),
|
|
}
|
|
|
|
|
|
def parse_drive(drive: Mapping[str, Any]) -> dict[str, Any]:
|
|
"""A Drive resource, as linked from Storage."""
|
|
state, health = status(drive)
|
|
return {
|
|
"id": text(drive.get("Id")),
|
|
"name": text(drive.get("Name")),
|
|
"manufacturer": ident(drive.get("Manufacturer")),
|
|
"model": ident(drive.get("Model")),
|
|
"serial_number": ident(drive.get("SerialNumber")),
|
|
"part_number": ident(drive.get("PartNumber")),
|
|
"revision": text(drive.get("Revision")),
|
|
"capacity_bytes": integer(drive.get("CapacityBytes")),
|
|
"media_type": text(drive.get("MediaType")),
|
|
"protocol": text(drive.get("Protocol")),
|
|
"failure_predicted": boolean(drive.get("FailurePredicted")),
|
|
"present": present(drive),
|
|
"state": state,
|
|
"health": health,
|
|
}
|
|
|
|
|
|
def parse_simple_storage(storage: Mapping[str, Any]) -> dict[str, Any]:
|
|
"""One member of ``Systems/{id}/SimpleStorage`` (older services, no Storage).
|
|
|
|
Its devices come out with the same keys as ``parse_drive``.
|
|
"""
|
|
state, health = status(storage)
|
|
devices = storage.get("Devices")
|
|
drives = [parse_drive(dev) for dev in devices if isinstance(dev, Mapping)] \
|
|
if isinstance(devices, list) else []
|
|
return {
|
|
"id": text(storage.get("Id")),
|
|
"name": text(storage.get("Name")),
|
|
"state": state,
|
|
"health": health,
|
|
"drives": drives,
|
|
}
|