tessl/pypi-prometheus-client

Python client for the Prometheus monitoring system.

—

Pending

Overview

Eval results

Files

Built-in Collectors

Name: tessl/pypi-prometheus-client
Author: tessl

Pre-built collectors for system metrics including process statistics, garbage collection data, and platform information. These collectors can be automatically registered to provide comprehensive monitoring of application and system health.

Capabilities

ProcessCollector

Collects system process metrics including CPU usage, memory consumption, file descriptors, and other process-level statistics. Useful for monitoring application resource usage.

class ProcessCollector:
    def __init__(
        self, 
        namespace: str = '', 
        pid: Callable = lambda: 'self', 
        proc: str = '/proc', 
        registry: Optional[CollectorRegistry] = REGISTRY
    ) -> None:
        """
        Create a ProcessCollector.
        
        Parameters:
        - namespace: Metric namespace prefix
        - pid: Function returning process ID ('self' for current process, or callable returning PID)
        - proc: Path to /proc filesystem (for Linux/Unix systems)
        - registry: Registry to register with (None to skip auto-registration)
        """
    
    def collect(self) -> Iterable[Metric]:
        """
        Collect process metrics.
        
        Returns:
        Iterable of Metric objects containing process statistics
        """

PROCESS_COLLECTOR: ProcessCollector = ProcessCollector()

Metrics Provided:

process_cpu_seconds_total - Total user and system CPU time
process_open_fds - Number of open file descriptors
process_max_fds - Maximum number of open file descriptors
process_virtual_memory_bytes - Virtual memory size in bytes
process_resident_memory_bytes - Resident memory size in bytes
process_start_time_seconds - Start time of the process since Unix epoch

Usage Example:

from prometheus_client import CollectorRegistry, generate_latest
from prometheus_client.process_collector import ProcessCollector

# Use default process collector (already registered with REGISTRY)
from prometheus_client import REGISTRY
print(generate_latest(REGISTRY).decode('utf-8'))

# Create custom process collector
custom_registry = CollectorRegistry()
process_collector = ProcessCollector(
    namespace='myapp',
    registry=custom_registry
)

# Metrics will be prefixed with 'myapp_'
output = generate_latest(custom_registry)
print(output.decode('utf-8'))

# Monitor specific process
import os
def get_child_pid():
    return str(child_process.pid)

child_collector = ProcessCollector(
    namespace='child',
    pid=get_child_pid,
    registry=custom_registry
)

# Disable auto-registration
unregistered_collector = ProcessCollector(registry=None)
# Manually register later if needed
custom_registry.register(unregistered_collector)

GCCollector

Collects Python garbage collection statistics including collection counts and timing information for each generation. Helps monitor memory management performance.

class GCCollector:
    def __init__(self, registry: CollectorRegistry = REGISTRY) -> None:
        """
        Create a GCCollector.
        
        Parameters:
        - registry: Registry to register with (None to skip auto-registration)
        """
    
    def collect(self) -> Iterable[Metric]:
        """
        Collect garbage collection metrics.
        
        Returns:
        Iterable of Metric objects containing GC statistics
        """

GC_COLLECTOR: GCCollector = GCCollector()

Metrics Provided:

python_gc_objects_collected_total - Objects collected during GC by generation
python_gc_objects_uncollectable_total - Uncollectable objects found by GC by generation
python_gc_collections_total - Number of times this generation was collected

Usage Example:

from prometheus_client import CollectorRegistry, generate_latest
from prometheus_client.gc_collector import GCCollector
import gc

# Use default GC collector (already registered with REGISTRY)
from prometheus_client import REGISTRY

# Force some garbage collection activity
data = []
for i in range(10000):
    data.append({'id': i, 'data': list(range(100))})
del data
gc.collect()

# Check GC metrics
output = generate_latest(REGISTRY)
print(output.decode('utf-8'))

# Create custom GC collector
custom_registry = CollectorRegistry()
gc_collector = GCCollector(registry=custom_registry)

