Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 16 additions & 10 deletions openviking/telemetry/execution.py
Original file line number Diff line number Diff line change
Expand Up @@ -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]],
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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,
Expand Down
26 changes: 26 additions & 0 deletions tests/metrics/integration/test_telemetry_bridge.py
Original file line number Diff line number Diff line change
@@ -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):
Expand Down Expand Up @@ -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))),
Expand Down
26 changes: 26 additions & 0 deletions tests/telemetry/test_execution.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Loading