File size: 1,152 Bytes
7a87926 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 | """
Telemetry helpers (structured events + optional spans).
This keeps OpenTelemetry optional so unit tests and lightweight installs don't
need to pull in tracing dependencies.
"""
from __future__ import annotations
from contextlib import contextmanager
from typing import Any, Dict, Iterator, Optional
def _try_get_tracer():
try:
from opentelemetry import trace # type: ignore
return trace.get_tracer("ylff")
except Exception: # pragma: no cover
return None
@contextmanager
def span(name: str, *, attributes: Optional[Dict[str, Any]] = None) -> Iterator[None]:
"""
Context manager that creates an OpenTelemetry span if available,
otherwise a no-op.
"""
tracer = _try_get_tracer()
if tracer is None:
yield
return
with tracer.start_as_current_span(name) as s: # pragma: no cover
if attributes:
for k, v in attributes.items():
try:
s.set_attribute(k, v)
except Exception:
# Avoid blowing up the business logic on bad attribute types.
pass
yield
|