wehub-resource-sync
220 lines
8.0 KiB
Python
220 lines
8.0 KiB
Python
# SPDX-License-Identifier: Apache-2.0
|
|
|
|
"""OpenTelemetry SDK initialization for the MP observability system.
|
|
|
|
Supports two modes, controlled by the ``otlp_endpoint`` field in
|
|
``ObservabilityConfig``:
|
|
|
|
- **OTLP push** (production): metrics/traces are pushed to an OTel collector.
|
|
- **Prometheus pull** (dev/debug): metrics are served on a local ``/metrics``
|
|
endpoint via ``prometheus_client``, no collector needed.
|
|
"""
|
|
|
|
# Future
|
|
from __future__ import annotations
|
|
|
|
# Standard
|
|
from collections.abc import Callable
|
|
from typing import TYPE_CHECKING
|
|
|
|
if TYPE_CHECKING:
|
|
# Third Party
|
|
from opentelemetry.sdk.resources import Resource
|
|
|
|
# First Party
|
|
from lmcache.logging import init_logger
|
|
|
|
logger = init_logger(__name__)
|
|
|
|
|
|
def _build_resource(resource_attributes: dict[str, str] | None) -> "Resource":
|
|
"""Build an OTel ``Resource`` from the given attribute dict.
|
|
|
|
Returns an empty ``Resource`` when *resource_attributes* is empty or
|
|
``None`` so that telemetry carries no stale process-level tags.
|
|
"""
|
|
# Third Party
|
|
from opentelemetry.sdk.resources import Resource
|
|
|
|
if not resource_attributes:
|
|
return Resource.create({})
|
|
return Resource.create(dict(resource_attributes))
|
|
|
|
|
|
def init_otel_metrics(
|
|
otlp_endpoint: str | None = None,
|
|
prometheus_port: int | None = None,
|
|
resource_attributes: dict[str, str] | None = None,
|
|
start_http_server: bool = True,
|
|
) -> None:
|
|
"""Set up the OpenTelemetry MeterProvider.
|
|
|
|
Args:
|
|
otlp_endpoint: OTLP gRPC endpoint (e.g. ``http://localhost:4317``).
|
|
When set, metrics are pushed to an OTel collector.
|
|
When ``None``, falls back to Prometheus pull mode.
|
|
prometheus_port: Port for the fallback Prometheus ``/metrics``
|
|
endpoint. Only used when *otlp_endpoint* is ``None``.
|
|
Defaults to 9090.
|
|
resource_attributes: Optional ``{attr_name: value}`` map attached
|
|
to the ``MeterProvider`` ``Resource``. Every metric emitted
|
|
through the provider carries these attributes. Intended for
|
|
process-level identity (e.g. ``service.instance.id``) — never
|
|
for per-request tags.
|
|
start_http_server: Whether to start a standalone Prometheus
|
|
HTTP server. Set to ``False`` when metrics are already
|
|
served by an external HTTP framework (e.g. FastAPI).
|
|
"""
|
|
# Third Party
|
|
from opentelemetry import metrics
|
|
from opentelemetry.sdk.metrics import MeterProvider
|
|
|
|
resource = _build_resource(resource_attributes)
|
|
|
|
if otlp_endpoint is not None:
|
|
# OTLP push mode
|
|
# Third Party
|
|
from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import (
|
|
OTLPMetricExporter,
|
|
)
|
|
from opentelemetry.sdk.metrics.export import (
|
|
PeriodicExportingMetricReader,
|
|
)
|
|
|
|
exporter = OTLPMetricExporter(endpoint=otlp_endpoint, insecure=True)
|
|
reader = PeriodicExportingMetricReader(exporter, export_interval_millis=10000)
|
|
provider = MeterProvider(metric_readers=[reader], resource=resource)
|
|
metrics.set_meter_provider(provider)
|
|
logger.info(
|
|
"OTel MeterProvider initialised with OTLP exporter (%s), resource=%s",
|
|
otlp_endpoint,
|
|
dict(resource.attributes),
|
|
)
|
|
else:
|
|
# Prometheus pull fallback — no collector needed
|
|
# Third Party
|
|
from opentelemetry.exporter.prometheus import PrometheusMetricReader
|
|
import prometheus_client
|
|
|
|
if prometheus_port is None:
|
|
prometheus_port = 9090
|
|
|
|
reader = PrometheusMetricReader()
|
|
provider = MeterProvider(metric_readers=[reader], resource=resource)
|
|
metrics.set_meter_provider(provider)
|
|
if start_http_server:
|
|
prometheus_client.start_http_server(prometheus_port)
|
|
logger.info(
|
|
"OTel MeterProvider initialised with Prometheus fallback "
|
|
"(http://0.0.0.0:%d/metrics), resource=%s",
|
|
prometheus_port,
|
|
dict(resource.attributes),
|
|
)
|
|
else:
|
|
logger.info(
|
|
"OTel MeterProvider initialised with Prometheus fallback "
|
|
"(standalone metrics HTTP server disabled; "
|
|
"/metrics must be exposed by the caller), resource=%s",
|
|
dict(resource.attributes),
|
|
)
|
|
|
|
|
|
def init_otel_tracing(
|
|
otlp_endpoint: str | None = None,
|
|
resource_attributes: dict[str, str] | None = None,
|
|
) -> None:
|
|
"""Set up the OpenTelemetry TracerProvider with an OTLP exporter.
|
|
|
|
Tracing requires an OTLP endpoint — there is no local fallback.
|
|
When *otlp_endpoint* is ``None``, tracing init is skipped.
|
|
|
|
Args:
|
|
otlp_endpoint: OTLP gRPC endpoint. When ``None``, tracing
|
|
init is skipped (no-op).
|
|
resource_attributes: Optional ``{attr_name: value}`` map attached
|
|
to the ``TracerProvider`` ``Resource``. Every span emitted
|
|
through the provider carries these attributes.
|
|
"""
|
|
if otlp_endpoint is None:
|
|
logger.debug("No OTLP endpoint configured, skipping tracing init")
|
|
return
|
|
|
|
# Third Party
|
|
from opentelemetry import trace
|
|
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import (
|
|
OTLPSpanExporter,
|
|
)
|
|
from opentelemetry.sdk.trace import TracerProvider
|
|
from opentelemetry.sdk.trace.export import BatchSpanProcessor
|
|
|
|
resource = _build_resource(resource_attributes)
|
|
exporter = OTLPSpanExporter(endpoint=otlp_endpoint, insecure=True)
|
|
provider = TracerProvider(resource=resource)
|
|
provider.add_span_processor(BatchSpanProcessor(exporter))
|
|
trace.set_tracer_provider(provider)
|
|
logger.info(
|
|
"OTel TracerProvider initialised with OTLP exporter (%s), resource=%s",
|
|
otlp_endpoint,
|
|
dict(resource.attributes),
|
|
)
|
|
|
|
|
|
def register_gauge(
|
|
meter_name: str,
|
|
gauge_name: str,
|
|
description: str,
|
|
func: Callable[[], int | float]
|
|
| Callable[[], list[tuple[int | float, dict[str, object]]]],
|
|
) -> None:
|
|
"""Register an OTel observable gauge with a callback.
|
|
|
|
This is a convenience wrapper that hides the OTel boilerplate.
|
|
If OTel is not available, the call is silently ignored.
|
|
|
|
Two callback shapes are accepted:
|
|
|
|
- **Single value (no attributes).** ``func`` returns ``int`` / ``float``
|
|
and the gauge emits one datapoint per scrape with no attributes.
|
|
Use this for whole-process metrics like
|
|
``lmcache_mp.active_prefetch_jobs``.
|
|
- **Per-attribute-set values.** ``func`` returns a list of
|
|
``(value, attrs)`` tuples; the gauge emits one datapoint per tuple
|
|
with the given attributes. Use this for per-adapter or per-tier
|
|
metrics that share a name but vary by attribute. An empty list
|
|
reports no datapoints (the metric simply does not appear in the
|
|
next scrape, which is the correct shape when there is nothing to
|
|
observe).
|
|
|
|
Args:
|
|
meter_name: OTel meter name (e.g. ``lmcache.mp_server``).
|
|
gauge_name: Metric name (e.g.
|
|
``lmcache_mp.active_prefetch_jobs``).
|
|
description: Human-readable description of the gauge.
|
|
func: Zero-arg callable. Either a function returning the current
|
|
scalar value, or a function returning a list of
|
|
``(value, attrs)`` tuples for tagged observations.
|
|
"""
|
|
try:
|
|
# Third Party
|
|
from opentelemetry import metrics as otel_metrics
|
|
|
|
def _callback(_options):
|
|
result = func()
|
|
if isinstance(result, list):
|
|
return [
|
|
otel_metrics.Observation(value, attrs) for value, attrs in result
|
|
]
|
|
return [otel_metrics.Observation(result)]
|
|
|
|
meter = otel_metrics.get_meter(meter_name)
|
|
meter.create_observable_gauge(
|
|
gauge_name,
|
|
callbacks=[_callback],
|
|
description=description,
|
|
)
|
|
except ImportError:
|
|
logger.debug(
|
|
"opentelemetry package not found, skipping gauge %s",
|
|
gauge_name,
|
|
)
|