diff --git a/nodescraper/plugins/inband/journal/journal_analyzer.py b/nodescraper/plugins/inband/journal/journal_analyzer.py index fbf14eaf..24887f0a 100644 --- a/nodescraper/plugins/inband/journal/journal_analyzer.py +++ b/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: diff --git a/nodescraper/plugins/inband/journal/journaldata.py b/nodescraper/plugins/inband/journal/journaldata.py index a6639422..fdc25e1f 100644 --- a/nodescraper/plugins/inband/journal/journaldata.py +++ b/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): diff --git a/test/unit/plugin/test_journal_collector.py b/test/unit/plugin/test_journal_collector.py index 4865baa2..0efe9b5d 100644 --- a/test/unit/plugin/test_journal_collector.py +++ b/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)