ccproxy.plugins.metrics.collector

ccproxy.plugins.metrics.collector

Prometheus metrics collector for the metrics plugin.

This module provides direct prometheus_client integration for fast operational metrics like request counts, response times, and resource usage. These metrics are optimized for real-time monitoring and alerting.

Key features: - Thread-safe metric operations using prometheus_client - Minimal overhead for high-frequency operations - Standard Prometheus metric types (Counter, Histogram, Gauge) - Automatic label management and validation - Integration with hook events for metric updates

PrometheusMetrics

PrometheusMetrics(
    namespace="ccproxy",
    registry=None,
    histogram_buckets=None,
)

Prometheus metrics collector for operational monitoring.

Provides thread-safe, high-performance metrics collection using prometheus_client. Designed for minimal overhead in request processing hot paths.

Parameters:

Name	Type	Description	Default
namespace	str	Metric name prefix	'ccproxy'
registry	CollectorRegistry | None	Custom Prometheus registry (uses default if None)	None
histogram_buckets	list[float] | None	Custom histogram bucket boundaries	None

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

def __init__(
    self,
    namespace: str = "ccproxy",
    registry: CollectorRegistry | None = None,
    histogram_buckets: list[float] | None = None,
):
    """
    Initialize Prometheus metrics.

    Args:
        namespace: Metric name prefix
        registry: Custom Prometheus registry (uses default if None)
        histogram_buckets: Custom histogram bucket boundaries
    """
    if not PROMETHEUS_AVAILABLE:
        logger.warning(
            "prometheus_client not available. Metrics will be disabled. "
            "Install with: pip install prometheus-client"
        )

    self.namespace = namespace
    # Use default registry if None is passed
    if registry is None and PROMETHEUS_AVAILABLE:
        from prometheus_client import REGISTRY

        self.registry: CollectorRegistry | None = REGISTRY
    else:
        self.registry = registry
    self._enabled = PROMETHEUS_AVAILABLE
    self._histogram_buckets = histogram_buckets or [
        0.01,
        0.05,
        0.1,
        0.25,
        0.5,
        1.0,
        2.5,
        5.0,
        10.0,
        25.0,
    ]

    if self._enabled:
        self._init_metrics()

record_request

record_request(
    method,
    endpoint,
    model=None,
    status="unknown",
    service_type=None,
)

Record a request event.

Parameters:

Name	Type	Description	Default
method	str	HTTP method (GET, POST, etc.)	required
endpoint	str	API endpoint path	required
model	str | None	Model name used	None
status	str | int	Response status code or status string	'unknown'
service_type	str | None	Service type (claude_sdk_service, proxy_service)	None

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

def record_request(
    self,
    method: str,
    endpoint: str,
    model: str | None = None,
    status: str | int = "unknown",
    service_type: str | None = None,
) -> None:
    """
    Record a request event.

    Args:
        method: HTTP method (GET, POST, etc.)
        endpoint: API endpoint path
        model: Model name used
        status: Response status code or status string
        service_type: Service type (claude_sdk_service, proxy_service)
    """
    if not self._enabled:
        return

    self.request_counter.labels(
        method=method,
        endpoint=endpoint,
        model=model or "unknown",
        status=str(status),
        service_type=service_type or "unknown",
    ).inc()

record_response_time

record_response_time(
    duration_seconds,
    model=None,
    endpoint="unknown",
    service_type=None,
)

Record response time.

Parameters:

Name	Type	Description	Default
duration_seconds	float	Response time in seconds	required
model	str | None	Model name used	None
endpoint	str	API endpoint	'unknown'
service_type	str | None	Service type (claude_sdk_service, proxy_service)	None

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

def record_response_time(
    self,
    duration_seconds: float,
    model: str | None = None,
    endpoint: str = "unknown",
    service_type: str | None = None,
) -> None:
    """
    Record response time.

    Args:
        duration_seconds: Response time in seconds
        model: Model name used
        endpoint: API endpoint
        service_type: Service type (claude_sdk_service, proxy_service)
    """
    if not self._enabled:
        return

    self.response_time.labels(
        model=model or "unknown",
        endpoint=endpoint,
        service_type=service_type or "unknown",
    ).observe(duration_seconds)

