Files
trex-core/trex_core/application/alarm_emitter.py
T

102 lines
3.6 KiB
Python

"""Alarm emitter — best-effort CRITICAL alarm publisher (dcy-033 / REQ-100).
Publishes REQ-098 alarm envelopes to the `trex.alerts` exchange via the
MessagingTransport port. No pika import here — dependency inversion
preserved (NFR-1). Domain layer is never touched.
Caller contract:
- Inject into execution / reconciliation / killswitch / ledger_outbox / etc.
- Call publish_alarm(code, title, fields=...) at each alarm point.
- publish_alarm() NEVER re-raises. Alarm failure is ERROR-logged and swallowed
so the core business path (slot transition / order execution / ledger) is
never blocked by alarm delivery.
transport=None is explicitly supported for CLI entry points (e.g. reset.py)
where no broker connection exists. In that case publish_alarm() logs and skips.
"""
from __future__ import annotations
import logging
import sys
from datetime import datetime, timedelta, timezone
from trex_core.adapters.messaging.topology import alarm_routing_key
from trex_core.application.messaging import MessagingTransport, TransportUnavailableError
_log = logging.getLogger(__name__)
_KST = timezone(timedelta(hours=9))
# Wire constants (REQ-098)
_SOURCE = "trex-core"
_SCHEMA_VERSION = "1"
class AlarmEmitter:
"""Application-layer alarm publisher (dcy-033 / REQ-100).
Wraps MessagingTransport.publish_alarm_event() with:
- envelope construction (source/severity/occurred_at/schema_version fixed)
- exception swallowing + ERROR log on failure
- None-transport guard for non-daemon entry points
"""
def __init__(self, transport: MessagingTransport | None) -> None:
self._transport = transport
def publish_alarm(
self,
code: str,
title: str,
*,
fields: dict | None = None,
severity: str = "critical",
) -> bool:
"""Publish a single alarm envelope. Best-effort: never re-raises.
Args:
code: Alarm identifier code (see ALARM-INVENTORY.md §A).
title: Human-readable 1-line summary. No credentials or raw
exception bodies — callers must sanitize.
fields: Optional structured context dict (opaque, default {}).
severity: 'critical' | 'warn' | 'info'. Default 'critical'.
Returns:
True if the alarm was successfully published, False otherwise.
"""
if self._transport is None:
_log.info(
"alarm skipped (no transport): code=%s title=%s",
code, title,
)
return False
routing_key = alarm_routing_key(severity, _SOURCE)
occurred_at = datetime.now(tz=_KST).isoformat()
payload = {
"source": _SOURCE,
"severity": severity,
"code": code,
"title": title,
"fields": fields or {},
"occurred_at": occurred_at,
"schema_version": _SCHEMA_VERSION,
}
try:
self._transport.publish_alarm_event(routing_key, payload)
return True
except TransportUnavailableError:
_log.error(
"alarm publish skipped (transport unavailable): code=%s title=%s",
code, title,
)
return False
except Exception: # noqa: BLE001 — alarm must never crash the caller
exc = sys.exc_info()[1]
exc_name = type(exc).__name__ if exc is not None else "Exception"
_log.error(
"alarm publish failed: code=%s title=%s error_type=%s",
code, title, exc_name,
)
return False