nic firmware enhancements

nodescraper/plugins/inband/network/analyzer_args.py

@@ -0,0 +1,19 @@
+###############################################################################
+#
+# MIT License
+#
+###############################################################################
+from typing import Optional
+
+from pydantic import Field
+
+from nodescraper.models import AnalyzerArgs
+
+
+class NetworkAnalyzerArgs(AnalyzerArgs):
+    """Arguments for network and ethtool analysis."""
+
+    expected_nic_firmware: Optional[str] = Field(
+        default=None,
+        description="Exact firmware version expected from each collected NIC.",
+    )

nodescraper/plugins/inband/network/network_analyzer.py

@@ -24,20 +24,86 @@
 #
 ###############################################################################
 import re
+from typing import Any, Dict, Optional
 
 from nodescraper.base.regexanalyzer import RegexAnalyzer
 from nodescraper.enums import EventCategory, EventPriority, ExecutionStatus
 from nodescraper.models import TaskResult
 
+from .analyzer_args import NetworkAnalyzerArgs
 from .networkdata import NetworkDataModel
 
 
-class NetworkAnalyzer(RegexAnalyzer[NetworkDataModel, None]):
+def _validate_firmware_policy(records: list[Dict[str, Any]], policies: Any) -> list[Dict[str, Any]]:
+    """Validate ethtool firmware locally against this analyzer's policies."""
+    policy_list = [policies] if isinstance(policies, dict) else policies or []
+    issues = []
+    for record in records:
+        policy = next(
+            (
+                item
+                for item in policy_list
+                if isinstance(item, dict) and _policy_matches(record, item)
+            ),
+            None,
+        )
+        if policy is None:
+            continue
+        actual = _normalize_firmware_version(record.get("version"))
+        expected = _normalize_firmware_version(policy.get("expected_nic_firmware"))
+        if not actual:
+            reason = "firmware version is unavailable"
+        elif expected and not _firmware_matches(actual, expected):
+            reason = (
+                f"actual {record.get('version')} != expected_nic_firmware "
+                f"{policy.get('expected_nic_firmware')}"
+            )
+        else:
+            continue
+        issues.append({"reason": reason, "record": record, "policy": dict(policy)})
+    return issues
+
+
+def _policy_matches(record: Dict[str, Any], policy: Dict[str, Any]) -> bool:
+    match = policy.get("match", {})
+    if not isinstance(match, dict):
+        return False
+    if not match:
+        match = {field: policy[field] for field in record if field in policy}
+    for field, expected in match.items():
+        actual = record.get(str(field))
+        if actual is None:
+            return False
+        expected_values = expected if isinstance(expected, list) else [expected]
+        if not any(
+            str(actual).strip().lower() == str(value).strip().lower() for value in expected_values
+        ):
+            return False
+    return True
+
+
+def _normalize_firmware_version(value: Any) -> Optional[str]:
+    if value is None:
+        return None
+    normalized = str(value).strip().strip("'\"").lower()
+    return normalized or None
+
+
+def _firmware_matches(actual: str, expected_pattern: str) -> bool:
+    try:
+        return re.fullmatch(expected_pattern, actual, flags=re.IGNORECASE) is not None
+    except re.error:
+        return False
+
+
+class NetworkAnalyzer(RegexAnalyzer[NetworkDataModel, NetworkAnalyzerArgs]):
     """Check network statistics for errors."""
 
     DATA_MODEL = NetworkDataModel
 
-    def analyze_data(self, data: NetworkDataModel, args=None) -> TaskResult:
+    def analyze_data(
+        self, data: NetworkDataModel, args: Optional[NetworkAnalyzerArgs] = None
+    ) -> TaskResult:
         """Analyze ethtool -S statistics via RDMA-scoped vendor models.
 
         Args:
@@ -47,6 +113,40 @@ def analyze_data(self, data: NetworkDataModel, args=None) -> TaskResult:
         Returns:
             TaskResult with OK, WARNING (no devices, or only warning-tier counters), or ERROR.
         """
+        args = args or NetworkAnalyzerArgs()
+        firmware_records = []
+        for interface, info in data.ethtool_info.items():
+            vendor = _vendor_for_driver(info.driver)
+            if vendor is None:
+                continue
+            firmware_records.append(
+                {
+                    "identity": interface,
+                    "version": info.firmware_version,
+                    "source": "ethtool",
+                    "vendor": vendor,
+                    "driver": info.driver,
+                    "interface": interface,
+                    "pci_bdf": info.bus_info,
+                }
+            )
+        firmware_policy_issues = _validate_firmware_policy(
+            firmware_records,
+            (
+                {"expected_nic_firmware": args.expected_nic_firmware}
+                if args.expected_nic_firmware
+                else None
+            ),
+        )
+        for issue in firmware_policy_issues:
+            self._log_event(
+                category=EventCategory.NETWORK,
+                description="Network adapter firmware policy mismatch",
+                data=issue,
+                priority=EventPriority.WARNING,
+                console_log=True,
+            )
+
         if not data.ethtool_info and not data.ethtool_statistics:
             self.result.message = "No network devices found"
             self.result.status = ExecutionStatus.WARNING
@@ -165,8 +265,25 @@ def analyze_data(self, data: NetworkDataModel, args=None) -> TaskResult:
         elif vendor_warning:
             self.result.message = "Network warning counters non-zero in statistics"
             self.result.status = ExecutionStatus.WARNING