record_tokens

record_tokens(
    token_count, token_type, model=None, service_type=None
)

Record token usage.

Parameters:

Name	Type	Description	Default
token_count	int	Number of tokens	required
token_type	str	Type of tokens (input, output, cache_read, cache_write)	required
model	str | None	Model name	None
service_type	str | None	Service type (claude_sdk_service, proxy_service)	None

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

def record_tokens(
    self,
    token_count: int,
    token_type: str,
    model: str | None = None,
    service_type: str | None = None,
) -> None:
    """
    Record token usage.

    Args:
        token_count: Number of tokens
        token_type: Type of tokens (input, output, cache_read, cache_write)
        model: Model name
        service_type: Service type (claude_sdk_service, proxy_service)
    """
    if not self._enabled or token_count <= 0:
        return

    self.token_counter.labels(
        type=token_type,
        model=model or "unknown",
        service_type=service_type or "unknown",
    ).inc(token_count)

record_cost

record_cost(
    cost_usd,
    model=None,
    cost_type="total",
    service_type=None,
)

Record cost.

Parameters:

Name	Type	Description	Default
cost_usd	float	Cost in USD	required
model	str | None	Model name	None
cost_type	str	Type of cost (input, output, cache, total)	'total'
service_type	str | None	Service type (claude_sdk_service, proxy_service)	None

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

def record_cost(
    self,
    cost_usd: float,
    model: str | None = None,
    cost_type: str = "total",
    service_type: str | None = None,
) -> None:
    """
    Record cost.

    Args:
        cost_usd: Cost in USD
        model: Model name
        cost_type: Type of cost (input, output, cache, total)
        service_type: Service type (claude_sdk_service, proxy_service)
    """
    if not self._enabled or cost_usd <= 0:
        return

    self.cost_counter.labels(
        model=model or "unknown",
        cost_type=cost_type,
        service_type=service_type or "unknown",
    ).inc(cost_usd)

record_error

record_error(
    error_type,
    endpoint="unknown",
    model=None,
    service_type=None,
)

Record an error event.

Parameters:

Name	Type	Description	Default
error_type	str	Type/name of error	required
endpoint	str	API endpoint where error occurred	'unknown'
model	str | None	Model name if applicable	None
service_type	str | None	Service type (claude_sdk_service, proxy_service)	None

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

def record_error(
    self,
    error_type: str,
    endpoint: str = "unknown",
    model: str | None = None,
    service_type: str | None = None,
) -> None:
    """
    Record an error event.

    Args:
        error_type: Type/name of error
        endpoint: API endpoint where error occurred
        model: Model name if applicable
        service_type: Service type (claude_sdk_service, proxy_service)
    """
    if not self._enabled:
        return

    self.error_counter.labels(
        error_type=error_type,
        endpoint=endpoint,
        model=model or "unknown",
        service_type=service_type or "unknown",
    ).inc()

set_active_requests

set_active_requests(count)

Set the current number of active requests.

Parameters:

Name	Type	Description	Default
count	int	Number of active requests	required

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

def set_active_requests(self, count: int) -> None:
    """
    Set the current number of active requests.

    Args:
        count: Number of active requests
    """
    if not self._enabled:
        return

    self.active_requests.set(count)

inc_active_requests

inc_active_requests()

Increment active request counter.

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

def inc_active_requests(self) -> None:
    """Increment active request counter."""
    if not self._enabled:
        return

    self.active_requests.inc()

dec_active_requests

dec_active_requests()

Decrement active request counter.

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

def dec_active_requests(self) -> None:
    """Decrement active request counter."""
    if not self._enabled:
        return

    self.active_requests.dec()

update_system_info

update_system_info(info)

Update system information.

Parameters:

