ccproxy.plugins.metrics.hook

ccproxy.plugins.metrics.hook

Hook-based metrics collection implementation.

MetricsHook

MetricsHook(config=None)

Bases: Hook

Hook-based metrics collection implementation.

This hook listens to request/response lifecycle events and updates Prometheus metrics accordingly. It provides event-driven metric collection without requiring direct metric calls in the code.

Parameters:

Name	Type	Description	Default
config	MetricsConfig | None	Metrics configuration	None

Source code in ccproxy/plugins/metrics/hook.py

def __init__(self, config: MetricsConfig | None = None) -> None:
    """Initialize the metrics hook.

    Args:
        config: Metrics configuration
    """
    self.config = config or MetricsConfig()

    # Initialize collectors based on config using an isolated registry to
    # avoid global REGISTRY collisions in multi-app/test environments.
    if self.config.enabled:
        registry = None
        try:
            from prometheus_client import (
                CollectorRegistry as CollectorRegistry,
            )

            registry = CollectorRegistry()
        except Exception:
            registry = None

        self.collector: PrometheusMetrics | None = PrometheusMetrics(
            namespace=self.config.namespace,
            histogram_buckets=self.config.histogram_buckets,
            registry=registry,
        )
    else:
        self.collector = None

    self.pushgateway: PushgatewayClient | None = (
        PushgatewayClient(self.config)
        if self.config.pushgateway_enabled and self.config.enabled
        else None
    )

    # Track active requests and their start times
    self._request_start_times: dict[str, float] = {}

    logger.debug(
        "metrics_configured",
        enabled=self.config.enabled,
        namespace=self.config.namespace,
        pushgateway_enabled=self.config.pushgateway_enabled,
        pushgateway_url=self.config.pushgateway_url,
    )

get_collector

get_collector()

Get the Prometheus metrics collector instance.

Returns:

Type	Description
PrometheusMetrics | None	The metrics collector or None if disabled

Source code in ccproxy/plugins/metrics/hook.py

def get_collector(self) -> PrometheusMetrics | None:
    """Get the Prometheus metrics collector instance.

    Returns:
        The metrics collector or None if disabled
    """
    return self.collector

get_pushgateway_client

get_pushgateway_client()

Get the Pushgateway client instance.

Returns:

Type	Description
PushgatewayClient | None	The pushgateway client or None if disabled

Source code in ccproxy/plugins/metrics/hook.py

def get_pushgateway_client(self) -> PushgatewayClient | None:
    """Get the Pushgateway client instance.

    Returns:
        The pushgateway client or None if disabled
    """
    return self.pushgateway

push_metrics async

push_metrics()

Push current metrics to Pushgateway.

Returns:

Type	Description
bool	True if push succeeded, False otherwise

Source code in ccproxy/plugins/metrics/hook.py

async def push_metrics(self) -> bool:
    """Push current metrics to Pushgateway.

    Returns:
        True if push succeeded, False otherwise
    """
    if not self.pushgateway or not self.collector or not self.collector.registry:
        return False

    return self.pushgateway.push_metrics(self.collector.registry)