Add collector for ssacli (#12)

* Add collector for ssacli

* Change status metrics to info metrics

prometheus_hardware_exporter/__main__.py

@@ -10,6 +10,7 @@
     LSISASControllerCollector,
     MegaRAIDCollector,
     PowerEdgeRAIDCollector,
+    SsaCLICollector,
 )
 from .config import DEFAULT_CONFIG, Config
 from .exporter import Exporter
@@ -49,6 +50,7 @@ def main() -> None:
     exporter.register(LSISASControllerCollector(2))
     exporter.register(LSISASControllerCollector(3))
     exporter.register(PowerEdgeRAIDCollector())
+    exporter.register(SsaCLICollector())
     exporter.run()
 
 

prometheus_hardware_exporter/collector.py

@@ -10,6 +10,7 @@
 from .collectors.ipmimonitoring import IpmiMonitoring
 from .collectors.perccli import PercCLI
 from .collectors.sasircu import LSISASCollectorHelper, Sasircu
+from .collectors.ssacli import SsaCLI
 from .collectors.storcli import MegaRAIDCollectorHelper, StorCLI
 from .core import BlockingCollector, Payload, Specification
 
@@ -733,3 +734,101 @@ def fetch(self) -> List[Payload]:
     def process(self, payloads: List[Payload], datastore: Dict[str, Payload]) -> List[Payload]:
         """Process the payload if needed."""
         return payloads
+
+
+class SsaCLICollector(BlockingCollector):
+    """Collector for storage arrays that support ssacli."""
+
+    ssacli = SsaCLI()
+
+    @property
+    def specifications(self) -> List[Specification]:
+        return [
+            Specification(
+                name="ssacli_command_success",
+                documentation="Indicates if the command is successful or not",
+                metric_class=GaugeMetricFamily,
+            ),
+            Specification(
+                name="ssacli_controllers",
+                documentation="Total number of controllers",
+                metric_class=GaugeMetricFamily,
+            ),
+            Specification(
+                name="ssacli_controller",
+                documentation="Shows the information about controller part",
+                metric_class=InfoMetricFamily,
+            ),
+            Specification(
+                name="ssacli_logical_drives",
+                documentation="The number of logical drives in the slot",
+                labels=["slot"],
+                metric_class=GaugeMetricFamily,
+            ),
+            Specification(
+                name="ssacli_physical_drives",
+                documentation="The number of physical drives in the slot",
+                labels=["slot"],
+                metric_class=GaugeMetricFamily,
+            ),
+            Specification(
+                name="ssacli_logical_drive",
+                documentation="Shows the information about logical drive",
+                metric_class=InfoMetricFamily,
+            ),
+            Specification(
+                name="ssacli_physical_drive",
+                documentation="Shows the information about physical drive",
+                metric_class=InfoMetricFamily,
+            ),
+        ]
+
+    def fetch(self) -> List[Payload]:
+        """Load the controller and drive status using ssacli tool."""
+        ssacli_payload = self.ssacli.get_payload()
+
+        if not ssacli_payload:
+            logger.error("Failed to get controllers info using ssacli.")
+            return [Payload(name="ssacli_command_success", value=0.0)]
+        payloads = [
+            Payload(name="ssacli_command_success", value=1.0),
+            Payload(name="ssacli_controllers", value=len(ssacli_payload)),
+        ]
+
+        for slot, payload in ssacli_payload.items():
+            ctrl_status = payload["controller_status"]
+            for part, status in ctrl_status.items():
+                payloads.append(
+                    Payload(
+                        name="ssacli_controller",
+                        value={"slot": slot, "part": part, "status": status},
+                    )
+                )
+            ld_status = payload["ld_status"]
+            payloads.append(
+                Payload(name="ssacli_logical_drives", labels=[slot], value=len(ld_status))
+            )
+            for drive_id, status in ld_status.items():
+                payloads.append(
+                    Payload(
+                        name="ssacli_logical_drive",
+                        value={"slot": slot, "drive_id": drive_id, "status": status},
+                    )
+                )
+            pd_status = payload["pd_status"]
+            payloads.append(
+                Payload(name="ssacli_physical_drives", labels=[slot], value=len(pd_status))
+            )
+            for drive_id, status in pd_status.items():
+                payloads.append(
+                    Payload(
+                        name="ssacli_physical_drive",
+                        value={"slot": slot, "drive_id": drive_id, "status": status},
+                    )
+                )
+
+        return payloads
+
+    def process(self, payloads: List[Payload], datastore: Dict[str, Payload]) -> List[Payload]:
+        """Process the payload if needed."""
+        return payloads

prometheus_hardware_exporter/collectors/ssacli.py

