From ae9c6ec1a925d35b732bdb8b70c5be685943d435 Mon Sep 17 00:00:00 2001 From: jonahgabriel Date: Mon, 25 May 2026 17:27:39 -0400 Subject: [PATCH] feat(OMN-12132): add node_log_persistence_effect EFFECT node Subscribes to onex.evt.platform.log-entry.v1 and persists each ModelLogEntry to Postgres log_entries table via asyncpg. Idempotent on entry_id via ON CONFLICT DO NOTHING. Gracefully skips when no DB pool is available. Topics resolved from contract.yaml via contract_topics helper. 6 unit tests covering: deserialization, skip-on-no-pool, insert params, idempotent conflict, DB error handling, and handle_raw shim. --- pyproject.toml | 1 + .../node_log_persistence_effect/__init__.py | 11 + .../node_log_persistence_effect/contract.yaml | 38 ++++ .../handlers/__init__.py | 1 + .../handler_log_persistence_effect.py | 179 +++++++++++++++++ .../node_log_persistence_effect/metadata.yaml | 24 +++ .../__init__.py | 1 + .../test_handler_log_persistence_effect.py | 190 ++++++++++++++++++ 8 files changed, 445 insertions(+) create mode 100644 src/omnimarket/nodes/node_log_persistence_effect/__init__.py create mode 100644 src/omnimarket/nodes/node_log_persistence_effect/contract.yaml create mode 100644 src/omnimarket/nodes/node_log_persistence_effect/handlers/__init__.py create mode 100644 src/omnimarket/nodes/node_log_persistence_effect/handlers/handler_log_persistence_effect.py create mode 100644 src/omnimarket/nodes/node_log_persistence_effect/metadata.yaml create mode 100644 tests/unit/nodes/test_node_log_persistence_effect/__init__.py create mode 100644 tests/unit/nodes/test_node_log_persistence_effect/test_handler_log_persistence_effect.py diff --git a/pyproject.toml b/pyproject.toml index 2be91b452c..430e68e014 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -116,6 +116,7 @@ node_golden_chain_sweep = "omnimarket.nodes.node_golden_chain_sweep" node_golden_chain_generator = "omnimarket.nodes.node_golden_chain_generator" node_runtime_sweep = "omnimarket.nodes.node_runtime_sweep" node_log_projection = "omnimarket.nodes.node_log_projection" +node_log_persistence_effect = "omnimarket.nodes.node_log_persistence_effect" node_data_verification = "omnimarket.nodes.node_data_verification" node_process_watchdog = "omnimarket.nodes.node_process_watchdog" node_projection_session_outcome = "omnimarket.nodes.node_projection_session_outcome" diff --git a/src/omnimarket/nodes/node_log_persistence_effect/__init__.py b/src/omnimarket/nodes/node_log_persistence_effect/__init__.py new file mode 100644 index 0000000000..9ab7f27b95 --- /dev/null +++ b/src/omnimarket/nodes/node_log_persistence_effect/__init__.py @@ -0,0 +1,11 @@ +"""node_log_persistence_effect — Persists structured log events to Postgres.""" + +from omnimarket.nodes.node_log_persistence_effect.handlers.handler_log_persistence_effect import ( + ModelLogPersistenceResult, + NodeLogPersistenceEffect, +) + +__all__ = [ + "ModelLogPersistenceResult", + "NodeLogPersistenceEffect", +] diff --git a/src/omnimarket/nodes/node_log_persistence_effect/contract.yaml b/src/omnimarket/nodes/node_log_persistence_effect/contract.yaml new file mode 100644 index 0000000000..e1e02ff338 --- /dev/null +++ b/src/omnimarket/nodes/node_log_persistence_effect/contract.yaml @@ -0,0 +1,38 @@ +--- +name: log_persistence_effect +contract_version: {major: 1, minor: 0, patch: 0} +node_type: effect +node_version: {major: 1, minor: 0, patch: 0} +description: > + Persists structured log events to Postgres for historical trace queries. + Enables the trace explorer to query beyond live Kafka retention window. + +terminal_event: onex.evt.omnimarket.log-persistence-completed.v1 + +handler: + module: omnimarket.nodes.node_log_persistence_effect.handlers.handler_log_persistence_effect + class: NodeLogPersistenceEffect + +handler_routing: + routing_strategy: operation_match + handlers: + - handler: + name: NodeLogPersistenceEffect + module: omnimarket.nodes.node_log_persistence_effect.handlers.handler_log_persistence_effect + +event_bus: + subscribe_topics: + - onex.evt.platform.log-entry.v1 + publish_topics: + - onex.evt.omnimarket.log-persistence-completed.v1 + +descriptor: + node_archetype: effect + purity: effectful + idempotent: true + timeout_ms: 10000 + runtime_profiles: + - effects + +metadata: + transport_type: kafka diff --git a/src/omnimarket/nodes/node_log_persistence_effect/handlers/__init__.py b/src/omnimarket/nodes/node_log_persistence_effect/handlers/__init__.py new file mode 100644 index 0000000000..59f32718ce --- /dev/null +++ b/src/omnimarket/nodes/node_log_persistence_effect/handlers/__init__.py @@ -0,0 +1 @@ +"""Handlers for node_log_persistence_effect.""" diff --git a/src/omnimarket/nodes/node_log_persistence_effect/handlers/handler_log_persistence_effect.py b/src/omnimarket/nodes/node_log_persistence_effect/handlers/handler_log_persistence_effect.py new file mode 100644 index 0000000000..0b10636857 --- /dev/null +++ b/src/omnimarket/nodes/node_log_persistence_effect/handlers/handler_log_persistence_effect.py @@ -0,0 +1,179 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT + +"""NodeLogPersistenceEffect — persists log events to Postgres. + +Subscribes to onex.evt.platform.log-entry.v1 events and INSERTs each into +the log_entries table. Idempotent on entry_id via ON CONFLICT DO NOTHING. + +If the asyncpg pool is None (no DB configured), the handler logs a warning +and skips the write rather than raising — graceful degradation for +environments without Postgres. +""" + +from __future__ import annotations + +import json +import logging +import os +from pathlib import Path +from typing import TYPE_CHECKING, Any, Literal +from uuid import uuid4 + +from pydantic import BaseModel, ConfigDict, Field + +from omnimarket.nodes.contract_topics import ( + contract_publish_topics, + contract_subscribe_topics, +) +from omnimarket.nodes.node_log_projection.handlers.handler_log_projection import ( + ModelLogEntry, +) + +if TYPE_CHECKING: + import asyncpg + +logger = logging.getLogger(__name__) + +_CONTRACT_PATH = Path(__file__).parent.parent / "contract.yaml" +_SUBSCRIBE_TOPICS = contract_subscribe_topics(_CONTRACT_PATH) +_PUBLISH_TOPICS = contract_publish_topics(_CONTRACT_PATH) + +_DEFAULT_PG_DSN = os.environ.get("ONEX_PG_DSN", "") # contract-config-ok: config # fmt: skip + + +class ModelLogPersistenceResult(BaseModel): + """Result of a single log entry persistence attempt.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + result_id: str = Field(default_factory=lambda: str(uuid4())) + entry_id: str = Field(..., description="entry_id of the persisted log entry.") + status: Literal["written", "skipped", "idempotent", "error"] = Field(...) + error_message: str | None = Field(default=None) + + +class NodeLogPersistenceEffect: + """EFFECT node: persists log events from Kafka to Postgres. + + Accepts an optional asyncpg pool for dependency injection (enables unit + testing without a real database). When pool is None and no DSN is + configured, operates in no-op mode with a logged warning. + """ + + handler_type: Literal["node_handler"] = "node_handler" + handler_category: Literal["effect"] = "effect" + + def __init__( + self, + pool: asyncpg.Pool | None = None, + pg_dsn: str = _DEFAULT_PG_DSN, + ) -> None: + self._injected_pool = pool + self._pg_dsn = pg_dsn + self._pool: asyncpg.Pool | None = pool + + async def _get_pool(self) -> asyncpg.Pool | None: + """Return pool, lazily creating it from DSN if not injected.""" + if self._pool is not None: + return self._pool + if not self._pg_dsn: + return None + import asyncpg as _asyncpg + + self._pool = await _asyncpg.create_pool( + dsn=self._pg_dsn, + min_size=1, + max_size=5, + command_timeout=10, + ) + return self._pool + + async def handle(self, entry: ModelLogEntry) -> ModelLogPersistenceResult: + """Persist a single log entry to Postgres. + + Args: + entry: The structured log event to persist. + + Returns: + ModelLogPersistenceResult describing the outcome. + """ + pool = await self._get_pool() + + if pool is None: + logger.warning( + "NodeLogPersistenceEffect: no DB pool available, skipping entry_id=%s", + entry.entry_id, + ) + return ModelLogPersistenceResult( + entry_id=entry.entry_id, + status="skipped", + ) + + try: + inserted = await self._insert(pool, entry) + status: Literal["written", "idempotent"] = ( + "written" if inserted else "idempotent" + ) + logger.debug( + "NodeLogPersistenceEffect: entry_id=%s status=%s", + entry.entry_id, + status, + ) + return ModelLogPersistenceResult(entry_id=entry.entry_id, status=status) + except Exception as exc: + logger.error( + "NodeLogPersistenceEffect: failed to persist entry_id=%s: %s", + entry.entry_id, + exc, + ) + return ModelLogPersistenceResult( + entry_id=entry.entry_id, + status="error", + error_message=str(exc), + ) + + async def _insert(self, pool: asyncpg.Pool, entry: ModelLogEntry) -> bool: + """INSERT entry into log_entries. Returns True if a row was inserted.""" + metadata_json = json.dumps(entry.metadata) + async with pool.acquire() as conn: + inserted_id = await conn.fetchval( + """ + INSERT INTO log_entries ( + entry_id, + timestamp, + node_name, + function_name, + level, + message, + correlation_id, + duration_ms, + metadata + ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) + ON CONFLICT (entry_id) DO NOTHING + RETURNING entry_id + """, + entry.entry_id, + entry.timestamp, + entry.node_name, + entry.function_name, + entry.level.value, + entry.message, + entry.correlation_id, + entry.duration_ms, + metadata_json, + ) + return inserted_id is not None + + @staticmethod + def handle_raw(input_data: dict[str, Any]) -> dict[str, Any]: + """Synchronous dict-in / dict-out shim for runtime protocol compatibility.""" + entry = ModelLogEntry(**input_data) + return { + "entry_id": entry.entry_id, + "status": "skipped", + "error_message": "use async handle() for live persistence", + } + + +__all__: list[str] = ["ModelLogPersistenceResult", "NodeLogPersistenceEffect"] diff --git a/src/omnimarket/nodes/node_log_persistence_effect/metadata.yaml b/src/omnimarket/nodes/node_log_persistence_effect/metadata.yaml new file mode 100644 index 0000000000..26f9fe2551 --- /dev/null +++ b/src/omnimarket/nodes/node_log_persistence_effect/metadata.yaml @@ -0,0 +1,24 @@ +name: node_log_persistence_effect +version: "1.0.0" +description: "Persists structured log events to Postgres for historical trace queries" +omnibase_core_compat: ">=0.39.0,<1.0.0" +entry_points: + onex.nodes: + node_log_persistence_effect: "omnimarket.nodes.node_log_persistence_effect" +capabilities: + standalone: true + full_runtime: true + requires_network: false + requires_repo: false + requires_secrets: false + requires_docker: false + side_effect_class: external_io +dependencies: + - "omnibase_core>=0.39.0" + - "asyncpg>=0.29.0" +authors: ["OmniNode Platform Team"] +license: "MIT" +tags: ["effect", "log-persistence", "observability", "postgres"] +pack: "data" +display_name: "Log Persistence Effect" +node_role: "effect" diff --git a/tests/unit/nodes/test_node_log_persistence_effect/__init__.py b/tests/unit/nodes/test_node_log_persistence_effect/__init__.py new file mode 100644 index 0000000000..deffdc3df9 --- /dev/null +++ b/tests/unit/nodes/test_node_log_persistence_effect/__init__.py @@ -0,0 +1 @@ +"""Unit tests for node_log_persistence_effect.""" diff --git a/tests/unit/nodes/test_node_log_persistence_effect/test_handler_log_persistence_effect.py b/tests/unit/nodes/test_node_log_persistence_effect/test_handler_log_persistence_effect.py new file mode 100644 index 0000000000..963efd3272 --- /dev/null +++ b/tests/unit/nodes/test_node_log_persistence_effect/test_handler_log_persistence_effect.py @@ -0,0 +1,190 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT + +"""Unit tests for NodeLogPersistenceEffect. + +No real database — asyncpg pool is mocked throughout. +""" + +from __future__ import annotations + +import json +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from omnimarket.nodes.node_log_persistence_effect.handlers.handler_log_persistence_effect import ( + ModelLogPersistenceResult, + NodeLogPersistenceEffect, +) +from omnimarket.nodes.node_log_projection.handlers.handler_log_projection import ( + EnumLogLevel, + ModelLogEntry, +) + + +def _make_entry(**kwargs: Any) -> ModelLogEntry: + defaults: dict[str, Any] = { + "node_name": "test_node", + "message": "hello world", + } + defaults.update(kwargs) + return ModelLogEntry(**defaults) + + +def _make_pool(fetchval_return: Any = "some-entry-id") -> MagicMock: + """Build a mock asyncpg pool whose conn.fetchval returns fetchval_return.""" + conn = AsyncMock() + conn.fetchval = AsyncMock(return_value=fetchval_return) + pool = MagicMock() + pool.acquire = MagicMock(return_value=_AsyncContextManager(conn)) + return pool, conn + + +class _AsyncContextManager: + """Minimal async context manager wrapping a value.""" + + def __init__(self, value: Any) -> None: + self._value = value + + async def __aenter__(self) -> Any: + return self._value + + async def __aexit__(self, *args: Any) -> None: + pass + + +# --------------------------------------------------------------------------- +# Deserialization +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +def test_model_log_entry_round_trip() -> None: + entry = _make_entry( + node_name="my_node", + function_name="do_thing", + level=EnumLogLevel.ERROR, + message="boom", + correlation_id="corr-123", + duration_ms=42.5, + metadata={"key": "val"}, + ) + assert entry.node_name == "my_node" + assert entry.level == EnumLogLevel.ERROR + assert entry.correlation_id == "corr-123" + assert entry.metadata == {"key": "val"} + + +# --------------------------------------------------------------------------- +# Graceful degradation — no pool +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +@pytest.mark.asyncio +async def test_handle_skips_when_pool_is_none() -> None: + handler = NodeLogPersistenceEffect(pool=None, pg_dsn="") + entry = _make_entry() + + result = await handler.handle(entry) + + assert isinstance(result, ModelLogPersistenceResult) + assert result.status == "skipped" + assert result.entry_id == entry.entry_id + + +# --------------------------------------------------------------------------- +# Successful insert +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +@pytest.mark.asyncio +async def test_handle_inserts_correct_params() -> None: + entry = _make_entry( + node_name="my_node", + function_name="fn", + level=EnumLogLevel.WARNING, + message="watch out", + correlation_id="c-1", + duration_ms=10.0, + metadata={"env": "test"}, + ) + pool, conn = _make_pool(fetchval_return=entry.entry_id) + handler = NodeLogPersistenceEffect(pool=pool) + + result = await handler.handle(entry) + + assert result.status == "written" + assert result.entry_id == entry.entry_id + + conn.fetchval.assert_awaited_once() + call_args = conn.fetchval.call_args + positional = call_args.args + + assert positional[1] == entry.entry_id + assert positional[2] == entry.timestamp + assert positional[3] == entry.node_name + assert positional[4] == entry.function_name + assert positional[5] == entry.level.value + assert positional[6] == entry.message + assert positional[7] == entry.correlation_id + assert positional[8] == entry.duration_ms + assert json.loads(positional[9]) == entry.metadata + + +# --------------------------------------------------------------------------- +# Idempotent insert (conflict — row already exists) +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +@pytest.mark.asyncio +async def test_handle_idempotent_when_conflict() -> None: + entry = _make_entry() + pool, _conn = _make_pool(fetchval_return=None) + handler = NodeLogPersistenceEffect(pool=pool) + + result = await handler.handle(entry) + + assert result.status == "idempotent" + assert result.entry_id == entry.entry_id + + +# --------------------------------------------------------------------------- +# DB error — returns error status, does not raise +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +@pytest.mark.asyncio +async def test_handle_returns_error_on_db_exception() -> None: + entry = _make_entry() + conn = AsyncMock() + conn.fetchval = AsyncMock(side_effect=RuntimeError("DB is down")) + pool = MagicMock() + pool.acquire = MagicMock(return_value=_AsyncContextManager(conn)) + handler = NodeLogPersistenceEffect(pool=pool) + + result = await handler.handle(entry) + + assert result.status == "error" + assert result.error_message is not None + assert "DB is down" in result.error_message + + +# --------------------------------------------------------------------------- +# handle_raw synchronous shim +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +def test_handle_raw_returns_skipped_status() -> None: + entry = _make_entry() + raw = entry.model_dump(mode="json") + result = NodeLogPersistenceEffect.handle_raw(raw) + + assert result["entry_id"] == entry.entry_id + assert result["status"] == "skipped"