# src/structum_lab.plugins.observability/prometheus.py
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: 2025 PythonWoods
"""Prometheus Metrics Implementation.
Provides a concrete implementation of MetricsInterface using prometheus_client.
"""
import logging
from typing import Any
try:
from prometheus_client import Counter, Gauge, Histogram # type: ignore[import-not-found]
PROMETHEUS_AVAILABLE = True
except ImportError:
PROMETHEUS_AVAILABLE = False
Counter = None # type: ignore
Gauge = None # type: ignore
Histogram = None # type: ignore
from structum_lab.monitoring import MetricsInterface
log = logging.getLogger(__name__)
[docs]
class PrometheusMetrics(MetricsInterface):
"""Prometheus implementation of MetricsInterface."""
[docs]
def __init__(self, namespace: str = "structum") -> None:
"""Initialize the Prometheus metrics exporter.
Args:
namespace: Prefix for all metric names.
Raises:
ImportError: If prometheus_client is not installed.
"""
if not PROMETHEUS_AVAILABLE:
raise ImportError(
"prometheus_client is required for PrometheusMetrics. "
"Install with: pip install prometheus-client"
)
self.namespace = namespace
self._counters: dict[str, Counter] = {}
self._gauges: dict[str, Gauge] = {}
self._histograms: dict[str, Histogram] = {}
def _get_counter(self, name: str, tags: dict[str, str] | None = None) -> Counter:
"""Get or create a Counter metric."""
metric_name = f"{self.namespace}_{name.replace('.', '_')}"
if metric_name not in self._counters:
labelnames = list[Any](tags.keys()) if tags else []
self._counters[metric_name] = Counter(
metric_name, f"Counter for {name}", labelnames=labelnames
)
return self._counters[metric_name]
def _get_gauge(self, name: str, tags: dict[str, str] | None = None) -> Gauge:
"""Get or create a Gauge metric."""
metric_name = f"{self.namespace}_{name.replace('.', '_')}"
if metric_name not in self._gauges:
labelnames = list[Any](tags.keys()) if tags else []
self._gauges[metric_name] = Gauge(
metric_name, f"Gauge for {name}", labelnames=labelnames
)
return self._gauges[metric_name]
def _get_histogram(self, name: str, tags: dict[str, str] | None = None) -> Histogram:
"""Get or create a Histogram metric."""
metric_name = f"{self.namespace}_{name.replace('.', '_')}_seconds"
if metric_name not in self._histograms:
labelnames = list[Any](tags.keys()) if tags else []
self._histograms[metric_name] = Histogram(
metric_name,
f"Histogram for {name}",
labelnames=labelnames,
buckets=[0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1.0, 5.0],
)
return self._histograms[metric_name]
[docs]
def increment(self, name: str, value: float = 1.0, tags: dict[str, str] | None = None) -> None:
"""Increment a counter."""
counter = self._get_counter(name, tags)
if tags:
counter.labels(**tags).inc(value)
else:
counter.inc(value)
[docs]
def gauge(self, name: str, value: float, tags: dict[str, str] | None = None) -> None:
"""Set a gauge value."""
gauge_metric = self._get_gauge(name, tags)
if tags:
gauge_metric.labels(**tags).set(value)
else:
gauge_metric.set(value)
[docs]
def timing(self, name: str, value: float, tags: dict[str, str] | None = None) -> None:
"""Record a timing (uses histogram)."""
self.histogram(name, value, tags)
[docs]
def histogram(self, name: str, value: float, tags: dict[str, str] | None = None) -> None:
"""Record a value in histogram."""
hist = self._get_histogram(name, tags)
if tags:
hist.labels(**tags).observe(value)
else:
hist.observe(value)