@@ -0,0 +1,137 @@
+"""Collector for HP Storage Array."""
+
+import re
+from logging import getLogger
+from typing import Dict, List
+
+from ..utils import Command
+
+logger = getLogger(__name__)
+
+
+class SsaCLI(Command):
+    """Command line tool for HP Storage Controller."""
+
+    prefix = ""
+    command = "ssacli"
+
+    def get_payload(self) -> Dict[str, Dict]:
+        """Get status of all controllers, logical drives and physical drives.
+
+        Returns:
+            payload: a dictionary where key, value pair is slot, controller information pair
+            in the following format:
+            {
+                "slot": {"controller_status": controller_status,
+                "ld_status": ld_status,
+                "pd_status": pd_status
+                }
+            }
+        """
+        slots = self._get_controller_slots()
+        payload = {}
+        for slot in slots:
+            controller_status = self._get_controller_status(slot)
+            ld_status = self._get_ld_status(slot)
+            pd_status = self._get_pd_status(slot)
+            payload[slot] = {
+                "controller_status": controller_status,
+                "ld_status": ld_status,
+                "pd_status": pd_status,
+            }
+        return payload
+
+    def _get_controller_slots(self) -> List[str]:
+        """Get the controller slots(s) available for probing.
+
+        Returns:
+            slots: List of slots where HP storage controllers are available, or []
+        """
+        result = self("ctrl all show")
+        if result.error:
+            logger.error(result.error)
+            return []
+
+        controllers_raw = result.data.strip().split("\n")
+        slots = []
+        for controller in controllers_raw:
+            if "in Slot" in controller:
+                slots.append(controller.split()[5])
+        return slots
+
+    def _get_controller_status(self, slot: str) -> Dict[str, str]:
+        """Get controller status for each part of the controller.
+
+        Returns:
+            ctrl_status: dict where each item is part, status pair for the controller.
+        """
+        result = self(f"ctrl slot={slot} show status")
+        if result.error:
+            logger.error(result.error)
+            return {}
+
+        ctrl_status = {}
+        for line in result.data.splitlines():
+            line = line.strip()
+            if (not line) or line.startswith("Smart Array") or line.startswith("Smart HBA"):
+                continue
+            if ":" not in line:
+                err = f"Unrecognised line for controller '{line}'"
+                logger.warning(err)
+                continue
+            part, status = line.split(":")
+            ctrl_status[part] = status.strip().upper()
+        return ctrl_status
+
+    def _get_ld_status(self, slot: str) -> Dict[str, str]:
+        """Get logical drives status.
+
+        Returns:
+            ld_status: dict where each item is drive_id, status pair.
+        """
+        result = self(f"ctrl slot={slot} ld all show status")
+        if result.error:
+            logger.error(result.error)
+            return {}
+
+        innocuous_errors = re.compile(
+            r"^Error: The specified (device|controller) does not have any logical"
+        )
+        drive_status_line = re.compile(r"^\s*logicaldrive")
+
+        ld_status = {}
+        for line in result.data.splitlines():
+            line = line.strip()
+            if not line or innocuous_errors.search(line) or not drive_status_line.search(line):
+                continue
+            drive_id = line.split()[1]
+            status = line.split("):")[1].lstrip().upper()
+            ld_status[drive_id] = status.strip().upper()
+
+        return ld_status
+
+    def _get_pd_status(self, slot: str) -> Dict[str, str]:
+        """Get physical drives status.
+
+        Returns:
+            pd_status: dict where each item is drive_id, status pair.
+        """
+        result = self(f"ctrl slot={slot} pd all show status")
+        if result.error:
+            logger.error(result.error)
+            return {}
+
+        innocuous_errors = re.compile(
+            r"^Error: The specified (device|controller) does not have any physical"
+        )
+        drive_status_line = re.compile(r"^\s*physicaldrive")
+        pd_status = {}
+        for line in result.data.splitlines():
+            line = line.strip()
+            if not line or innocuous_errors.search(line) or not drive_status_line.search(line):
+                continue
+            drive_id = line.split()[1]
+            status = line.split("):")[1].lstrip().upper()
+            pd_status[drive_id] = status.strip().upper()
+
+        return pd_status

tests/unit/test_collector.py

@@ -8,6 +8,7 @@
     LSISASControllerCollector,
     MegaRAIDCollector,
     PowerEdgeRAIDCollector,
+    SsaCLICollector,
 )
 
 SAMPLE_IPMI_SEL_ENTRIES = [
@@ -475,6 +476,33 @@ def test_51_ipmimonitoring_installed_and_okay(self):
         for payload in payloads:
             self.assertIn(payload.name, available_metrics)
 
+    def test_60_ssacli_not_installed(self):
+        ssacli_collector = SsaCLICollector()
+        ssacli_collector.ssacli = Mock()
+        ssacli_collector.ssacli.installed = False
+        ssacli_collector.ssacli.get_payload.return_value = {}
+        payloads = ssacli_collector.collect()
+
+        self.assertEqual(len(list(payloads)), 1)
+
+    def test_61_ssacli_installed_and_okay(self):
+        ssacli_collector = SsaCLICollector()
+        ssacli_collector.ssacli = Mock()
+        mock_payload = {
+            "2": {
+                "controller_status": {"Battery/Capacitor Status": " OK"},
+                "ld_status": {"1": "OK"},
+                "pd_status": {"2I:0:1": "corrupt"},
+            }
+        }
+        ssacli_collector.ssacli.get_payload.return_value = mock_payload
+        payloads = ssacli_collector.collect()
+
+        available_metrics = [spec.name for spec in ssacli_collector.specifications]
+        self.assertEqual(len(list(payloads)), len(available_metrics))
+        for payload in payloads:
+            self.assertIn(payload.name, available_metrics)
+
     def test_101_perccli_collector_command_success(self):
         with patch.object(PowerEdgeRAIDCollector, "perccli") as mock_cli:
             # 1 success, 1 fail

tests/unit/test_resources/ssacli/sample_outputs.py

@@ -0,0 +1,26 @@
+CTRL_ALL_SHOW = """
+
+Smart Array P222 in Slot 2                (sn: PDSXH0BRH6G0QU)
+
+"""
+CTRL_SHOW_STATUS = """
+
+Smart Array P222 in Slot 2
+   Random bad output line
+   Controller Status: OK
+   Cache Status: OK
+   Battery/Capacitor Status: OK
+
+
+"""
+CTRL_LD_ALL_SHOW_STATUS = """
+
+   logicaldrive 1 (931.48 GB, RAID 1): OK
+
+"""
+CTRL_PD_ALL_SHOW_STATUS = """
+
+   physicaldrive 2I:0:1 (port 2I:box 0:bay 1, 1 TB): OK
+   physicaldrive 2I:0:2 (port 2I:box 0:bay 2, 1 TB): OK
+
+"""