Writing Custom Prometheus Exporters in Python and Go

Introduction

Prometheus exporters are programs that expose metrics in the Prometheus text format. While hundreds of exporters exist for common systems (node_exporter for Linux, mysqld_exporter for MySQL), you often need to monitor custom applications, proprietary systems, or business metrics. This guide teaches you to write production-quality custom exporters in Python and Go.

Understanding the Prometheus Text Format

Every exporter exposes metrics at an HTTP endpoint (typically /metrics) in a simple text format:

TEXT

# HELP http_requests_total Total number of HTTP requests
# TYPE http_requests_total counter
http_requests_total{method="GET",status="200"} 1234
http_requests_total{method="POST",status="200"} 567
http_requests_total{method="GET",status="404"} 23

# HELP request_duration_seconds HTTP request duration in seconds
# TYPE request_duration_seconds histogram
request_duration_seconds_bucket{le="0.005"} 100
request_duration_seconds_bucket{le="0.01"} 145
request_duration_seconds_bucket{le="0.025"} 210
request_duration_seconds_bucket{le="+Inf"} 250
request_duration_seconds_sum 12.345
request_duration_seconds_count 250

Metric Types

Type	Use Case	Example
Counter	Values that only increase	Requests served, errors
Gauge	Values that go up and down	Memory usage, queue size
Histogram	Distribution of values	Request duration, response size
Summary	Like histogram with quantiles	99th percentile latency

Writing a Python Exporter

PYTHON

#!/usr/bin/env python3
# custom_exporter.py - Monitor a custom application API
import time
import requests
from prometheus_client import start_http_server, Counter, Gauge, Histogram, CollectorRegistry, REGISTRY
from prometheus_client.core import GaugeMetricFamily, CounterMetricFamily

# Define metrics (module-level)
REQUEST_COUNT = Counter(
    'myapp_requests_total',
    'Total requests to myapp',
    ['endpoint', 'status_code']
)

ACTIVE_USERS = Gauge(
    'myapp_active_users',
    'Current number of active users'
)

RESPONSE_TIME = Histogram(
    'myapp_response_duration_seconds',
    'API response time in seconds',
    ['endpoint'],
    buckets=[0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0]
)

QUEUE_SIZE = Gauge(
    'myapp_queue_size',
    'Number of jobs in processing queue',
    ['queue_name']
)

def collect_metrics():
    '''Collect metrics from the application API.'''
    try:
        # Fetch stats from your application
        start = time.time()
        resp = requests.get('http://myapp:8080/internal/stats', timeout=5)
        resp.raise_for_status()
        duration = time.time() - start

        RESPONSE_TIME.labels(endpoint='/internal/stats').observe(duration)
        REQUEST_COUNT.labels(endpoint='/stats', status_code='200').inc()

        data = resp.json()
        ACTIVE_USERS.set(data['active_users'])

        for queue_name, size in data['queues'].items():
            QUEUE_SIZE.labels(queue_name=queue_name).set(size)

    except requests.RequestException as e:
        REQUEST_COUNT.labels(endpoint='/stats', status_code='error').inc()
        print(f"Failed to collect metrics: {e}")

if __name__ == '__main__':
    # Start Prometheus HTTP server on port 9100
    start_http_server(9100)
    print("Exporter running on :9100/metrics")

    while True:
        collect_metrics()
        time.sleep(15)  # Collect every 15 seconds

Custom Collector Class (More Control)

PYTHON

#!/usr/bin/env python3
# database_exporter.py - Expose database query metrics
import psycopg2
from prometheus_client import start_http_server, REGISTRY
from prometheus_client.core import GaugeMetricFamily, CounterMetricFamily
import time

