reading logs

Author: alexandraBara

nodescraper/plugins/inband/dmesg/dmesg_collector.py

@@ -23,7 +23,10 @@
 # SOFTWARE.
 #
 ###############################################################################
+import re
+
 from nodescraper.base import InBandDataCollector
+from nodescraper.connection.inband import FileArtifact
 from nodescraper.enums import EventCategory, EventPriority, OSFamily
 from nodescraper.models import TaskResult
 
@@ -39,6 +42,84 @@ class DmesgCollector(InBandDataCollector[DmesgData, None]):
 
     DMESG_CMD = "dmesg --time-format iso -x"
 
+    DMESG_LIST_CMD = (
+        r"ls -1 /var/log/dmesg /var/log/dmesg.1 /var/log/dmesg.[0-9]*.gz 2>/dev/null || true"
+    )
+
+    def _shell_quote(self, s: str) -> str:
+        """POSIX single-quote."""
+        return "'" + s.replace("'", "'\"'\"'") + "'"
+
+    def _nice_dmesg_name(self, path: str) -> str:
+        """Map path to filename"""
+        if path.endswith("/dmesg"):
+            return "dmesg.log"
+        if path.endswith("/dmesg.1"):
+            return "dmesg.1.log"
+        m = re.search(r"/dmesg\.(\d+)\.gz$", path)
+        if m:
+            return f"dmesg.{m.group(1)}.log"
+        base = path.rsplit("/", 1)[-1]
+        return base.replace(".gz", "") + ".log"
+
+    def _collect_dmesg_rotations(self) -> int:
+        """
+        Collect /var/log/dmesg, /var/log/dmesg.1, and /var/log/dmesg.N.gz (decompressed).
+        Attaches each as a text artifact.
+
+        Returns:
+             list: list of logs collected.
+
+        """
+        list_res = self._run_sut_cmd(self.DMESG_LIST_CMD)
+        paths = [p.strip() for p in (list_res.stdout or "").splitlines() if p.strip()]
+        if not paths:
+            self._log_event(
+                category=EventCategory.OS,
+                description="No /var/log/dmesg files found (including rotations).",
+                data={"list_exit_code": list_res.exit_code},
+                priority=EventPriority.WARNING,
+            )
+            return 0
+
+        collected, failed = [], []
+        for p in paths:
+            qp = self._shell_quote(p)
+            if p.endswith(".gz"):
+                cmd = f"(command -v gzip >/dev/null && gzip -dc {qp}) || (command -v zcat >/dev/null && zcat {qp}) || cat {qp}"
+            else:
+                cmd = f"cat {qp}"
+
+            res = self._run_sut_cmd(cmd)
+            if res.exit_code == 0 and res.stdout is not None:
+                fname = self._nice_dmesg_name(p)
+                self.result.artifacts.append(FileArtifact(filename=fname, contents=res.stdout))
+                collected.append(
+                    {"path": p, "as": fname, "bytes": len(res.stdout.encode("utf-8", "ignore"))}
+                )
+            else:
+                failed.append({"path": p, "exit_code": res.exit_code, "stderr": res.stderr})
+
+        if collected:
+            self._log_event(
+                category=EventCategory.OS,
+                description="Collected dmesg rotated files",
+                data={"collected": collected},
+                priority=EventPriority.INFO,
+            )
+            # self.result.status = self.result.status or ExecutionStatus.OK
+            self.result.message = self.result.message or "dmesg rotated files collected"
+
+        if failed:
+            self._log_event(
+                category=EventCategory.OS,
+                description="Some dmesg files could not be collected.",
+                data={"failed": failed},
+                priority=EventPriority.WARNING,
+            )
+
+        return collected
+
     def _get_dmesg_content(self) -> str:
         """run dmesg command on system and return output
 
@@ -68,6 +149,7 @@ def collect_data(
             tuple[TaskResult, DmesgData | None]: tuple containing the result of the task and the dmesg data if available
         """
         dmesg_content = self._get_dmesg_content()
+        _ = self._collect_dmesg_rotations()
 
         if dmesg_content:
             dmesg_data = DmesgData(dmesg_content=dmesg_content)

nodescraper/plugins/inband/dmesg/dmesgdata.py

@@ -34,6 +34,7 @@ class DmesgData(DataModel):
     """Data model for in band dmesg log"""
 
     dmesg_content: str
+    dmesg_logs: list[dict] = None
 
     @classmethod
     def get_new_dmesg_lines(cls, current_dmesg: str, new_dmesg: str) -> str: