Writing Custom Prometheus Exporters in Python for Microservices

Beyond Standard Metrics: The Need for Custom Exporters

Standard exporters like Node Exporter and cAdvisor provide comprehensive architecture-level observability, capturing CPU usage, memory consumption, and container statistics. However, when application-specific telemetry is required such as tracking pending job queues in legacy databases, monitoring third-party API queue depths, or tracking domain-specific business indicators pre-built exporters are insufficient. Under these circumstances, developing a custom Python exporter offers a straightforward and flexible solution.

The Custom Collector Pattern

A frequent architectural anti-pattern when developing custom exporters involves mutating global Gauge or Counter instances inside background worker loops or web request handlers. If an underlying monitored dependency is decommissioned, persistent global metric instances can continue exporting stale, invalid data indefinitely.

To eliminate stale metric exports, official Prometheus client libraries provide support for custom collector classes. Rather than persisting metric state continuously, custom collectors compute and emit metric values dynamically on demand whenever Prometheus scrapes the /metrics endpoint.

Below is an implementation of a custom collector pattern written in Python:

import time
import random
from prometheus_client import start_http_server
from prometheus_client.core import GaugeMetricFamily, REGISTRY
class PaymentGatewayCollector(object):
 def __init__(self, endpoint):
 self._endpoint = endpoint
def collect(self):
 # 1. Define the metric family dynamically for each scrape
 metric = GaugeMetricFamily(
 'payment_gateway_queue_size',
 'Number of pending transactions in the gateway queue',
 labels=['gateway_provider']
 )
# 2. Fetch live data (simulated here with API or DB queries)
 stripe_queue = self._fetch_queue_size('stripe')
 paypal_queue = self._fetch_queue_size('paypal')
# 3. Attach metric values and labels
 metric.add_metric(['stripe'], stripe_queue)
 metric.add_metric(['paypal'], paypal_queue)
# 4. Yield the populated metric family
 yield metric
def _fetch_queue_size(self, provider):
 # Simulate API latency
 time.sleep(0.1) 
 return random.randint(0, 100)
if __name__ == '__main__':
 # Register the custom collector instance
 REGISTRY.register(PaymentGatewayCollector(endpoint="api.payments.internal"))
# Start HTTP server to serve /metrics
 start_http_server(8000)
 print("Prometheus exporter running on port 8000...")
# Keep main thread alive
 while True:
 time.sleep(1)

How It Works

Each time Prometheus issues an HTTP GET request to http://localhost:8000/metrics, the custom collector's collect() method executes automatically. The GaugeMetricFamily object constructs the metric response payload on the fly. Because telemetry data is gathered at scrape time, stale gauge values are completely avoided if an underlying metric target stops returning data or is removed, the collector simply omits that metric series, ensuring Grafana dashboards and alerting rules reflect true setup state.