+        elif firmware_policy_issues:
+            self.result.message = "Network adapter firmware policy mismatch"
+            self.result.status = ExecutionStatus.WARNING
         else:
             self.result.message = "No network errors detected in statistics"
             self.result.status = ExecutionStatus.OK
 
         return self.result
+
+
+def _vendor_for_driver(driver: Optional[str]) -> Optional[str]:
+    """Return a stable vendor label for common ethtool driver families."""
+    if not driver:
+        return None
+    normalized = driver.lower()
+    if normalized.startswith(("bnxt", "bnx2")):
+        return "Broadcom"
+    if normalized.startswith(("mlx", "ib_")):
+        return "Mellanox"
+    if normalized.startswith(("ionic",)):
+        return "Pensando"
+    return None

nodescraper/plugins/inband/network/network_collector.py

@@ -450,6 +450,24 @@ def _parse_ethtool(self, interface: str, output: str) -> EthtoolInfo:
 
         return ethtool_info
 
+    def _parse_ethtool_driver_info(self, info: EthtoolInfo, output: str) -> None:
+        """Merge ``ethtool -i`` identity and firmware fields into an interface model."""
+        for line in output.splitlines():
+            if ":" not in line:
+                continue
+            key, value = (part.strip() for part in line.split(":", 1))
+            if not key or not value:
+                continue
+            info.settings[key] = value
+            normalized_key = key.lower().replace("-", "_").replace(" ", "_")
+            if normalized_key == "driver":
+                info.driver = value
+            elif normalized_key in {"bus_info", "bus"}:
+                info.bus_info = value
+            elif normalized_key in {"firmware_version", "firmware"}:
+                firmware_match = re.search(r"\d+(?:\.\d+){2,}", value)
+                info.firmware_version = firmware_match.group(0) if firmware_match else value
+
     def _parse_ethtool_statistics(self, output: str, interface: str) -> Dict[str, str]:
         """Parse 'ethtool -S <interface>' output into a key-value dictionary.
 
@@ -522,6 +540,18 @@ def _collect_ethtool_info(
                     priority=EventPriority.WARNING,
                 )
 
+            # ``ethtool <interface>`` does not include firmware identity on all
+            # drivers, so always retain the separate ``ethtool -i`` response.
+            res_driver = self._run_sut_cmd(
+                self.CMD_ETHTOOL_I_TEMPLATE.format(interface=iface.name), sudo=True
+            )
+            if res_driver.exit_code == 0:
+                ethtool_info = ethtool_data.get(
+                    iface.name, EthtoolInfo(interface=iface.name, raw_output="")
+                )
+                self._parse_ethtool_driver_info(ethtool_info, res_driver.stdout)
+                ethtool_data[iface.name] = ethtool_info
+
         return ethtool_data, skipped
 
     def _collect_ethtool_statistic(self, netdev: str, driver: str) -> Optional[EthtoolStatistics]:

nodescraper/plugins/inband/network/network_plugin.py

@@ -25,19 +25,21 @@
 ###############################################################################
 from nodescraper.base import InBandDataPlugin
 
+from .analyzer_args import NetworkAnalyzerArgs
 from .collector_args import NetworkCollectorArgs
 from .network_analyzer import NetworkAnalyzer
 from .network_collector import NetworkCollector
 from .networkdata import NetworkDataModel
 
 
-class NetworkPlugin(InBandDataPlugin[NetworkDataModel, NetworkCollectorArgs, None]):
+class NetworkPlugin(InBandDataPlugin[NetworkDataModel, NetworkCollectorArgs, NetworkAnalyzerArgs]):
     """Plugin for collection of network configuration data"""
 
     DATA_MODEL = NetworkDataModel
 
     COLLECTOR = NetworkCollector
 
     COLLECTOR_ARGS = NetworkCollectorArgs
+    ANALYZER_ARGS = NetworkAnalyzerArgs
 
     ANALYZER = NetworkAnalyzer

nodescraper/plugins/inband/network/networkdata.py

@@ -97,6 +97,9 @@ class EthtoolInfo(BaseModel):
 
     interface: str  # Interface name this info belongs to
     raw_output: str  # Raw ethtool command output
+    driver: Optional[str] = None  # Kernel driver from ethtool -i
+    bus_info: Optional[str] = None  # PCI bus identity from ethtool -i
+    firmware_version: Optional[str] = None  # Firmware version from ethtool -i
     settings: Dict[str, str] = Field(default_factory=dict)  # Parsed key-value settings
     supported_link_modes: List[str] = Field(default_factory=list)  # Supported link modes
     advertised_link_modes: List[str] = Field(default_factory=list)  # Advertised link modes

nodescraper/plugins/inband/nic/analyzer_args.py

@@ -33,6 +33,10 @@
 class NicAnalyzerArgs(AnalyzerArgs):
     """Analyzer args for niccli/nicctl data, with expected_values keyed by canonical command key."""
 
+    expected_nic_firmware: Optional[str] = Field(
+        default=None,
+        description="Exact firmware version expected from each collected NIC.",
+    )
     expected_values: Optional[Dict[str, Dict[str, Any]]] = Field(
         default=None,
         description="Per-command expected checks keyed by canonical key (see command_to_canonical_key).",