# Monitor GC with custom namespace
class NamespacedGCCollector(GCCollector):
    def __init__(self, namespace='', registry=REGISTRY):
        self.namespace = namespace
        super().__init__(registry=None)  # Don't auto-register
        if registry:
            registry.register(self)
    
    def collect(self):
        for metric in super().collect():
            # Add namespace to metric name
            if self.namespace:
                metric.name = f"{self.namespace}_{metric.name}"
            yield metric

app_gc_collector = NamespacedGCCollector(namespace='myapp', registry=custom_registry)

PlatformCollector

Collects Python platform and interpreter information including version, implementation, and build details. Useful for environment monitoring and debugging.

class PlatformCollector:
    def __init__(
        self, 
        registry: Optional[CollectorRegistry] = REGISTRY, 
        platform=None
    ) -> None:
        """
        Create a PlatformCollector.
        
        Parameters:
        - registry: Registry to register with (None to skip auto-registration)
        - platform: Platform module to use (defaults to built-in platform module)
        """
    
    def collect(self) -> Iterable[Metric]:
        """
        Collect platform information metrics.
        
        Returns:
        Iterable of Metric objects containing platform information
        """

PLATFORM_COLLECTOR: PlatformCollector = PlatformCollector()

Metrics Provided:

python_info - Python platform information (version, implementation, etc.)

Usage Example:

from prometheus_client import CollectorRegistry, generate_latest
from prometheus_client.platform_collector import PlatformCollector

# Use default platform collector (already registered with REGISTRY)
from prometheus_client import REGISTRY
output = generate_latest(REGISTRY)
print(output.decode('utf-8'))

# Example output includes:
# python_info{implementation="CPython",major="3",minor="9",patchlevel="7",version="3.9.7"} 1.0

# Create custom platform collector
custom_registry = CollectorRegistry()
platform_collector = PlatformCollector(registry=custom_registry)

# Custom platform information for testing
class MockPlatform:
    def python_version_tuple(self):
        return ('3', '10', '0')
    
    def python_implementation(self):
        return 'PyPy'

test_collector = PlatformCollector(
    registry=custom_registry,
    platform=MockPlatform()
)

Collector Registration and Management

Utilities for managing built-in collectors in custom registries.

Usage Example:

from prometheus_client import CollectorRegistry
from prometheus_client.process_collector import ProcessCollector, PROCESS_COLLECTOR
from prometheus_client.gc_collector import GCCollector, GC_COLLECTOR
from prometheus_client.platform_collector import PlatformCollector, PLATFORM_COLLECTOR

# Create registry without default collectors
custom_registry = CollectorRegistry()

# Add only specific collectors
ProcessCollector(registry=custom_registry)
GCCollector(registry=custom_registry)
# Skip platform collector for this registry

# Remove collectors from default registry if needed
from prometheus_client import REGISTRY
REGISTRY.unregister(PROCESS_COLLECTOR)
REGISTRY.unregister(GC_COLLECTOR)
REGISTRY.unregister(PLATFORM_COLLECTOR)

# Re-register with custom settings
ProcessCollector(namespace='app', registry=REGISTRY)
GCCollector(registry=REGISTRY)
PlatformCollector(registry=REGISTRY)

# Check what's registered
collectors = []
for metric in REGISTRY.collect():
    collectors.append(metric.name)
print("Registered metrics:", collectors)

Combined Collector Usage

Example of using all built-in collectors together for comprehensive system monitoring.

Usage Example:

from prometheus_client import (
    CollectorRegistry, Counter, Gauge, generate_latest,
    start_http_server
)
from prometheus_client.process_collector import ProcessCollector  
from prometheus_client.gc_collector import GCCollector
from prometheus_client.platform_collector import PlatformCollector
import time
import threading
import random

# Create application registry with all system collectors
app_registry = CollectorRegistry()

