tessl/pypi-prometheus-client
Python client for the Prometheus monitoring system.
—
Pending
core-metrics.mddocs/
Core Metrics
The fundamental metric types in prometheus-client for collecting different kinds of measurements. Each metric type is designed for specific use cases and provides thread-safe operations with support for multi-dimensional labeling.
Capabilities
Counter
A monotonically increasing counter that can only increase or be reset to zero. Counters are ideal for measuring cumulative values like total requests, errors, or completed tasks.
class Counter:
def __init__(
self,
name: str,
documentation: str,
labelnames: Iterable[str] = (),
namespace: str = '',
subsystem: str = '',
unit: str = '',
registry: Optional[CollectorRegistry] = REGISTRY,
_labelvalues: Optional[Sequence[str]] = None
) -> None:
"""
Create a new Counter metric.
Parameters:
- name: Metric name (without _total suffix)
- documentation: Help text describing the metric
- labelnames: Sequence of label names for multi-dimensional metrics
- namespace: Optional namespace prefix
- subsystem: Optional subsystem prefix
- unit: Optional unit suffix
- registry: Registry to register this metric with
"""
def inc(self, amount: float = 1, exemplar: Optional[Dict[str, str]] = None) -> None:
"""
Increment the counter by amount.
Parameters:
- amount: Amount to increment by (must be >= 0)
- exemplar: Optional exemplar for tracing
"""
def reset(self) -> None:
"""Reset the counter to zero."""
def count_exceptions(self, exception=Exception) -> ExceptionCounter:
"""
Return a context manager that counts exceptions.
Parameters:
- exception: Exception type to count
Returns:
ExceptionCounter context manager
"""
def labels(self, *labelvalues, **labelkwargs) -> Counter:
"""
Return the Counter for the given label values.
Parameters:
- labelvalues: Label values in order of labelnames
- labelkwargs: Label values as keyword arguments
Returns:
Counter instance for the specific label combination
"""
def remove(self, *labelvalues) -> None:
"""Remove a labeled metric."""
def clear(self) -> None:
"""Remove all labeled metrics."""
def describe(self) -> Iterable[Metric]:
"""Return metric metadata."""
def collect(self) -> Iterable[Metric]:
"""Return current metric samples."""
def __str__(self) -> str:
"""Return string representation."""
def __repr__(self) -> str:
"""Return detailed string representation."""Usage Example:
from prometheus_client import Counter
# Simple counter
requests_total = Counter('http_requests_total', 'Total HTTP requests')
requests_total.inc() # Increment by 1
requests_total.inc(5) # Increment by 5
# Counter with labels
requests_by_status = Counter(
'http_requests_total',
'Total HTTP requests',
['method', 'status']
)
requests_by_status.labels(method='GET', status='200').inc()
requests_by_status.labels('POST', '404').inc(2)
# Exception counting
with requests_total.count_exceptions():
# Code that might raise exceptions
risky_operation()Gauge
A gauge represents a single numerical value that can go up or down. Gauges are suitable for measured values like temperatures, current memory usage, or number of concurrent requests.
class Gauge:
def __init__(
self,
name: str,
documentation: str,
labelnames: Iterable[str] = (),
namespace: str = '',
subsystem: str = '',
unit: str = '',
registry: Optional[CollectorRegistry] = REGISTRY,
_labelvalues: Optional[Sequence[str]] = None,
multiprocess_mode: Literal['all', 'liveall', 'min', 'livemin', 'max', 'livemax', 'sum', 'livesum', 'mostrecent', 'livemostrecent'] = 'all'
) -> None:
"""
Create a new Gauge metric.
Parameters:
- name: Metric name
- documentation: Help text describing the metric
- labelnames: Sequence of label names for multi-dimensional metrics
- namespace: Optional namespace prefix
- subsystem: Optional subsystem prefix
- unit: Optional unit suffix
- registry: Registry to register this metric with
- multiprocess_mode: How to aggregate in multiprocess mode ('all', 'liveall', 'min', 'livemin', 'max', 'livemax', 'sum', 'livesum', 'mostrecent', 'livemostrecent')
"""
def set(self, value: float) -> None:
"""Set the gauge to the given value."""
def inc(self, amount: float = 1) -> None:
"""Increment the gauge by amount."""
def dec(self, amount: float = 1) -> None:
"""Decrement the gauge by amount."""
def set_to_current_time(self) -> None:
"""Set the gauge to the current Unix timestamp."""
def track_inprogress(self) -> InprogressTracker:
"""
Return a context manager that tracks in-progress operations.
Returns:
InprogressTracker context manager
"""
def time(self) -> Timer:
"""
Return a context manager/decorator that times operations.
Returns:
Timer context manager
"""
def set_function(self, f: Callable[[], float]) -> None:
"""
Set the gauge to the return value of a function.
Parameters:
- f: Function that returns a float value
"""
def labels(self, *labelvalues, **labelkwargs) -> Gauge:
"""Return the Gauge for the given label values."""
def remove(self, *labelvalues) -> None:
"""Remove a labeled metric."""
def clear(self) -> None:
"""Remove all labeled metrics."""
def describe(self) -> Iterable[Metric]:
"""Return metric metadata."""
def collect(self) -> Iterable[Metric]:
"""Return current metric samples."""
def __str__(self) -> str:
"""Return string representation."""
def __repr__(self) -> str:
"""Return detailed string representation."""Usage Example:
from prometheus_client import Gauge
import psutil
# Simple gauge
memory_usage = Gauge('memory_usage_bytes', 'Current memory usage in bytes')
memory_usage.set(psutil.virtual_memory().used)
# Gauge with labels
cpu_usage = Gauge('cpu_usage_percent', 'CPU usage by core', ['core'])
for i, percent in enumerate(psutil.cpu_percent(percpu=True)):
cpu_usage.labels(core=str(i)).set(percent)
# Auto-updating gauge
def get_disk_usage():
return psutil.disk_usage('/').percent
disk_gauge = Gauge('disk_usage_percent', 'Disk usage percentage')
disk_gauge.set_function(get_disk_usage)
# Timing operations
response_time = Gauge('response_time_seconds', 'Response time')
with response_time.time():
# Timed operation
time.sleep(0.1)
# Tracking in-progress work
active_requests = Gauge('active_requests', 'Number of active requests')
with active_requests.track_inprogress():
# Process request
handle_request()Histogram
A histogram samples observations and counts them in configurable buckets, also providing sum and count of observations. Histograms are ideal for measuring request durations, response sizes, or any value where you want to understand the distribution.
class Histogram:
DEFAULT_BUCKETS = (.005, .01, .025, .05, .075, .1, .25, .5, .75, 1.0, 2.5, 5.0, 7.5, 10.0, INF)
def __init__(
self,
name: str,
documentation: str,
labelnames: Iterable[str] = (),
namespace: str = '',
subsystem: str = '',
unit: str = '',
registry: Optional[CollectorRegistry] = REGISTRY,
_labelvalues: Optional[Sequence[str]] = None,
buckets: Sequence[Union[float, str]] = DEFAULT_BUCKETS
) -> None:
"""
Create a new Histogram metric.
Parameters:
- name: Metric name
- documentation: Help text describing the metric
- labelnames: Sequence of label names for multi-dimensional metrics
- namespace: Optional namespace prefix
- subsystem: Optional subsystem prefix
- unit: Optional unit suffix
- registry: Registry to register this metric with
- buckets: Sequence of bucket upper bounds
"""
def observe(self, amount: float, exemplar: Optional[Dict[str, str]] = None) -> None:
"""
Observe a value.
Parameters:
- amount: Value to observe
- exemplar: Optional exemplar for tracing
"""
def time(self) -> Timer:
"""
Return a context manager/decorator that times operations.
Returns:
Timer context manager that observes elapsed time
"""
def labels(self, *labelvalues, **labelkwargs) -> Histogram:
"""Return the Histogram for the given label values."""
def remove(self, *labelvalues) -> None:
"""Remove a labeled metric."""
def clear(self) -> None:
"""Remove all labeled metrics."""Usage Example:
from prometheus_client import Histogram
import time
import random
# Request duration histogram
request_duration = Histogram(
'http_request_duration_seconds',
'HTTP request duration in seconds',
['method', 'endpoint']
)
# Observe values
request_duration.labels('GET', '/api/users').observe(0.25)
request_duration.labels('POST', '/api/users').observe(0.8)
# Custom buckets for specific use case
response_size = Histogram(
'http_response_size_bytes',
'HTTP response size in bytes',
buckets=[100, 1000, 10000, 100000, 1000000, float('inf')]
)
response_size.observe(1024)
# Time operations automatically
with request_duration.labels('GET', '/api/health').time():
# Simulated work
time.sleep(random.uniform(0.1, 0.5))
# As a decorator
@request_duration.labels('POST', '/api/data').time()
def process_data():
time.sleep(random.uniform(0.2, 1.0))
return "processed"Summary
A summary samples observations and provides count and sum of observations, plus configurable quantiles over a sliding time window. Summaries are useful when you need quantile information (like median, 95th percentile) but want to avoid the overhead of histograms.
class Summary:
def __init__(
self,
name: str,
documentation: str,
labelnames=(),
namespace='',
subsystem='',
unit='',
registry=REGISTRY,
_labelvalues=None
) -> None:
"""
Create a new Summary metric.
Parameters:
- name: Metric name
- documentation: Help text describing the metric
- labelnames: Sequence of label names for multi-dimensional metrics
- namespace: Optional namespace prefix
- subsystem: Optional subsystem prefix
- unit: Optional unit suffix
- registry: Registry to register this metric with
"""
def observe(self, amount: float) -> None:
"""
Observe a value.
Parameters:
- amount: Value to observe
"""
def time(self) -> Timer:
"""
Return a context manager/decorator that times operations.
Returns:
Timer context manager that observes elapsed time
"""
def labels(self, *labelvalues, **labelkwargs) -> Summary:
"""Return the Summary for the given label values."""
def remove(self, *labelvalues) -> None:
"""Remove a labeled metric."""
def clear(self) -> None:
"""Remove all labeled metrics."""Usage Example:
from prometheus_client import Summary
import time
import random
# Response time summary
response_time = Summary(
'http_response_time_seconds',
'HTTP response time in seconds',
['service']
)
# Observe values
response_time.labels('auth').observe(0.12)
response_time.labels('database').observe(0.045)
# Time operations
with response_time.labels('api').time():
# Simulated API call
time.sleep(random.uniform(0.05, 0.2))Info
An info metric represents a set of key-value pairs. It is useful for exposing textual information like version numbers, build information, or configuration details.
class Info:
def __init__(
self,
name: str,
documentation: str,
labelnames=(),
namespace='',
subsystem='',
registry=REGISTRY,
_labelvalues=None
) -> None:
"""
Create a new Info metric.
Parameters:
- name: Metric name
- documentation: Help text describing the metric
- labelnames: Sequence of label names for multi-dimensional metrics
- namespace: Optional namespace prefix
- subsystem: Optional subsystem prefix
- registry: Registry to register this metric with
"""
def info(self, val: Dict[str, str]) -> None:
"""
Set the info metric with key-value pairs.
Parameters:
- val: Dictionary of string key-value pairs
"""
def labels(self, *labelvalues, **labelkwargs) -> Info:
"""Return the Info for the given label values."""
def remove(self, *labelvalues) -> None:
"""Remove a labeled metric."""
def clear(self) -> None:
"""Remove all labeled metrics."""Usage Example:
from prometheus_client import Info
import sys
import platform
# Application info
app_info = Info('app_info', 'Application information')
app_info.info({
'version': '1.2.3',
'python_version': sys.version.split()[0],
'platform': platform.system(),
'build_date': '2023-10-15'
})
# Environment-specific info
env_info = Info('environment_info', 'Environment information', ['env'])
env_info.labels('production').info({
'datacenter': 'us-east-1',
'cluster': 'prod-cluster-01'
})Enum
An enum metric represents a state set where exactly one of a set of states is true. It's useful for representing the current state of a system or component.
class Enum:
def __init__(
self,
name: str,
documentation: str,
labelnames=(),
namespace='',
subsystem='',
registry=REGISTRY,
_labelvalues=None,
states=None
) -> None:
"""
Create a new Enum metric.
Parameters:
- name: Metric name
- documentation: Help text describing the metric
- labelnames: Sequence of label names for multi-dimensional metrics
- namespace: Optional namespace prefix
- subsystem: Optional subsystem prefix
- registry: Registry to register this metric with
- states: Sequence of possible state names
"""
def state(self, state: str) -> None:
"""
Set the current state.
Parameters:
- state: Name of the current state
"""
def labels(self, *labelvalues, **labelkwargs) -> Enum:
"""Return the Enum for the given label values."""
def remove(self, *labelvalues) -> None:
"""Remove a labeled metric."""
def clear(self) -> None:
"""Remove all labeled metrics."""Usage Example:
from prometheus_client import Enum
# Service status
service_status = Enum(
'service_status',
'Current service status',
states=['starting', 'running', 'stopping', 'stopped']
)
# Set current state
service_status.state('running')
# With labels for multiple services
component_status = Enum(
'component_status',
'Component status',
['component'],
states=['healthy', 'degraded', 'unhealthy']
)
component_status.labels('database').state('healthy')
component_status.labels('cache').state('degraded')Configuration Functions
def enable_created_metrics() -> None:
"""Enable exporting _created metrics on counters, histograms, and summaries."""
def disable_created_metrics() -> None:
"""Disable exporting _created metrics on counters, histograms, and summaries."""Usage Example:
from prometheus_client import enable_created_metrics, disable_created_metrics
# Control _created series export
enable_created_metrics() # Enable timestamp tracking
disable_created_metrics() # Disable timestamp tracking (saves memory)Install with Tessl CLI
npx tessl i tessl/pypi-prometheus-client