← Back to Blog

Writing Custom Prometheus Exporters in Python for Microservices

How I built a reliable WhatsApp AI shopping assistant for Clickmothercare that survives hallucinated products, silent save failures, and multi-agent handoff bugs.

Anas Rhimi
Anas Rhimi August 2026 • 8 min read

Beyond Standard Metrics: The Need for Custom Exporters

While Node Exporter and cAdvisor provide excellent system-level metrics, deep observability requires domain-specific insights. When integrating with legacy databases or external APIs that don't natively expose Prometheus metrics, writing a custom Python exporter is the most effective pattern.

The Custom Collector Pattern

Instead of updating global gauge variables (which can lead to stale metrics if a resource disappears), we should implement the Custom Collector pattern. This ensures metrics are fetched dynamically at scrape time.

import time
from prometheus_client import start_http_server
from prometheus_client.core import GaugeMetricFamily, REGISTRY
import random

class PaymentGatewayCollector(object):
    def __init__(self, endpoint):
        self._endpoint = endpoint

    def collect(self):
        # 1. Initialize the metric family
        # We define a Gauge since queue sizes can go up and down
        metric = GaugeMetricFamily(
            'payment_gateway_queue_size',
            'Number of pending transactions in the gateway queue',
            labels=['gateway_provider']
        )
        
        # 2. Fetch the data (simulated here)
        # In a real scenario, this would be an API call or DB query
        stripe_queue = self._fetch_queue_size('stripe')
        paypal_queue = self._fetch_queue_size('paypal')
        
        # 3. Add metrics with specific label values
        metric.add_metric(['stripe'], stripe_queue)
        metric.add_metric(['paypal'], paypal_queue)
        
        # 4. Yield the metric family to the Prometheus client
        yield metric

    def _fetch_queue_size(self, provider):
        # Simulate network latency and data fetching
        time.sleep(0.1) 
        return random.randint(0, 100)

if __name__ == '__main__':
    # Unregister standard metrics if you only want your custom ones
    # REGISTRY.unregister(prometheus_client.GC_COLLECTOR)
    
    # Register our custom collector
    REGISTRY.register(PaymentGatewayCollector(endpoint="api.payments.internal"))
    
    # Start the HTTP server to expose metrics on /metrics
    start_http_server(8000)
    print("Prometheus exporter running on port 8000...")
    
    # Keep the main thread alive
    while True:
        time.sleep(1)

In this script, every time Prometheus scrapes :8000/metrics, the collect() method is invoked. The GaugeMetricFamily dynamically constructs the metric output. This guarantees that Prometheus always receives the most up-to-date state, avoiding the "stale gauge" problem common in poorly written exporters.

Is your AI agent's infrastructure secure and reliable?

Book a Free 15-Min Technical Audit
Hire Me