tessl/pypi-prometheus-client
Python client for the Prometheus monitoring system.
—
Pending
registry.mddocs/
Registry System
The registry system provides centralized management of metrics collection, organization, and retrieval. It serves as the collection point for all metrics and enables custom collectors to be integrated into the monitoring system.
Capabilities
CollectorRegistry
The main registry class that manages metric collectors and provides unified access to all metrics. It supports automatic registration, custom collectors, and multiple exposition formats.
class CollectorRegistry:
def __init__(self, auto_describe: bool = False, target_info: Optional[Dict[str, str]] = None) -> None:
"""
Create a new CollectorRegistry.
Parameters:
- auto_describe: Whether to auto-call describe() on registered collectors
- target_info: Optional target info labels for this registry
"""
def register(self, collector: Collector) -> None:
"""
Register a collector with this registry.
Parameters:
- collector: Collector instance to register
Raises:
ValueError: If collector is already registered or conflicts with existing collector
"""
def unregister(self, collector: Collector) -> None:
"""
Unregister a collector from this registry.
Parameters:
- collector: Collector instance to unregister
Raises:
KeyError: If collector is not registered
"""
def collect(self) -> Iterable[Metric]:
"""
Collect metrics from all registered collectors.
Returns:
Iterable of Metric objects from all collectors
"""
def restricted_registry(self, names: Iterable[str]) -> "RestrictedRegistry":
"""
Create a restricted view of this registry containing only specified metrics.
Parameters:
- names: Metric names to include in restricted view
Returns:
RestrictedRegistry instance
"""
def set_target_info(self, labels: Optional[Dict[str, str]]) -> None:
"""
Set target info labels for this registry.
Parameters:
- labels: Dictionary of target info labels, or None to clear
"""
def get_target_info(self) -> Optional[Dict[str, str]]:
"""
Get current target info labels.
Returns:
Dictionary of target info labels, or None if not set
"""
def get_sample_value(self, name: str, labels: Optional[Dict[str, str]] = None) -> Optional[float]:
"""
Get the current sample value for a metric.
Parameters:
- name: Metric name
- labels: Optional labels dictionary
Returns:
Current sample value, or None if not found
"""Usage Example:
from prometheus_client import CollectorRegistry, Counter, Gauge
# Create custom registry
custom_registry = CollectorRegistry()
# Create metrics with custom registry
requests = Counter('requests_total', 'Total requests', registry=custom_registry)
memory = Gauge('memory_usage', 'Memory usage', registry=custom_registry)
# Use metrics
requests.inc()
memory.set(1024)
# Get sample values
print(custom_registry.get_sample_value('requests_total'))
print(custom_registry.get_sample_value('memory_usage'))
# Create restricted view
restricted = custom_registry.restricted_registry(['requests_total'])
# Set target info
custom_registry.set_target_info({
'service': 'my-app',
'version': '1.0.0'
})Global Registry
The default global registry instance used by all metrics when no specific registry is provided.
REGISTRY: CollectorRegistry = CollectorRegistry(auto_describe=True)Usage Example:
from prometheus_client import REGISTRY, Counter, generate_latest
# Metrics automatically register with REGISTRY
counter = Counter('my_counter', 'My counter')
counter.inc()
# Generate output from global registry
metrics_output = generate_latest(REGISTRY)
print(metrics_output.decode('utf-8'))
# Register custom collector with global registry
class MyCollector:
def collect(self):
# Return metrics
pass
REGISTRY.register(MyCollector())Collector Interface
The abstract base class for all collectors that can be registered with a registry.
class Collector:
def collect(self) -> Iterable[Metric]:
"""
Collect and return metrics.
This method must be implemented by all collectors.
Returns:
Iterable of Metric objects
"""
def describe(self) -> Iterable[Metric]:
"""
Describe the metrics that this collector will provide.
Default implementation returns collect() results.
Override for better performance if metrics are expensive to collect.
Returns:
Iterable of Metric objects (without samples)
"""Usage Example:
from prometheus_client import Collector, Metric, REGISTRY
import psutil
class SystemCollector(Collector):
def collect(self):
# CPU usage metric
cpu_metric = Metric('system_cpu_percent', 'CPU usage percentage', 'gauge')
cpu_metric.add_sample('system_cpu_percent', {}, psutil.cpu_percent())
yield cpu_metric
# Memory usage metric
memory = psutil.virtual_memory()
memory_metric = Metric('system_memory_bytes', 'Memory usage in bytes', 'gauge')
memory_metric.add_sample('system_memory_bytes', {'type': 'used'}, memory.used)
memory_metric.add_sample('system_memory_bytes', {'type': 'available'}, memory.available)
yield memory_metric
# Register custom collector
system_collector = SystemCollector()
REGISTRY.register(system_collector)
# Later, unregister if needed
REGISTRY.unregister(system_collector)Metric Class for Custom Collectors
The core Metric class used to represent metrics and their samples in custom collectors.
class Metric:
def __init__(self, name: str, documentation: str, typ: str, unit: str = '') -> None:
"""
Create a new Metric.
Parameters:
- name: Metric name
- documentation: Help text describing the metric
- typ: Metric type ('counter', 'gauge', 'histogram', 'summary', 'info', 'stateset', 'unknown')
- unit: Optional unit suffix
"""
def add_sample(
self,
name: str,
labels: Dict[str, str],
value: float,
timestamp=None,
exemplar=None,
native_histogram=None
) -> None:
"""
Add a sample to this metric.
Parameters:
- name: Sample name (usually metric name + suffix)
- labels: Dictionary of label names to values
- value: Sample value
- timestamp: Optional timestamp (Unix time)
- exemplar: Optional exemplar for tracing
- native_histogram: Optional native histogram value
"""
# Properties
name: str # Metric name
documentation: str # Help text
type: str # Metric type
unit: str # Unit suffix
samples: List[Sample] # List of samplesUsage Example:
from prometheus_client import Metric
import time
# Create a metric
metric = Metric('my_custom_metric', 'A custom metric', 'gauge', unit='bytes')
# Add samples
metric.add_sample('my_custom_metric_bytes', {'instance': 'server1'}, 1024.0)
metric.add_sample('my_custom_metric_bytes', {'instance': 'server2'}, 2048.0)
# Add sample with timestamp
metric.add_sample(
'my_custom_metric_bytes',
{'instance': 'server3'},
4096.0,
timestamp=time.time()
)
# Access metric properties
print(f"Name: {metric.name}")
print(f"Type: {metric.type}")
print(f"Samples: {len(metric.samples)}")Registry Utilities
Helper functions for working with registries and metrics.
def generate_latest(registry: CollectorRegistry = REGISTRY) -> bytes:
"""
Generate the latest Prometheus text format output.
Parameters:
- registry: Registry to collect metrics from
Returns:
Metrics in Prometheus text format as bytes
"""Usage Example:
from prometheus_client import CollectorRegistry, Counter, generate_latest
# Create registry with metrics
registry = CollectorRegistry()
counter = Counter('test_counter', 'Test counter', registry=registry)
counter.inc(42)
# Generate output
output = generate_latest(registry)
print(output.decode('utf-8'))
# Example output:
# # HELP test_counter_total Test counter
# # TYPE test_counter_total counter
# test_counter_total 42.0Multiprocess Registry Support
Support for collecting metrics across multiple processes using shared memory.
class MultiProcessCollector:
def __init__(self, registry, path=None) -> None:
"""
Create a MultiProcessCollector.
Parameters:
- registry: Registry to register with
- path: Path to multiprocess metrics directory (defaults to PROMETHEUS_MULTIPROC_DIR env var)
"""
@staticmethod
def merge(files, accumulate=True):
"""
Merge metrics from multiple process files.
Parameters:
- files: List of file paths containing metrics
- accumulate: Whether to accumulate values across processes
Returns:
Dictionary of merged metrics
"""Usage Example:
import os
from prometheus_client import CollectorRegistry, Counter
from prometheus_client.multiprocess import MultiProcessCollector
# Set multiprocess directory
os.environ['PROMETHEUS_MULTIPROC_DIR'] = '/tmp/prometheus_metrics'
# Create registry with multiprocess support
registry = CollectorRegistry()
MultiProcessCollector(registry)
# Create metrics (will be shared across processes)
counter = Counter('worker_tasks', 'Tasks processed by worker', registry=registry)
# In worker processes
counter.inc() # Each process can increment independently
# Parent process can collect aggregated metrics
from prometheus_client import generate_latest
print(generate_latest(registry).decode('utf-8'))Install with Tessl CLI
npx tessl i tessl/pypi-prometheus-client