"""Gateway Health & Diagnostics OTLP export runtime. This exporter emits operator-owned gateway service-health metrics plus narrow redacted diagnostic events. It is deliberately in-process and fail-open so it works under systemd, launchd, s6, containers, tmux, nohup, or a simple shell without a sidecar/watchdog dependency. """ from __future__ import annotations import logging import os import re import threading from dataclasses import dataclass from typing import Any, Dict, Optional logger = logging.getLogger(__name__) _DEFAULT_DIAGNOSTIC_SCOPE = "hermes.gateway.diagnostics" _RESOURCE_ATTRIBUTE_KEYS = frozenset({ "service.name", "service.namespace", "service.version", "service.instance.id", "deployment.environment.name", "cloud.provider", "cloud.platform", "cloud.region", "telemetry.scope", }) _DIAGNOSTIC_ATTRIBUTE_KEYS = frozenset({ "name", "subsystem", "error_class", "error_code", "platform", "old_state", "new_state", "version", "severity", }) _SAFE_RESOURCE_VALUE = re.compile(r"^[A-Za-z0-9._:/-]{1,128}$") def _redact_string(raw: Any, *, limit: int = 500) -> str: try: from agent.monitoring.redaction import redact_for_export return (redact_for_export(str(raw or "")) or "[redacted]")[:limit] except Exception: return "[redaction-unavailable]" def _safe_resource_attributes(raw: Any) -> Dict[str, str]: """Allowlist bounded resource labels and reject values changed by redaction.""" attrs: Dict[str, str] = {} if not isinstance(raw, dict): return attrs for key, value in raw.items(): key = str(key) if key not in _RESOURCE_ATTRIBUTE_KEYS or value is None: continue if key == "service.instance.id": from agent.monitoring.gateway_health import _safe_instance_id attrs[key] = _safe_instance_id(value) continue text = str(value) if not _SAFE_RESOURCE_VALUE.fullmatch(text): continue if _redact_string(text, limit=128) != text: continue attrs[key] = text return attrs def _runtime_resource_attributes( config: Dict[str, Any], *, telemetry_scope: str ) -> Dict[str, str]: """Build the safe OTLP resource shared by metrics and diagnostic logs.""" gh = _gateway_health_config(config) attrs = _safe_resource_attributes(gh.get("resource_attributes")) from agent.monitoring.gateway_health import _safe_instance_id attrs["service.name"] = "hermes-gateway" attrs["service.instance.id"] = _safe_instance_id(_install_id(config)) attrs["telemetry.scope"] = telemetry_scope return attrs def _diagnostic_log_attributes(event: Dict[str, Any]) -> Dict[str, Any]: attrs: Dict[str, Any] = {} for key in _DIAGNOSTIC_ATTRIBUTE_KEYS: value = event.get(key) if value is None: continue attrs[f"hermes.{key}"] = _redact_string(value) if isinstance(value, str) else value return attrs @dataclass(slots=True) class GatewayHealthExportRuntime: enabled: bool reason: str = "disabled" streamer: Any = None metric_provider: Any = None log_handler: Any = None log_streamer: Any = None thread: Optional[threading.Thread] = None stop_event: Optional[threading.Event] = None def shutdown(self) -> None: if self.stop_event is not None: self.stop_event.set() if self.thread is not None: self.thread.join(timeout=0.25) if self.log_handler is not None: try: logging.getLogger().removeHandler(self.log_handler) except Exception: pass # All producers above are now stopped. Drain queued and in-flight # events before detaching subscribers so the terminal lifecycle event # cannot race exporter shutdown. The barrier is bounded and fail-open. try: from agent.monitoring.emitter import get_emitter emitter = get_emitter() emitter.flush(timeout=1.0) if self.streamer is not None: emitter.unsubscribe(self.streamer) if self.log_streamer is not None: emitter.unsubscribe(self.log_streamer) except Exception: pass # Network flush/close runs under one bounded daemon-thread deadline and # can never delay gateway teardown indefinitely. closeables = [ item for item in (self.streamer, self.log_streamer, self.metric_provider) if item is not None ] def _close() -> None: for item in closeables: try: item.shutdown() except Exception: pass if closeables: worker = threading.Thread( target=_close, name="hermes-gateway-health-export-shutdown", daemon=True, ) worker.start() worker.join(timeout=2.0) self.streamer = None self.log_streamer = None self.metric_provider = None self.thread = None self.stop_event = None def _gateway_health_config(config: Dict[str, Any]) -> Dict[str, Any]: mon = (config or {}).get("monitoring") or {} return mon.get("gateway_health_export") or {} def _otlp_config(config: Dict[str, Any]) -> Dict[str, Any]: mon = (config or {}).get("monitoring") or {} export = mon.get("export") or {} return export.get("otlp") or {} def _enabled(config: Dict[str, Any]) -> bool: gh = _gateway_health_config(config) otlp = _otlp_config(config) return bool(gh.get("enabled") and otlp.get("enabled") and otlp.get("endpoint")) def _require_metrics_sdk(*, auto_install: bool = True, prompt: bool = False) -> Dict[str, Any]: if auto_install: try: from tools.lazy_deps import ensure as _lazy_ensure _lazy_ensure("export.otlp", prompt=prompt) except Exception: pass try: from opentelemetry.exporter.otlp.proto.http._log_exporter import OTLPLogExporter from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter from opentelemetry.metrics import Observation from opentelemetry.trace import INVALID_SPAN_ID, INVALID_TRACE_ID, TraceFlags from opentelemetry._logs import LogRecord from opentelemetry._logs.severity import SeverityNumber from opentelemetry.sdk._logs import LoggerProvider from opentelemetry.sdk._logs.export import BatchLogRecordProcessor from opentelemetry.sdk.metrics import MeterProvider from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader from opentelemetry.sdk.resources import Resource return { "OTLPLogExporter": OTLPLogExporter, "OTLPMetricExporter": OTLPMetricExporter, "Observation": Observation, "LogRecord": LogRecord, "LoggerProvider": LoggerProvider, "INVALID_SPAN_ID": INVALID_SPAN_ID, "INVALID_TRACE_ID": INVALID_TRACE_ID, "TraceFlags": TraceFlags, "SeverityNumber": SeverityNumber, "BatchLogRecordProcessor": BatchLogRecordProcessor, "MeterProvider": MeterProvider, "PeriodicExportingMetricReader": PeriodicExportingMetricReader, "Resource": Resource, } except Exception as exc: raise RuntimeError(f"OTLP metrics SDK unavailable: {exc}") from exc def _resolve_headers(headers_env: Optional[Dict[str, str]]) -> Dict[str, str]: resolved: Dict[str, str] = {} for header_name, env_name in (headers_env or {}).items(): val = os.environ.get(str(env_name)) if val: resolved[str(header_name)] = val return resolved def _metric_endpoint(endpoint: str) -> str: if endpoint.endswith("/v1/traces"): return endpoint[: -len("/v1/traces")] + "/v1/metrics" return endpoint def _logs_endpoint(endpoint: str) -> str: if endpoint.endswith("/v1/traces"): return endpoint[: -len("/v1/traces")] + "/v1/logs" if endpoint.endswith("/v1/metrics"): return endpoint[: -len("/v1/metrics")] + "/v1/logs" return endpoint def _version() -> str: try: from hermes_cli import __version__ return str(__version__) except Exception: return "unknown" def _profile() -> str: try: from hermes_cli.profiles import get_active_profile_name return str(get_active_profile_name() or "default") except Exception: return "default" def _install_id(config: Dict[str, Any]) -> str: try: from agent.monitoring.policy import ensure_install_id return str(ensure_install_id(config)) except Exception: return "unknown" def _supervision_mode() -> str: if os.environ.get("INVOCATION_ID"): return "systemd" if os.environ.get("S6_CMD_ARG0") or os.environ.get("S6_VERSION"): return "s6" if os.environ.get("container") or os.path.exists("/.dockerenv"): return "container" if os.environ.get("LAUNCHD_SOCKET"): return "launchd" return "manual" def _read_gateway_snapshot(config: Dict[str, Any]): from agent.monitoring.gateway_health import build_gateway_health_snapshot try: from gateway.status import read_runtime_status runtime = read_runtime_status() or {} except Exception: runtime = {} return build_gateway_health_snapshot( runtime, gateway_running=True, profile=_profile(), install_id=_install_id(config), version=_version(), supervision_mode=_supervision_mode(), ) def _read_cron_snapshot(): from agent.monitoring.cron_health import build_cron_health_snapshot return build_cron_health_snapshot() def _read_background_work_count() -> int: """Count live background/subagent work that ``active_agents`` does NOT include. ``hermes.gateway.active_agents`` counts foreground turns + in-flight cron jobs + API runs, but deliberately excludes backgrounded ``delegate_task`` subagents, ``terminal(background=true)`` processes, kanban workers, and the runner's own background tasks (they are tracked only for the scale-to-zero suspend guard, ``_scale_to_zero_has_live_background_work``). Without this metric a peer churning through delegated subagents shows ``active_agents=0`` on the fleet dashboard. Best-effort and content-free: a single integer, no job/task identity. Returns 0 if a source can't be imported. Delegation is counted TASK-granular (``active_task_count``): a fan-out batch of N subagents contributes N, not 1, so the metric reflects real concurrent subagent load rather than dispatch-unit/pool-slot count. This intentionally differs from the async pool's capacity accounting (one batch = one slot). """ total = 0 try: from tools.async_delegation import active_task_count total += max(0, int(active_task_count())) except Exception: logger.debug("background-work async-delegation count failed", exc_info=True) try: from tools.process_registry import process_registry total += max(0, int(process_registry.count_running())) except Exception: logger.debug("background-work process-registry count failed", exc_info=True) return total def _read_background_delegations_count() -> int: """Count live async delegation UNITS (dispatch/pool slots). Complements ``_read_background_work_count`` (which is task-granular): this counts each ``delegate_task`` dispatch as ONE regardless of fan-out width, matching the async pool's capacity accounting (a batch = one slot). Together the two metrics let an operator see both slot pressure (``background_delegations``, alert vs ``max_concurrent_children``) and real concurrent subagent load (``background_work``). Delegations only — it does not include ``terminal(background)`` / kanban work, which are already folded into ``background_work``. Best-effort; 0 if the source can't be imported. """ try: from tools.async_delegation import active_count return max(0, int(active_count())) except Exception: logger.debug("background-delegations count failed", exc_info=True) return 0 def _read_runtime_snapshot(config: Dict[str, Any]): gateway_snapshot = _read_gateway_snapshot(config) # Background/subagent work — a distinct metric from active_agents (which # never counts it). Appended to the gateway snapshot so it rides the same # base resource attributes (service.instance.id etc.). try: from agent.monitoring.gateway_health import GatewayMetric base = dict(gateway_snapshot.metrics[0].attributes) if gateway_snapshot.metrics else {} gateway_snapshot.metrics.append( GatewayMetric( name="hermes.gateway.background_work", value=_read_background_work_count(), attributes=base, ) ) gateway_snapshot.metrics.append( GatewayMetric( name="hermes.gateway.background_delegations", value=_read_background_delegations_count(), attributes=base, ) ) except Exception as exc: logger.warning( "background-work snapshot unavailable; metric not exported (error_type=%s)", type(exc).__name__, ) logger.debug("background-work snapshot traceback", exc_info=True) try: cron_snapshot = _read_cron_snapshot() except Exception as exc: # Content-free visibility: cron telemetry silently dropping out is a # release-relevant regression, so surface it at WARNING with only the # exception *type* name (never the message, which could carry paths or # other environment detail). exc_info stays on the DEBUG record. logger.warning( "cron health snapshot unavailable; cron telemetry not exported (error_type=%s)", type(exc).__name__, ) logger.debug("cron health snapshot traceback", exc_info=True) return gateway_snapshot gateway_snapshot.metrics.extend(cron_snapshot.metrics) return gateway_snapshot def _emit_snapshot_events(config: Dict[str, Any]) -> None: gh = _gateway_health_config(config) if not gh.get("diagnostic_events_enabled", True): return try: from agent.monitoring import emitter snapshot = _read_runtime_snapshot(config) for event in snapshot.events: emitter.emit(event) except Exception: logger.debug("gateway health snapshot emit failed", exc_info=True) def _start_metric_provider(config: Dict[str, Any], sdk: Dict[str, Any]) -> Any: gh = _gateway_health_config(config) if not gh.get("metrics_enabled", True): return None otlp = _otlp_config(config) endpoint = _metric_endpoint(str(otlp.get("endpoint"))) headers = _resolve_headers(otlp.get("headers_env")) exporter = sdk["OTLPMetricExporter"](endpoint=endpoint, headers=headers or None) interval_ms = max(5, int(gh.get("export_interval_seconds", 60))) * 1000 reader = sdk["PeriodicExportingMetricReader"](exporter, export_interval_millis=interval_ms) resource_attrs = _runtime_resource_attributes( config, telemetry_scope="gateway_health" ) provider = sdk["MeterProvider"]( metric_readers=[reader], resource=sdk["Resource"].create(resource_attrs), ) meter = provider.get_meter("hermes.gateway.health") Observation = sdk["Observation"] metric_names = [ "hermes.gateway.up", "hermes.gateway.state", "hermes.gateway.active_agents", "hermes.gateway.busy", "hermes.gateway.drainable", "hermes.gateway.restart_requested", "hermes.gateway.background_work", "hermes.gateway.background_delegations", "hermes.platform.up", "hermes.platform.degraded", "hermes.cron.scheduler.heartbeat_age_seconds", "hermes.cron.scheduler.last_success_age_seconds", "hermes.cron.scheduler.catch_up_occurrences", "hermes.cron.jobs.enabled", "hermes.cron.jobs.running", "hermes.cron.jobs.overdue", ] def callback(name: str): def _cb(_options=None): try: snapshot = _read_runtime_snapshot(config) return [Observation(m.value, m.attributes) for m in snapshot.metrics if m.name == name] except Exception: logger.debug("gateway metric callback failed", exc_info=True) return [] return _cb for metric_name in metric_names: meter.create_observable_gauge(metric_name, callbacks=[callback(metric_name)]) return provider def _severity_number(sdk: Dict[str, Any], severity: Any) -> Any: SeverityNumber = sdk["SeverityNumber"] sev = str(severity or "warning").lower() if sev in {"critical", "fatal"}: return SeverityNumber.FATAL if sev == "error": return SeverityNumber.ERROR if sev in {"info", "information"}: return SeverityNumber.INFO if sev == "debug": return SeverityNumber.DEBUG return SeverityNumber.WARN class GatewayDiagnosticLogStreamer: """Emitter subscriber that sends gateway diagnostic events as OTLP logs.""" def __init__(self, config: Dict[str, Any], sdk: Dict[str, Any]): otlp = _otlp_config(config) headers = _resolve_headers(otlp.get("headers_env")) endpoint = _logs_endpoint(str(otlp.get("endpoint"))) resource_attrs = _runtime_resource_attributes( config, telemetry_scope="gateway_diagnostics" ) self._provider = sdk["LoggerProvider"](resource=sdk["Resource"].create(resource_attrs)) self._processor = sdk["BatchLogRecordProcessor"]( sdk["OTLPLogExporter"](endpoint=endpoint, headers=headers or None) ) self._provider.add_log_record_processor(self._processor) self._logger = self._provider.get_logger(_DEFAULT_DIAGNOSTIC_SCOPE) self._LogRecord = sdk["LogRecord"] self._sdk = sdk self.exported = 0 def __call__(self, batch: list[Dict[str, Any]]) -> None: from agent.monitoring.gateway_health import source_logger_for_export for ev in batch: if ev.get("event") != "gateway_diagnostic": continue attrs = _diagnostic_log_attributes(ev) # Preserve the source-controlled Python logger as the OTel # instrumentation scope. This adds precise code attribution without # turning a fluid module layout into a maintained subsystem enum. # Rendered messages stay out because they may contain arbitrary IDs, # names, paths, or configured strings. A future, separately gated # ``diagnostic_detail: redacted_message`` mode may add best-effort # free text when an observability plane defines that privacy policy. source_logger = source_logger_for_export(ev.get("source_logger")) otel_logger = ( self._provider.get_logger(source_logger) if source_logger is not None else self._logger ) body = "gateway diagnostic" record = self._LogRecord( timestamp=ev.get("ts_ns"), trace_id=self._sdk["INVALID_TRACE_ID"], span_id=self._sdk["INVALID_SPAN_ID"], trace_flags=self._sdk["TraceFlags"].DEFAULT, severity_text=str(ev.get("severity") or "warning").upper(), severity_number=_severity_number(self._sdk, ev.get("severity")), body=_redact_string(body), attributes=attrs, ) otel_logger.emit(record) self.exported += 1 def shutdown(self) -> None: try: from agent.monitoring.emitter import get_emitter get_emitter().unsubscribe(self) except Exception: pass try: self._processor.force_flush() self._provider.shutdown() except Exception: pass def _start_diagnostic_log_streamer(config: Dict[str, Any], sdk: Dict[str, Any]) -> GatewayDiagnosticLogStreamer: from agent.monitoring.emitter import get_emitter streamer = GatewayDiagnosticLogStreamer(config, sdk) get_emitter().subscribe(streamer) return streamer def _start_snapshot_thread(config: Dict[str, Any], stop_event: threading.Event) -> threading.Thread: interval = max(5, int(_gateway_health_config(config).get("logs_export_interval_seconds", 5))) def _run() -> None: while not stop_event.wait(interval): _emit_snapshot_events(config) thread = threading.Thread(target=_run, name="hermes-gateway-health-export", daemon=True) thread.start() return thread def _attach_log_handler(config: Dict[str, Any]) -> Any: gh = _gateway_health_config(config) if not gh.get("diagnostic_events_enabled", True) or not gh.get("warning_error_events_enabled", True): return None from agent.monitoring.gateway_health import GatewayDiagnosticLogHandler handler = GatewayDiagnosticLogHandler(profile=_profile(), version=_version()) root = logging.getLogger() if handler not in root.handlers: root.addHandler(handler) return handler def _gateway_health_event(ev: Dict[str, Any]) -> bool: return ev.get("event") in {"gateway_health", "cron_execution"} def start_gateway_health_export(config: Dict[str, Any]) -> GatewayHealthExportRuntime: """Start P0 gateway health export if configured. Never raises.""" if not _enabled(config): return GatewayHealthExportRuntime(enabled=False, reason="disabled") gh = _gateway_health_config(config) runtime = GatewayHealthExportRuntime(enabled=True, reason="enabled") sdk: Optional[Dict[str, Any]] = None if gh.get("metrics_enabled", True) or gh.get("diagnostic_events_enabled", True): try: sdk = _require_metrics_sdk(prompt=False) except Exception: logger.warning( "monitoring.gateway_health_export.enabled but OTLP SDK is unavailable; " "install 'hermes-agent[otlp]'", exc_info=True, ) return GatewayHealthExportRuntime(enabled=False, reason="otlp_unavailable") if gh.get("metrics_enabled", True) and sdk is not None: try: runtime.metric_provider = _start_metric_provider(config, sdk) except Exception: logger.warning("gateway health OTLP metrics failed to start", exc_info=True) runtime.shutdown() return GatewayHealthExportRuntime(enabled=False, reason="metrics_start_failed") if gh.get("diagnostic_events_enabled", True) and sdk is not None: try: from agent.monitoring import otlp_exporter runtime.streamer = otlp_exporter.start_streaming(config, event_filter=_gateway_health_event) if runtime.streamer is None: raise RuntimeError("gateway health span streamer did not start") runtime.log_streamer = _start_diagnostic_log_streamer(config, sdk) except Exception: logger.debug("gateway diagnostic OTLP export failed to start", exc_info=True) runtime.shutdown() return GatewayHealthExportRuntime(enabled=False, reason="diagnostics_start_failed") try: runtime.log_handler = _attach_log_handler(config) except Exception: logger.debug("gateway diagnostic log handler failed to attach", exc_info=True) if gh.get("diagnostic_events_enabled", True): try: _emit_snapshot_events(config) runtime.stop_event = threading.Event() runtime.thread = _start_snapshot_thread(config, runtime.stop_event) except Exception: logger.debug("gateway health snapshot thread failed to start", exc_info=True) return runtime __all__ = [ "GatewayHealthExportRuntime", "start_gateway_health_export", ]