Vendor info enhancement

nodescraper/plugins/inband/amdsmi/amdsmi_analyzer.py

@@ -31,17 +31,20 @@
 from nodescraper.enums import EventCategory, EventPriority
 from nodescraper.interfaces import DataAnalyzer
 from nodescraper.models import TaskResult
+from nodescraper.utils import unit_size_to_bytes
 
 from .amdsmidata import (
     AmdSmiDataModel,
     AmdSmiMetric,
     AmdSmiStatic,
     EccData,
+    EccState,
     Fabric,
     Fw,
     LinkStatusTable,
     Partition,
     Processes,
+    ValueUnit,
     XgmiLinks,
     XgmiMetrics,
 )
@@ -83,6 +86,24 @@ def _gpu_unavailable_description(
     return description, description
 
 
+def _umc_ecc_state(ecc_block_state: Union[dict, str]) -> Optional[EccState]:
+    """Return the UMC block state from amd-smi static RAS data."""
+    if not isinstance(ecc_block_state, dict):
+        return None
+    for name, state in ecc_block_state.items():
+        if str(name).strip().upper() == "UMC":
+            return state if isinstance(state, EccState) else None
+    return None
+
+
+def _vram_size_gib(size: ValueUnit) -> Optional[float]:
+    """Convert an amd-smi VRAM size to GiB."""
+    nbytes = unit_size_to_bytes(size.value, size.unit)
+    if nbytes is None:
+        return None
+    return nbytes / (1024**3)
+
+
 def _is_all_zeros(value: Union[str, list[str]]) -> bool:
     """Return True if every comma or space separated entry is zero."""
     values = value if isinstance(value, list) else [value]
@@ -605,6 +626,71 @@ def check_gpu_memory(
                     console_log=True,
                 )
 
+    def check_min_hbm_gib_per_gpu(
+        self, amdsmi_static_data: list[AmdSmiStatic], minimum_gib: float
+    ) -> None:
+        """Require each GPU's static VRAM size to be at least minimum_gib."""
+        short: dict[int, float] = {}
+        unavailable: list[int] = []
+        for gpu_data in amdsmi_static_data:
+            size = gpu_data.vram.size if gpu_data.vram is not None else None
+            gib = _vram_size_gib(size) if size is not None else None
+            if gib is None:
+                unavailable.append(gpu_data.gpu)
+            elif gib < minimum_gib:
+                short[gpu_data.gpu] = round(gib, 2)
+
+        if unavailable:
+            desc, details = _gpu_unavailable_description(
+                "HBM size",
+                unavailable,
+                expected=f"{minimum_gib} GiB",
+            )
+            self._log_event(
+                category=EventCategory.PLATFORM,
+                description=desc,
+                priority=EventPriority.ERROR,
+                data={"gpus": unavailable, "min_hbm_gib_per_gpu": minimum_gib, "details": details},
+                console_log=True,
+            )
+        if short:
+            desc, details = _gpu_mismatch_description(
+                "HBM size",
+                minimum_gib,
+                short,
+                unit=" GiB",
+            )
+            self._log_event(
+                category=EventCategory.PLATFORM,
+                description=desc,
+                priority=EventPriority.ERROR,
+                data={
+                    "hbm_gib_by_gpu": short,
+                    "min_hbm_gib_per_gpu": minimum_gib,
+                    "details": details,
+                },
+                console_log=True,
+            )
+
+    def check_umc_ras_enabled(self, amdsmi_static_data: list[AmdSmiStatic]) -> None:
+        """Require UMC RAS ENABLED. That is the in-band proxy for HBM ECC being on."""
+        actual_by_gpu: dict[int, str] = {}
+        for gpu_data in amdsmi_static_data:
+            umc_state = _umc_ecc_state(gpu_data.ras.ecc_block_state)
+            if umc_state != EccState.ENABLED:
+                shown = umc_state.value if isinstance(umc_state, EccState) else "missing"
+                actual_by_gpu[gpu_data.gpu] = shown
+        if not actual_by_gpu:
+            return
+        desc, details = _gpu_mismatch_description("UMC RAS (HBM ECC)", "ENABLED", actual_by_gpu)
+        self._log_event(
+            category=EventCategory.RAS,
+            description=desc,
+            priority=EventPriority.ERROR,
+            data={"umc_ras_by_gpu": actual_by_gpu, "details": details},
+            console_log=True,
+        )
+
     def check_amdsmi_metric_ecc(self, amdsmi_metric_data: list[AmdSmiMetric]):
         """Check ECC counts in all blocks for all GPUs
 
@@ -1162,7 +1248,7 @@ def _add(gpu: int, field: str, expected: object, actual: object) -> None:
             gpus_affected = len({i["gpu"] for i in issues})
             self._log_event(
                 category=EventCategory.NETWORK,
-                description=f"Fabric data mismatch on {gpus_affected} GPU(s): {details}",
+                description=f"Fabric data mismatch on {gpus_affected} GPU(s)",
                 priority=EventPriority.ERROR,
                 data={
                     "expected_accel_state": expected_state,
@@ -1283,6 +1369,13 @@ def analyze_data(
                 priority=EventPriority.WARNING,
                 data={"amdsmi_static_data": data.static},
             )
+            if args.min_hbm_gib_per_gpu is not None or args.ecc_required:
+                self._log_event(
+                    category=EventCategory.PLATFORM,
+                    description="AMD SMI static data is required for HBM size and ECC gates",
+                    priority=EventPriority.ERROR,
+                    console_log=True,
+                )
         else:
             if args.expected_max_power:
                 self.check_expected_max_power(data.static, args.expected_max_power)
@@ -1307,6 +1400,10 @@ def analyze_data(
                     (args.devid_ep, args.devid_ep),
                     sku_name=args.sku_name,
                 )
+            if args.min_hbm_gib_per_gpu is not None:
+                self.check_min_hbm_gib_per_gpu(data.static, args.min_hbm_gib_per_gpu)
+            if args.ecc_required:
+                self.check_umc_ras_enabled(data.static)
 
         if args.expected_memory_partition_mode or args.expected_compute_partition_mode:
             self.check_expected_memory_partition_mode(

nodescraper/plugins/inband/amdsmi/amdsmi_collector.py

@@ -370,6 +370,28 @@ def _valueunit_req(self, v: object, unit: str) -> ValueUnit:
         assert vu is not None
         return vu
 
+    def _static_vram_size(self, vram: dict) -> Optional[ValueUnit]:
+        """Read VRAM capacity from current or older amd-smi static JSON.
+
+        Current amd-smi reports ``size: {value, unit}``. Older output reports
+        ``vram_size_mb`` as a number of megabytes.
+        """
+        raw_size = vram.get("size")
+        if isinstance(raw_size, dict):
+            value = raw_size.get("value")
+            if value in (None, "", "N/A"):
+                return None
+            unit = str(raw_size.get("unit") or "B")
+            return self._valueunit(value, unit)
+
+        if vram.get("vram_size_mb") is not None:
+            try:
+                nbytes = int(vram["vram_size_mb"]) * 1024 * 1024
+            except (TypeError, ValueError):
+                return None
+            return self._valueunit(nbytes, "B")
+        return None
+
     def _normalize(self, val: object, default: str = "unknown", slot_type: bool = False) -> str:
         """Normalize strings
 
@@ -1327,17 +1349,10 @@ def get_static(self) -> Optional[list[AmdSmiStatic]]:
             vram_type = str(vram.get("vram_type", "") or "unknown")
             vram_vendor = vram.get("vram_vendor")
             vram_bits = vram.get("vram_bit_width")
-            vram_size_b: Optional[int] = None
-            if vram.get("vram_size_mb") is not None:
-                try:
-                    vram_size_b = int(vram["vram_size_mb"]) * 1024 * 1024
-                except Exception:
-                    vram_size_b = None
-
             vram_model = StaticVram(
                 type=vram_type,
                 vendor=None if vram_vendor in (None, "", "N/A") else str(vram_vendor),
-                size=self._valueunit(vram_size_b, "B"),
+                size=self._static_vram_size(vram),
                 bit_width=self._valueunit(vram_bits, "bit"),
                 max_bandwidth=None,
             )

nodescraper/plugins/inband/amdsmi/analyzer_args.py

@@ -90,6 +90,21 @@ class AmdSmiAnalyzerArgs(AnalyzerArgs):
         default=None,
         description="Minimum free VRAM threshold to validate for each GPU.",
     )
