Make journal timestamps timezone-aware

nodescraper/plugins/inband/journal/journal_analyzer.py

@@ -23,7 +23,7 @@
 # SOFTWARE.
 #
 ###############################################################################
-from datetime import datetime
+from datetime import datetime, timezone
 from typing import Optional, TypedDict
 
 from nodescraper.enums import EventCategory, EventPriority, ExecutionStatus
@@ -74,6 +74,14 @@ def filter_journal(
             list[JournalJsonEntry]: filtered journal log
         """
 
+        def as_utc(value: Optional[datetime]) -> Optional[datetime]:
+            if value is not None and value.tzinfo is None:
+                return value.replace(tzinfo=timezone.utc)
+            return value
+
+        analysis_range_start = as_utc(analysis_range_start)
+        analysis_range_end = as_utc(analysis_range_end)
+
         filtered_journal = []
 
         found_start = False if analysis_range_start else True
@@ -141,25 +149,20 @@ def _analyze_journal_entries_by_priority(
                         console_log=False,
                     )
                 else:
-                    # Handle MESSAGE as either string or list
-                    message = entry.MESSAGE
-                    if isinstance(message, list):
-                        message = " ".join(message)
-
-                    # Create a tuple key from message and priority
-                    entry_key = (message, entry.PRIORITY)
+                    entry_key = (entry.MESSAGE, entry.PRIORITY)
+                    missing_timestamp = datetime.fromtimestamp(0, tz=timezone.utc)
                     if journal_event_map.get(entry_key) is None:
                         journal_event_map[entry_key] = {
                             "count": 1,
                             "first_occurrence": (
                                 entry.REALTIME_TIMESTAMP
                                 if entry.REALTIME_TIMESTAMP
-                                else datetime.fromtimestamp(0)
+                                else missing_timestamp
                             ),
                             "last_occurrence": (
                                 entry.REALTIME_TIMESTAMP
                                 if entry.REALTIME_TIMESTAMP
-                                else datetime.fromtimestamp(0)
+                                else missing_timestamp
                             ),
                         }
                     else:

nodescraper/plugins/inband/journal/journaldata.py

@@ -24,10 +24,10 @@
 #
 ###############################################################################
 import os
-from datetime import datetime
+from datetime import datetime, timezone
 from typing import Optional, Union
 
-from pydantic import ConfigDict, Field, field_validator
+from pydantic import AwareDatetime, ConfigDict, Field, field_validator
 
 from nodescraper.models import DataModel
 
@@ -42,35 +42,34 @@ class JournalJsonEntry(DataModel):
     SYSLOG_IDENTIFIER: Optional[str] = Field(None, alias="SYSLOG_IDENTIFIER")
     CURSOR: Optional[str] = Field(None, alias="__CURSOR")
     SYSLOG_FACILITY: Optional[int] = Field(None, alias="SYSLOG_FACILITY")
-    SOURCE_REALTIME_TIMESTAMP: Optional[datetime] = Field(None, alias="_SOURCE_REALTIME_TIMESTAMP")
-    REALTIME_TIMESTAMP: Optional[datetime] = Field(None, alias="__REALTIME_TIMESTAMP")
+    SOURCE_REALTIME_TIMESTAMP: Optional[AwareDatetime] = Field(
+        None, alias="_SOURCE_REALTIME_TIMESTAMP"
+    )
+    REALTIME_TIMESTAMP: Optional[AwareDatetime] = Field(None, alias="__REALTIME_TIMESTAMP")
     PRIORITY: int = Field(default=7, alias="PRIORITY")  # Default to DEBUG (7) if not present
     BOOT_ID: Optional[str] = Field(None, alias="_BOOT_ID")
     SOURCE_MONOTONIC_TIMESTAMP: Optional[float] = Field(None, alias="_SOURCE_MONOTONIC_TIMESTAMP")
     MONOTONIC_TIMESTAMP: Optional[float] = Field(None, alias="__MONOTONIC_TIMESTAMP")
-    MESSAGE: Union[str, list[str]] = Field(default="", alias="MESSAGE")
+    MESSAGE: str = Field(default="", alias="MESSAGE")
 
-    # assume datetime has microseconds
-    @field_validator("SOURCE_REALTIME_TIMESTAMP", mode="before")
-    @classmethod
-    def validate_source_realtime_timestamp(cls, v):
-        if v is None:
-            return None
-        try:
-            return datetime.fromisoformat(v)
-        except ValueError:
-            return datetime.fromtimestamp(float(v) / 1e6)
-
-    # assume datetime has microseconds
-    @field_validator("REALTIME_TIMESTAMP", mode="before")
+    @field_validator("SOURCE_REALTIME_TIMESTAMP", "REALTIME_TIMESTAMP", mode="before")
     @classmethod
     def validate_realtime_timestamp(cls, v):
+        """Return a UTC datetime."""
         if v is None:
             return None
-        try:
-            return datetime.fromisoformat(v)
-        except ValueError:
-            return datetime.fromtimestamp(float(v) / 1e6)
+
+        if isinstance(v, datetime):
+            parsed = v
+        else:
+            try:
+                parsed = datetime.fromisoformat(str(v))
+            except ValueError:
+                return datetime.fromtimestamp(float(v) / 1e6, tz=timezone.utc)
+
+        if parsed.tzinfo is None:
+            return parsed.replace(tzinfo=timezone.utc)
+        return parsed.astimezone(timezone.utc)
 
     @field_validator("SOURCE_MONOTONIC_TIMESTAMP", mode="before")
     @classmethod
@@ -113,16 +112,18 @@ def validate_syslog_facility(cls, v):
     @field_validator("MESSAGE", mode="before")
     @classmethod
     def validate_message(cls, v):
-        """Convert MESSAGE field to string or list[str] based on input type"""
-        if v is None:
-            return ""
-
-        # If it's a list but not all items are strings, convert each item to string
-        if isinstance(v, list):
-            return [str(item) for item in v]
-
-        # Return string representation for any other type
-        return str(v)
+        """Convert MESSAGE to text, including journald's JSON byte-array form."""
+
+        def to_text(item) -> str:
+            if item is None:
+                return ""
+            if isinstance(item, list):
+                return bytes(item).decode("utf-8", errors="replace")
+            return str(item)
+
+        if isinstance(v, list) and not all(isinstance(b, int) for b in v):
+            return " ".join(to_text(item) for item in v)
+        return to_text(v)
 
 
 class JournalData(DataModel):

test/unit/plugin/test_journal_collector.py

@@ -24,7 +24,7 @@
 #
 ###############################################################################
 import types
-from datetime import datetime
+from datetime import datetime, timezone
 
 from nodescraper.enums import ExecutionStatus
 from nodescraper.enums.systeminteraction import SystemInteractionLevel
@@ -75,6 +75,46 @@ def run_map(cmd, **kwargs):
     assert data.journal_log == '{"MESSAGE":"hello"}\n'
 
 
+def test_journal_entry_timestamps_are_utc_aware():
+    entry = JournalJsonEntry(
+        **{
+            "_SOURCE_REALTIME_TIMESTAMP": "0",
+            "__REALTIME_TIMESTAMP": "2025-02-23T21:03:46.685975",
+        }
+    )
+
+    assert entry.SOURCE_REALTIME_TIMESTAMP == datetime(1970, 1, 1, tzinfo=timezone.utc)
+    assert entry.REALTIME_TIMESTAMP == datetime(2025, 2, 23, 21, 3, 46, 685975, tzinfo=timezone.utc)
+
+
+def test_journal_entry_decodes_non_utf8_message_byte_array():
+    entry = JournalJsonEntry(MESSAGE=[104, 195, 169, 255])
+
+    assert entry.MESSAGE == "hé�"
+
+
+def test_journal_entry_joins_repeated_message_values():
+    strings = JournalJsonEntry(MESSAGE=["first", "second"])
+    mixed = JournalJsonEntry(MESSAGE=[[104, 105], "x"])
+    with_null = JournalJsonEntry(MESSAGE=["first line", [104, 105], None])
+
+    assert strings.MESSAGE == "first second"
+    assert mixed.MESSAGE == "hi x"
+    assert with_null.MESSAGE == "first line hi "
+
+
+def test_journal_entry_converts_offset_and_datetime_inputs_to_utc():
+    offset = JournalJsonEntry(**{"__REALTIME_TIMESTAMP": "2025-02-23T21:03:46+05:00"})
+    assert offset.REALTIME_TIMESTAMP == datetime(2025, 2, 23, 16, 3, 46, tzinfo=timezone.utc)
+
+    naive = JournalJsonEntry(**{"__REALTIME_TIMESTAMP": datetime(2025, 2, 23, 21, 3, 46)})
+    assert naive.REALTIME_TIMESTAMP == datetime(2025, 2, 23, 21, 3, 46, tzinfo=timezone.utc)
+
+    aware = datetime(2025, 2, 23, 16, 3, 46, tzinfo=timezone.utc)
+    from_aware = JournalJsonEntry(**{"__REALTIME_TIMESTAMP": aware})
+    assert from_aware.REALTIME_TIMESTAMP == aware
+
+
 def test_journal_filter():
     """Test filtering journal entries based on timestamp range."""
     journal_data = [
@@ -142,8 +182,8 @@ def test_journal_filter():
 
     journal_content_json = [JournalJsonEntry(**entry) for entry in journal_data]
 
-    start_range = datetime.fromisoformat("2025-02-23T21:03:46.686006")
-    end_range = datetime.fromisoformat("2025-02-23T21:03:46.686020")
+    start_range = datetime.fromisoformat("2025-02-23T21:03:46.686006").replace(tzinfo=timezone.utc)
+    end_range = datetime.fromisoformat("2025-02-23T21:03:46.686020").replace(tzinfo=timezone.utc)
 
     filtered_journal = JournalAnalyzer.filter_journal(journal_content_json, start_range, end_range)
     assert filtered_journal == [JournalJsonEntry(**entry) for entry in journal_data[1:3]]
@@ -154,6 +194,11 @@ def test_journal_filter():
     filtered_journal = JournalAnalyzer.filter_journal(journal_content_json, None, end_range)
     assert filtered_journal == [JournalJsonEntry(**entry) for entry in journal_data[:3]]
 
+    naive_start = datetime.fromisoformat("2025-02-23T21:03:46.686006")
+    naive_end = datetime.fromisoformat("2025-02-23T21:03:46.686020")
+    filtered_journal = JournalAnalyzer.filter_journal(journal_content_json, naive_start, naive_end)
+    assert filtered_journal == [JournalJsonEntry(**entry) for entry in journal_data[1:3]]
+
 
 def test_check_priority(system_info):
     """Test checking priority of journal entries."""
@@ -230,3 +275,18 @@ def test_check_priority(system_info):
 
     assert res.status == ExecutionStatus.ERROR
     assert len(res.events) == 3
+
+
+def test_grouped_missing_timestamp_uses_utc_epoch(system_info):
+    journal_data = JournalData(
+        journal_log="",
+        journal_content_json=[
+            JournalJsonEntry(MESSAGE="no time", PRIORITY=3, REALTIME_TIMESTAMP=None)
+        ],
+    )
+    analyzer = JournalAnalyzer(system_info=system_info)
+    res = analyzer.analyze_data(data=journal_data, args=JournalAnalyzerArgs(check_priority=5))
+
+    assert res.status == ExecutionStatus.ERROR
+    assert res.events[0].data["first_occurrence"] == datetime(1970, 1, 1, tzinfo=timezone.utc)
+    assert res.events[0].data["last_occurrence"] == datetime(1970, 1, 1, tzinfo=timezone.utc)