From fb5570155be2a2433dc558a9f407cc2f45ba9e61 Mon Sep 17 00:00:00 2001 From: Andreas Wrede Date: Fri, 10 Jul 2026 14:47:42 -0400 Subject: [PATCH] feat: events journal write primitives (log_event, backfill, get_events_journal) Co-Authored-By: Claude Fable 5 --- hbd/server/journal.py | 58 ++++++++++++++++-- tests/test_events_journal.py | 110 +++++++++++++++++++++++++++++++++++ 2 files changed, 164 insertions(+), 4 deletions(-) create mode 100644 tests/test_events_journal.py diff --git a/hbd/server/journal.py b/hbd/server/journal.py index f359a4e..24f6770 100644 --- a/hbd/server/journal.py +++ b/hbd/server/journal.py @@ -11,7 +11,7 @@ import os import asyncio from datetime import datetime from pathlib import Path -from typing import Dict, Any, Optional +from typing import Dict, Any, Optional, Callable, Iterable, List, Tuple, Union logger = logging.getLogger(__name__) @@ -149,10 +149,35 @@ class MessageJournal: self._current_size += len(json_bytes) logger.debug(f"Logged message from {addr[0]}: {msg.get('ID', 'UNKNOWN')}") - + except Exception as e: logger.error(f"Error writing to journal: {e}") - + + async def log_event(self, event: Dict[str, Any]): + """Write a caller-provided dict verbatim as one JSONL line (with rotation).""" + if not self.enabled or not self._initialized: + return + async with self._lock: + try: + line = json.dumps(event, separators=(',', ':')) + '\n' + nbytes = len(line.encode('utf-8')) + if self._current_size + nbytes > self.max_size: + await self._rotate() + if self._file_handle: + self._file_handle.write(line) + self._file_handle.flush() + self._current_size += nbytes + except Exception as e: + logger.error(f"Error writing event to journal: {e}") + + async def backfill(self, events: Iterable[Dict[str, Any]]): + """One-time seed: write *events* only if the journal file is currently empty.""" + if not self.enabled or not self._initialized or self._current_size > 0: + return + for ev in events: + if isinstance(ev, dict): + await self.log_event(ev) + async def _rotate(self): """ Rotate the journal file. @@ -332,7 +357,7 @@ def get_journal(config: Optional[Dict[str, Any]] = None) -> MessageJournal: async def log_message(msg: Dict[str, Any], addr: tuple, timestamp: Optional[float] = None): """ Convenience function to log a message using the global journal. - + Args: msg: Parsed message dictionary addr: Source address (ip, port) tuple @@ -340,3 +365,28 @@ async def log_message(msg: Dict[str, Any], addr: tuple, timestamp: Optional[floa """ journal = get_journal() await journal.log_message(msg, addr, timestamp) + + +# Global events journal instance (human-readable event log, written by notify.eventlog) +_events_journal_instance: Optional[MessageJournal] = None + + +def get_events_journal(config: Optional[Dict[str, Any]] = None) -> MessageJournal: + """Get or create the global events journal instance. + + Uses the events_journal_* config keys; shares journal_dir and + journal_enabled with the raw-datagram journal. + """ + global _events_journal_instance + if _events_journal_instance is None: + cfg = config or {} + _events_journal_instance = MessageJournal( + { + 'journal_dir': cfg.get('journal_dir', '/var/log/heartbeat'), + 'journal_file': cfg.get('events_journal_file', 'events.journal'), + 'journal_max_size': cfg.get('events_journal_max_size', 10 * 1024 * 1024), + 'journal_max_backups': cfg.get('events_journal_max_backups', 10), + 'journal_enabled': cfg.get('journal_enabled', True), + } + ) + return _events_journal_instance diff --git a/tests/test_events_journal.py b/tests/test_events_journal.py new file mode 100644 index 0000000..fd8b257 --- /dev/null +++ b/tests/test_events_journal.py @@ -0,0 +1,110 @@ +"""Tests for the dedicated events journal (write path and read path).""" +import asyncio +import json + +from hbd.server import journal + + +def _make_journal(tmp_path, **overrides): + cfg = {"journal_dir": str(tmp_path), "journal_file": "events.journal"} + cfg.update(overrides) + j = journal.MessageJournal(cfg) + assert asyncio.run(j.initialize()) + return j + + +def _read_lines(tmp_path, name="events.journal"): + return (tmp_path / name).read_text(encoding="utf-8").splitlines() + + +EV1 = {"ts": 1000.0, "host": "h1", "level": "INFO", "service": None, "message": "host up"} +EV2 = {"ts": 2000.0, "host": "h2", "level": "CRITICAL", "service": "cpu", "message": "cpu high"} + + +def test_log_event_writes_one_json_line(tmp_path): + j = _make_journal(tmp_path) + asyncio.run(j.log_event(EV1)) + asyncio.run(j.close()) + lines = _read_lines(tmp_path) + assert len(lines) == 1 + assert json.loads(lines[0]) == EV1 + + +def test_log_event_appends_in_order(tmp_path): + j = _make_journal(tmp_path) + asyncio.run(j.log_event(EV1)) + asyncio.run(j.log_event(EV2)) + asyncio.run(j.close()) + lines = _read_lines(tmp_path) + assert [json.loads(ln)["ts"] for ln in lines] == [1000.0, 2000.0] + + +def test_log_event_rotates_at_max_size(tmp_path): + # max_size fits one serialized event line (~77 bytes) but not two, so the + # second write triggers exactly one rotation + j = _make_journal(tmp_path, journal_max_size=120) + asyncio.run(j.log_event(EV1)) + asyncio.run(j.log_event(EV2)) + asyncio.run(j.close()) + backups = list(tmp_path.glob("events.journal.*")) + assert len(backups) == 1 + assert json.loads(backups[0].read_text().splitlines()[0]) == EV1 + assert json.loads(_read_lines(tmp_path)[0]) == EV2 + + +def test_log_event_noop_when_disabled(tmp_path): + j = journal.MessageJournal( + {"journal_dir": str(tmp_path), "journal_file": "events.journal", "journal_enabled": False} + ) + asyncio.run(j.initialize()) + asyncio.run(j.log_event(EV1)) + assert not (tmp_path / "events.journal").exists() + + +def test_backfill_seeds_empty_journal(tmp_path): + j = _make_journal(tmp_path) + asyncio.run(j.backfill([EV1, EV2])) + asyncio.run(j.close()) + lines = _read_lines(tmp_path) + assert len(lines) == 2 + assert json.loads(lines[0]) == EV1 + + +def test_backfill_skipped_when_journal_nonempty(tmp_path): + (tmp_path / "events.journal").write_text(json.dumps(EV1) + "\n") + j = _make_journal(tmp_path) # initialize() picks up the existing size + asyncio.run(j.backfill([EV2])) + asyncio.run(j.close()) + assert len(_read_lines(tmp_path)) == 1 + + +def test_get_events_journal_uses_events_config_keys(tmp_path): + journal._events_journal_instance = None + try: + ej = journal.get_events_journal( + { + "journal_dir": str(tmp_path), + "events_journal_file": "ev.jsonl", + "events_journal_max_size": 12345, + "events_journal_max_backups": 3, + } + ) + assert ej.journal_file == "ev.jsonl" + assert ej.max_size == 12345 + assert ej.max_backups == 3 + assert ej.journal_dir == tmp_path + # singleton: second call returns the same instance + assert journal.get_events_journal() is ej + finally: + journal._events_journal_instance = None + + +def test_get_events_journal_defaults(tmp_path): + journal._events_journal_instance = None + try: + ej = journal.get_events_journal({"journal_dir": str(tmp_path)}) + assert ej.journal_file == "events.journal" + assert ej.max_size == 10 * 1024 * 1024 + assert ej.max_backups == 10 + finally: + journal._events_journal_instance = None