ccproxy.observability.metrics

ccproxy.observability.metrics

Prometheus metrics for operational monitoring.

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 - Pushgateway integration for batch metric pushing

PrometheusMetrics

PrometheusMetrics(
    namespace="ccproxy",
    registry=None,
    pushgateway_client=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
pushgateway_client	Any | None	Optional pushgateway client for dependency injection	None

Source code in ccproxy/observability/metrics.py

def __init__(
    self,
    namespace: str = "ccproxy",
    registry: CollectorRegistry | None = None,
    pushgateway_client: Any | None = None,
):
    """
    Initialize Prometheus metrics.

    Args:
        namespace: Metric name prefix
        registry: Custom Prometheus registry (uses default if None)
        pushgateway_client: Optional pushgateway client for dependency injection
    """
    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._pushgateway_client = pushgateway_client

    if self._enabled:
        self._init_metrics()
        # Initialize pushgateway client if not provided via DI
        if self._pushgateway_client is None:
            self._init_pushgateway()

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/observability/metrics.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/observability/metrics.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/observability/metrics.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/observability/metrics.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/observability/metrics.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/observability/metrics.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/observability/metrics.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/observability/metrics.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/observability/metrics.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/observability/metrics.py

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

push_to_gateway

push_to_gateway(method='push')

Push current metrics to Pushgateway using official prometheus_client methods.

Parameters:

Name	Type	Description	Default
method	str	Push method - "push" (replace), "pushadd" (add), or "delete"	'push'

Returns:

Type	Description
bool	True if push succeeded, False otherwise

Source code in ccproxy/observability/metrics.py

def push_to_gateway(self, method: str = "push") -> bool:
    """
    Push current metrics to Pushgateway using official prometheus_client methods.

    Args:
        method: Push method - "push" (replace), "pushadd" (add), or "delete"

    Returns:
        True if push succeeded, False otherwise
    """

    if not self._enabled or not self._pushgateway_client:
        return False

    result = self._pushgateway_client.push_metrics(self.registry, method)
    return bool(result)

push_add_to_gateway

push_add_to_gateway()

Add current metrics to existing job/instance in Pushgateway (pushadd operation).

This is useful when you want to add metrics without replacing existing ones.

Returns:

Type	Description
bool	True if push succeeded, False otherwise

Source code in ccproxy/observability/metrics.py

def push_add_to_gateway(self) -> bool:
    """
    Add current metrics to existing job/instance in Pushgateway (pushadd operation).

    This is useful when you want to add metrics without replacing existing ones.

    Returns:
        True if push succeeded, False otherwise
    """
    return self.push_to_gateway(method="pushadd")

delete_from_gateway

delete_from_gateway()

Delete all metrics for the configured job from Pushgateway.

This removes all metrics associated with the job, useful for cleanup.

Returns:

Type	Description
bool	True if delete succeeded, False otherwise

Source code in ccproxy/observability/metrics.py

def delete_from_gateway(self) -> bool:
    """
    Delete all metrics for the configured job from Pushgateway.

    This removes all metrics associated with the job, useful for cleanup.

    Returns:
        True if delete succeeded, False otherwise
    """

    if not self._enabled or not self._pushgateway_client:
        return False

    result = self._pushgateway_client.delete_metrics()
    return bool(result)

is_pushgateway_enabled

is_pushgateway_enabled()

Check if Pushgateway client is enabled and configured.

Source code in ccproxy/observability/metrics.py

def is_pushgateway_enabled(self) -> bool:
    """Check if Pushgateway client is enabled and configured."""
    return (
        self._pushgateway_client is not None
        and self._pushgateway_client.is_enabled()
    )

get_metrics

get_metrics(
    namespace="ccproxy",
    registry=None,
    pushgateway_client=None,
    settings=None,
)

Get or create global metrics instance with dependency injection.

Parameters:

Name	Type	Description	Default
namespace	str	Metric namespace prefix	'ccproxy'
registry	CollectorRegistry | None	Custom Prometheus registry	None
pushgateway_client	Any | None	Optional pushgateway client for dependency injection	None
settings	Any | None	Optional settings instance to avoid circular imports	None

Returns:

Type	Description
PrometheusMetrics	PrometheusMetrics instance with full pushgateway support:
PrometheusMetrics	push_to_gateway(): Replace all metrics (default)
PrometheusMetrics	push_add_to_gateway(): Add metrics to existing job
PrometheusMetrics	delete_from_gateway(): Delete all metrics for job

Source code in ccproxy/observability/metrics.py

def get_metrics(
    namespace: str = "ccproxy",
    registry: CollectorRegistry | None = None,
    pushgateway_client: Any | None = None,
    settings: Any | None = None,
) -> PrometheusMetrics:
    """
    Get or create global metrics instance with dependency injection.

    Args:
        namespace: Metric namespace prefix
        registry: Custom Prometheus registry
        pushgateway_client: Optional pushgateway client for dependency injection
        settings: Optional settings instance to avoid circular imports

    Returns:
        PrometheusMetrics instance with full pushgateway support:
        - push_to_gateway(): Replace all metrics (default)
        - push_add_to_gateway(): Add metrics to existing job
        - delete_from_gateway(): Delete all metrics for job
    """
    global _global_metrics

    if _global_metrics is None:
        # Create pushgateway client if not provided via DI
        if pushgateway_client is None:
            from .pushgateway import get_pushgateway_client

            pushgateway_client = get_pushgateway_client()

        _global_metrics = PrometheusMetrics(
            namespace=namespace,
            registry=registry,
            pushgateway_client=pushgateway_client,
        )

    return _global_metrics

reset_metrics

reset_metrics()

Reset global metrics instance (mainly for testing).

Source code in ccproxy/observability/metrics.py

def reset_metrics() -> None:
    """Reset global metrics instance (mainly for testing)."""
    global _global_metrics
    _global_metrics = None

    # Clear Prometheus registry to avoid duplicate metrics in tests
    if PROMETHEUS_AVAILABLE:
        try:
            from prometheus_client import REGISTRY

            # Clear all collectors from the registry
            collectors = list(REGISTRY._collector_to_names.keys())
            for collector in collectors:
                REGISTRY.unregister(collector)
        except Exception:
            # If clearing the registry fails, just continue
            # This is mainly for testing and shouldn't break functionality
            pass

    # Also reset pushgateway client
    from .pushgateway import reset_pushgateway_client

    reset_pushgateway_client()