Name	Type	Description	Default
info	dict[str, str]	Dictionary of system information key-value pairs	required

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

def update_system_info(self, info: dict[str, str]) -> None:
    """
    Update system information.

    Args:
        info: Dictionary of system information key-value pairs
    """
    if not self._enabled:
        return

    self.system_info.info(info)

is_enabled

is_enabled()

Check if metrics collection is enabled.

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

def is_enabled(self) -> bool:
    """Check if metrics collection is enabled."""
    return self._enabled

update_pool_gauges

update_pool_gauges(
    total_clients, available_clients, active_clients
)

Update pool gauge metrics (current state).

Parameters:

Name	Type	Description	Default
total_clients	int	Total number of clients in pool	required
available_clients	int	Number of available clients	required
active_clients	int	Number of active clients	required

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

def update_pool_gauges(
    self,
    total_clients: int,
    available_clients: int,
    active_clients: int,
) -> None:
    """
    Update pool gauge metrics (current state).

    Args:
        total_clients: Total number of clients in pool
        available_clients: Number of available clients
        active_clients: Number of active clients
    """
    if not self._enabled:
        return

    # Update gauges
    self.pool_clients_total.set(total_clients)
    self.pool_clients_available.set(available_clients)
    self.pool_clients_active.set(active_clients)

record_pool_acquisition_time

record_pool_acquisition_time(duration_seconds)

Record the time taken to acquire a client from the pool.

Parameters:

Name	Type	Description	Default
duration_seconds	float	Time in seconds to acquire client	required

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

def record_pool_acquisition_time(self, duration_seconds: float) -> None:
    """
    Record the time taken to acquire a client from the pool.

    Args:
        duration_seconds: Time in seconds to acquire client
    """
    if not self._enabled:
        return

    self.pool_acquisition_duration.observe(duration_seconds)

inc_pool_connections_created

inc_pool_connections_created()

Increment the pool connections created counter.

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

def inc_pool_connections_created(self) -> None:
    """Increment the pool connections created counter."""
    if not self._enabled:
        return

    self.pool_connections_created_total.inc()

inc_pool_connections_closed

inc_pool_connections_closed()

Increment the pool connections closed counter.

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

def inc_pool_connections_closed(self) -> None:
    """Increment the pool connections closed counter."""
    if not self._enabled:
        return

    self.pool_connections_closed_total.inc()

inc_pool_acquisitions

inc_pool_acquisitions()

Increment the pool acquisitions counter.

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

def inc_pool_acquisitions(self) -> None:
    """Increment the pool acquisitions counter."""
    if not self._enabled:
        return

    self.pool_acquisitions_total.inc()

inc_pool_releases

inc_pool_releases()

Increment the pool releases counter.

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

def inc_pool_releases(self) -> None:
    """Increment the pool releases counter."""
    if not self._enabled:
        return

    self.pool_releases_total.inc()

inc_pool_health_check_failures

inc_pool_health_check_failures()

Increment the pool health check failures counter.

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

def inc_pool_health_check_failures(self) -> None:
    """Increment the pool health check failures counter."""
    if not self._enabled:
        return

    self.pool_health_check_failures_total.inc()

set_pool_clients_total

set_pool_clients_total(count)

Set the total number of clients in the pool.

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

def set_pool_clients_total(self, count: int) -> None:
    """Set the total number of clients in the pool."""
    if not self._enabled:
        return

    self.pool_clients_total.set(count)

set_pool_clients_available

set_pool_clients_available(count)

Set the number of available clients in the pool.

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

def set_pool_clients_available(self, count: int) -> None:
    """Set the number of available clients in the pool."""
    if not self._enabled:
        return

    self.pool_clients_available.set(count)

set_pool_clients_active

set_pool_clients_active(count)

Set the number of active clients in the pool.

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

def set_pool_clients_active(self, count: int) -> None:
    """Set the number of active clients in the pool."""
    if not self._enabled:
        return

    self.pool_clients_active.set(count)