class DatabaseCollector:
    '''Custom collector for PostgreSQL metrics.'''

    def __init__(self, dsn):
        self.dsn = dsn

    def collect(self):
        '''Called by Prometheus client on each scrape.'''
        conn = None
        try:
            conn = psycopg2.connect(self.dsn)
            cur = conn.cursor()

            # Table sizes
            table_size = GaugeMetricFamily(
                'postgres_table_size_bytes',
                'Size of each table in bytes',
                labels=['database', 'schema', 'table']
            )
            cur.execute('''
                SELECT current_database(), schemaname, tablename,
                       pg_total_relation_size(schemaname||'.'||tablename)
                FROM pg_tables
                WHERE schemaname NOT IN ('pg_catalog', 'information_schema')
            ''')
            for row in cur.fetchall():
                table_size.add_metric([row[0], row[1], row[2]], row[3])
            yield table_size

            # Active connections
            connections = GaugeMetricFamily(
                'postgres_connections',
                'Number of database connections',
                labels=['state']
            )
            cur.execute('''
                SELECT state, count(*)
                FROM pg_stat_activity
                WHERE datname = current_database()
                GROUP BY state
            ''')
            for row in cur.fetchall():
                state = row[0] or 'null'
                connections.add_metric([state], row[1])
            yield connections

            # Slow queries (> 1 second)
            slow_queries = GaugeMetricFamily(
                'postgres_slow_queries',
                'Number of queries running longer than 1 second'
            )
            cur.execute('''
                SELECT count(*) FROM pg_stat_activity
                WHERE state = 'active'
                AND now() - query_start > interval '1 second'
            ''')
            slow_queries.add_metric([], cur.fetchone()[0])
            yield slow_queries

        except Exception as e:
            print(f"Collection failed: {e}")
        finally:
            if conn:
                conn.close()

if __name__ == '__main__':
    REGISTRY.register(DatabaseCollector(
        "postgresql://monitor:pass@localhost:5432/appdb"
    ))
    start_http_server(9187)
    print("PostgreSQL exporter on :9187/metrics")
    while True:
        time.sleep(60)

Dockerizing Your Exporter

DOCKERFILE

# Dockerfile
FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY custom_exporter.py .
EXPOSE 9100

USER nobody
CMD ["python", "custom_exporter.py"]

YAML

# docker-compose.yml addition
services:
  custom-exporter:
    build: ./exporters/myapp
    restart: unless-stopped
    ports:
      - "9100:9100"
    environment:
      - APP_URL=http://myapp:8080
    networks:
      - monitoring

Prometheus Scrape Configuration

YAML

# prometheus.yml
scrape_configs:
  - job_name: 'custom-myapp'
    scrape_interval: 15s
    scrape_timeout: 10s
    static_configs:
      - targets: ['custom-exporter:9100']
        labels:
          environment: production
          app: myapp

  # Multiple instances with relabeling
  - job_name: 'myapp-instances'
    static_configs:
      - targets:
        - 'app-01:9100'
        - 'app-02:9100'
        - 'app-03:9100'
    relabel_configs:
      - source_labels: [__address__]
        regex: '(.*):.*'
        target_label: instance
        replacement: '$1'

Business Metrics Exporter

PYTHON

# business_metrics.py - Monitor business KPIs
class BusinessMetricsCollector:
    def __init__(self, db_dsn):
        self.db_dsn = db_dsn

    def collect(self):
        # Orders today
        orders_today = GaugeMetricFamily(
            'business_orders_today_total',
            'Number of orders placed today'
        )
        # Revenue today
        revenue = GaugeMetricFamily(
            'business_revenue_today_usd',
            'Revenue generated today in USD'
        )
        # Active subscriptions
        subscriptions = GaugeMetricFamily(
            'business_active_subscriptions',
            'Number of active subscriptions',
            labels=['plan']
        )

        conn = psycopg2.connect(self.db_dsn)
        cur = conn.cursor()

        cur.execute("SELECT count(*) FROM orders WHERE created_at::date = CURRENT_DATE")
        orders_today.add_metric([], cur.fetchone()[0])
        yield orders_today

        cur.execute("SELECT COALESCE(sum(amount), 0) FROM orders WHERE created_at::date = CURRENT_DATE")
        revenue.add_metric([], float(cur.fetchone()[0]))
        yield revenue

        cur.execute("SELECT plan, count(*) FROM subscriptions WHERE status='active' GROUP BY plan")
        for plan, count in cur.fetchall():
            subscriptions.add_metric([plan], count)
        yield subscriptions

        conn.close()

Testing Your Exporter

BASH

# Start exporter
python custom_exporter.py &

# Check metrics are exposed
curl -s http://localhost:9100/metrics | head -20

# Verify specific metric
curl -s http://localhost:9100/metrics | grep myapp_active_users

# Test with promtool
promtool check metrics http://localhost:9100/metrics

Writing custom exporters turns any system that has an API or database into a first-class Prometheus citizen. The key principle: collect metrics close to the source, use appropriate metric types, and add labels that enable useful aggregations and filtering.