+    min_hbm_gib_per_gpu: Optional[float] = Field(
+        default=None,
+        gt=0,
+        description=(
+            "Minimum HBM capacity in GiB required on each GPU. "
+            "Compared to amd-smi static vram size."
+        ),
+    )
+    ecc_required: bool = Field(
+        default=False,
+        description=(
+            "When true, require UMC RAS ENABLED on each GPU. "
+            "That block state is the in-band proxy for HBM ECC being on."
+        ),
+    )
     check_xgmi_or_peer_links_status: bool = Field(
         default=False,
         description="Check XGMI or peer-link status for each GPU.",

nodescraper/plugins/inband/bios/analyzer_args.py

@@ -28,37 +28,46 @@
 from pydantic import Field, field_validator
 
 from nodescraper.models import AnalyzerArgs
-from nodescraper.plugins.inband.os.osdata import OsDataModel
+from nodescraper.plugins.inband.bios.biosdata import BiosDataModel
 
 
 class BiosAnalyzerArgs(AnalyzerArgs):
     exp_bios_version: list[str] = Field(
         default_factory=list,
         description="Expected BIOS version(s) to match against collected value (str or list).",
     )
+    exp_server_vendor: list[str] = Field(
+        default_factory=list,
+        description="Expected server vendor(s) from DMI system manufacturer (str or list).",
+    )
+    exp_server_model: list[str] = Field(
+        default_factory=list,
+        description="Expected server model(s) from DMI system product name (str or list).",
+    )
     regex_match: bool = Field(
         default=False,
-        description="If True, match exp_bios_version as regex; otherwise exact match.",
+        description="If True, match expected BIOS, vendor, and model values as regex; otherwise exact match.",
     )
 
-    @field_validator("exp_bios_version", mode="before")
+    @field_validator("exp_bios_version", "exp_server_vendor", "exp_server_model", mode="before")
     @classmethod
-    def validate_exp_bios_version(cls, exp_bios_version: Union[str, list]) -> list:
-        """support str or list input for exp_bios_version
+    def validate_expected_str_list(cls, value: Union[str, list, None]) -> list:
+        """support str or list input for expected string fields
 
         Args:
-            exp_bios_version (Union[str, list]): expected BIOS version(s) to match against
+            value (Union[str, list, None]): expected value(s) to match against
 
         Returns:
-            list: a list of expected BIOS versions
+            list: a list of expected values
         """
-        if isinstance(exp_bios_version, str):
-            exp_bios_version = [exp_bios_version]
-
-        return exp_bios_version
+        if value is None:
+            return []
+        if isinstance(value, str):
+            return [value]
+        return value
 
     @classmethod
-    def build_from_model(cls, datamodel: OsDataModel) -> "BiosAnalyzerArgs":
+    def build_from_model(cls, datamodel: BiosDataModel) -> "BiosAnalyzerArgs":
         """build analyzer args from data model
 
         Args:
@@ -67,4 +76,10 @@ def build_from_model(cls, datamodel: OsDataModel) -> "BiosAnalyzerArgs":
         Returns:
             BiosAnalyzerArgs: instance of analyzer args class
         """
-        return cls(exp_bios_version=datamodel.bios_version)
+        return cls(
+            exp_bios_version=datamodel.bios_version,
+            exp_server_vendor=(
+                [datamodel.server_vendor] if datamodel.server_vendor is not None else []
+            ),
+            exp_server_model=[datamodel.server_model] if datamodel.server_model is not None else [],
+        )