From a3cf0c893f7ebe606363283afb59a189d1bde62c Mon Sep 17 00:00:00 2001 From: Hao Zhe Date: Sun, 18 Oct 2026 15:42:55 +0800 Subject: [PATCH] fix(telemetry): publish failed operation metrics --- openviking/telemetry/execution.py | 26 ++++++++++++------- .../integration/test_telemetry_bridge.py | 26 +++++++++++++++++++ tests/telemetry/test_execution.py | 26 +++++++++++++++++++ 3 files changed, 68 insertions(+), 10 deletions(-) diff --git a/openviking/telemetry/execution.py b/openviking/telemetry/execution.py index d533e04fd1..82cb85511f 100644 --- a/openviking/telemetry/execution.py +++ b/openviking/telemetry/execution.py @@ -60,6 +60,20 @@ def _log_telemetry_summary(snapshot: Optional[TelemetrySnapshot]) -> None: ) +def _publish_telemetry_summary(snapshot: Optional[TelemetrySnapshot]) -> None: + """Publish a finished operation summary without affecting its outcome.""" + if snapshot is None: + return + try: + from openviking.metrics.datasources.telemetry_bridge import ( + TelemetryBridgeEventDataSource, + ) + + TelemetryBridgeEventDataSource.record_summary(snapshot.summary) + except Exception: + logger.debug("failed to record telemetry summary into metrics bridge", exc_info=True) + + def attach_telemetry_payload( result: Any, telemetry_payload: Optional[dict[str, Any]], @@ -119,7 +133,6 @@ def _fill_telemetry_summary_to_span(snapshot, span): logger.debug("failed to fill telemetry summary into span", exc_info=True) - async def run_with_telemetry( *, operation: str, @@ -171,6 +184,7 @@ async def run_with_telemetry( collector.set_error(operation, type(exc).__name__, str(exc)) snapshot = collector.finish(status=error_status) _log_telemetry_summary(snapshot) + _publish_telemetry_summary(snapshot) # If a span exists, record exception and backfill summary (best-effort). if otel_trace is not None: @@ -200,15 +214,7 @@ async def run_with_telemetry( except Exception: logger.debug("failed to backfill telemetry summary into span", exc_info=True) - try: - if snapshot is not None: - from openviking.metrics.datasources.telemetry_bridge import ( - TelemetryBridgeEventDataSource, - ) - - TelemetryBridgeEventDataSource.record_summary(snapshot.summary) - except Exception: - logger.debug("failed to record telemetry summary into metrics bridge", exc_info=True) + _publish_telemetry_summary(snapshot) telemetry_payload = build_telemetry_payload( snapshot, selection, diff --git a/tests/metrics/integration/test_telemetry_bridge.py b/tests/metrics/integration/test_telemetry_bridge.py index 29118905f8..401d39e882 100644 --- a/tests/metrics/integration/test_telemetry_bridge.py +++ b/tests/metrics/integration/test_telemetry_bridge.py @@ -1,10 +1,13 @@ # Copyright (c) 2026 Beijing Volcano Engine Technology Co., Ltd. # SPDX-License-Identifier: AGPL-3.0 +import pytest + from openviking.metrics.datasources.telemetry_bridge import TelemetryBridgeEventDataSource from openviking.metrics.global_api import init_metrics_from_server_config, shutdown_metrics from openviking.observability.events import try_publish_event from openviking.server.config import MetricsConfig, ObservabilityConfig, ServerConfig +from openviking.telemetry.execution import run_with_telemetry def test_queue_duration_event_is_wired_to_metrics_registry(registry, render_prometheus): @@ -140,6 +143,29 @@ def test_telemetry_bridge_records_operation_and_resource_metrics(registry, rende shutdown_metrics(app=None) +@pytest.mark.asyncio +async def test_failed_operation_is_wired_to_metrics_registry(registry, render_prometheus): + init_metrics_from_server_config( + ServerConfig(observability=ObservabilityConfig(metrics=MetricsConfig(enabled=True))), + app=None, + registry=registry, + ) + + async def fail() -> None: + raise RuntimeError("injected failure") + + try: + with pytest.raises(RuntimeError, match="injected failure"): + await run_with_telemetry(operation="search.search", telemetry=False, fn=fail) + + text = render_prometheus(registry) + labels = 'account_id="__unknown__",operation="search.search",status="error"' + assert f"openviking_operation_requests_total{{{labels}}} 1" in text + assert f"openviking_operation_duration_seconds_count{{{labels}}} 1" in text + finally: + shutdown_metrics(app=None) + + def test_telemetry_bridge_semantic_nodes_total_is_cumulative(registry, render_prometheus): init_metrics_from_server_config( ServerConfig(observability=ObservabilityConfig(metrics=MetricsConfig(enabled=True))), diff --git a/tests/telemetry/test_execution.py b/tests/telemetry/test_execution.py index 79ea9cdc21..6e7cb3b1a3 100644 --- a/tests/telemetry/test_execution.py +++ b/tests/telemetry/test_execution.py @@ -206,6 +206,32 @@ async def _run(): assert execution.telemetry["summary"]["operation"] == "search.find" +@pytest.mark.asyncio +async def test_run_with_telemetry_preserves_error_when_metrics_publish_fails(monkeypatch): + from openviking.metrics.datasources.telemetry_bridge import TelemetryBridgeEventDataSource + from openviking.telemetry.execution import run_with_telemetry + + publish_attempts = [] + + def fail_publish(_summary): + publish_attempts.append(True) + raise RuntimeError("metrics unavailable") + + async def fail_operation(): + raise RuntimeError("operation failed") + + monkeypatch.setattr(TelemetryBridgeEventDataSource, "record_summary", fail_publish) + + with pytest.raises(RuntimeError, match="operation failed"): + await run_with_telemetry( + operation="search.find", + telemetry=False, + fn=fail_operation, + ) + + assert publish_attempts == [True] + + @pytest.mark.asyncio async def test_run_with_telemetry_raises_invalid_argument_for_bad_request(): from openviking.telemetry.execution import run_with_telemetry