# Add system collectors with namespace
ProcessCollector(namespace='myapp', registry=app_registry)
GCCollector(registry=app_registry) 
PlatformCollector(registry=app_registry)

# Add application metrics
request_count = Counter(
    'myapp_requests_total', 
    'Total application requests',
    ['endpoint'],
    registry=app_registry
)

active_connections = Gauge(
    'myapp_active_connections',
    'Active client connections', 
    registry=app_registry
)

# Start metrics server
start_http_server(8000, registry=app_registry)

# Simulate application activity
def app_activity():
    endpoints = ['/api/users', '/api/orders', '/health']
    
    while True:
        # Simulate requests
        endpoint = random.choice(endpoints)
        request_count.labels(endpoint=endpoint).inc()
        
        # Simulate connection changes
        active_connections.set(random.randint(10, 100))
        
        # Create some garbage to trigger GC
        temp_data = [list(range(1000)) for _ in range(10)]
        del temp_data
        
        time.sleep(1)

# Run application simulation
app_thread = threading.Thread(target=app_activity, daemon=True)
app_thread.start()

print("Application metrics server running on http://localhost:8000")
print("Metrics include:")
print("- Application metrics (requests, connections)")  
print("- Process metrics (CPU, memory, file descriptors)")
print("- Python GC metrics (collections, objects)")
print("- Platform info (Python version, implementation)")

try:
    while True:
        time.sleep(10)
        print(f"Current metrics count: {len(list(app_registry.collect()))}")
except KeyboardInterrupt:
    print("Shutting down...")

Custom Collector Integration

How to create custom collectors that work alongside built-in collectors.

Usage Example:

from prometheus_client import Collector, Metric, CollectorRegistry
from prometheus_client.process_collector import ProcessCollector
import psutil
import time

class SystemCollector(Collector):
    """Custom collector for additional system metrics."""
    
    def collect(self):
        # Disk usage
        disk_metric = Metric('system_disk_usage_bytes', 'Disk usage by mount point', 'gauge')
        for partition in psutil.disk_partitions():
            try:
                usage = psutil.disk_usage(partition.mountpoint)
                disk_metric.add_sample(
                    'system_disk_usage_bytes',
                    {'mountpoint': partition.mountpoint, 'type': 'used'},
                    usage.used
                )
                disk_metric.add_sample(
                    'system_disk_usage_bytes', 
                    {'mountpoint': partition.mountpoint, 'type': 'free'},
                    usage.free
                )
            except PermissionError:
                continue
        yield disk_metric
        
        # Network I/O
        net_io = psutil.net_io_counters()
        net_metric = Metric('system_network_bytes_total', 'Network I/O bytes', 'counter')
        net_metric.add_sample('system_network_bytes_total', {'direction': 'sent'}, net_io.bytes_sent)
        net_metric.add_sample('system_network_bytes_total', {'direction': 'received'}, net_io.bytes_recv)
        yield net_metric

# Create comprehensive monitoring registry
monitoring_registry = CollectorRegistry()

# Add built-in collectors
ProcessCollector(namespace='app', registry=monitoring_registry)
GCCollector(registry=monitoring_registry)
PlatformCollector(registry=monitoring_registry)

# Add custom collector
system_collector = SystemCollector()
monitoring_registry.register(system_collector)

# Generate comprehensive metrics
output = generate_latest(monitoring_registry)
print("Comprehensive system metrics:")
print(output.decode('utf-8'))

Install with Tessl CLI

npx tessl i tessl/pypi-prometheus-client

docs

tessl/pypi-prometheus-client

collectors.md.css-3qkkll{font-size:var(--chakra-font-sizes-sm);font-weight:var(--chakra-font-weights-normal);color:var(--chakra-colors-gray-300);}docs/

Built-in Collectors

Capabilities

ProcessCollector

GCCollector

PlatformCollector

Collector Registration and Management

Combined Collector Usage

Custom Collector Integration

collectors.mddocs/