Write Prometheus Metrics Directly to QuestDB: A Zero-Overhead Ingestion Pipeline

# docker-compose.yml
services:
  questdb:
    image: questdb/questdb:latest
    ports:
      - "8812:8812"   # PostgreSQL query
      - "8880:8880"   # Web Console
      - "9092:9092"   # InfluxDB 1.x write compatibility
    volumes:
      - ./qdb:/root/.questdb

# prom_to_ilp.py — receive Prometheus remote_write, translate to ILP, write to QuestDB
# HTTP rather than gRPC so deployment only needs the stdlib + requests
import http.server, socket, gzip

class RemoteWriteHandler(http.server.BaseHTTPRequestHandler):
    def do_POST(self):
        # Prometheus remote_write bodies may be gzip-compressed
        if self.headers.get('Content-Encoding') == 'gzip':
            body = gzip.decompress(self.rfile.read(int(self.headers['Content-Length'])))
        else:
            body = self.rfile.read(int(self.headers['Content-Length']))
        ilp_lines = self.body_to_ilp(body)
        s = socket.socket()
        s.connect(('localhost', 9009))
        s.sendall('\n'.join(ilp_lines).encode())
        s.close()
        self.send_response(204)
        self.end_headers()

    def body_to_ilp(self, body):
        # Simplified: real protobuf needs per-TimeSeries decoding
        # Each TimeSeries → measurement + tags + field + timestamp
        # Assumes you have parse_ts(body) → [(name, tags, val, ts_ns)]
        from proto_parse import parse_ts   # your protobuf decode module
        lines = []
        for name, tags, val, ts_ns in parse_ts(body):
            tag_str = ','.join(f'{k}={v}' for k, v in tags.items())
            lines.append(f'{name},{tag_str} {val} {ts_ns}')
        return lines

http.server.HTTPServer(('0.0.0.0', 9093), RemoteWriteHandler).serve_forever()

# append to prometheus.yml
remote_write:
  - url: "http://edgebox:9093/write"
    queue_config:
      max_samples_per_send: 10000
      min_shards: 8
      max_shards: 64

SELECT ts, hostname, value
FROM node_cpu_seconds_total
WHERE ts >= now() - INTERVAL 1 HOUR
SAMPLE BY 1m
ORDER BY ts ASC;

# collector.py — sample business metrics every 15 s, write to QuestDB via ILP
import socket, time, requests

class MetricCollector:
    def __init__(self):
        self.sock = socket.socket()
        self.sock.connect(('localhost', 9009))

    def push(self, metric: str, tags: dict, value: float, ts_ns: int):
        tag_str = ','.join(f'{k}={v}' for k, v in tags.items())
        self.sock.sendall(f'{metric},{tag_str} {value} {ts_ns}\n'.encode())

    def run(self):
        c = MetricCollector()
        while True:
            ts_ns = int(time.time() * 1_000_000_000)
            # pull a value from your own service
            api_latency = requests.get('http://localhost:5000/health', timeout=5).elapsed.total_seconds()
            c.push('api_latency_ms', {'route': '/health'}, api_latency * 1000, ts_ns)
            c.push('order_count', {'region': 'cn'}, 42, ts_ns)
            time.sleep(15)

if __name__ == '__main__':
    MetricCollector().run()

SELECT ts, hostname, AVG(value) AS avg_val
FROM node_cpu_seconds_total
WHERE ts >= now() - INTERVAL 1 HOUR
SAMPLE BY 1m BY hostname
ORDER BY ts ASC;

# crontab -e
0 3 * * * python3 /opt/retention/drop_old_partitions.py

# drop_old_partitions.py
import subprocess
from datetime import date, timedelta
import psycopg2

cutoff = (date.today() - timedelta(days=90)).isoformat()
conn = psycopg2.connect(host='localhost', port=8812, user='admin', password='quest')
cur = conn.cursor()
cur.execute(
    "SELECT partition_path FROM partitions "
    "WHERE table_name = 'node_cpu_seconds_total' AND partition_path < %s",
    (cutoff,)
)
for (path,) in cur.fetchall():
    print(f"dropping {path}")
    subprocess.run(['questdb-admin', 'drop-partition', path], check=True)