Comparison: Prometheus 3.0 vs. InfluxDB 3.0 for Local Metric Collection

// prometheus-custom-exporter.go
// Exports custom application metrics to Prometheus 3.0, compliant with Prometheus exposition format.
// Requires github.com/prometheus/client_golang v1.19.0 or later.
// Run: go run prometheus-custom-exporter.go, then curl http://localhost:9091/metrics
package main

import (
    "log"
    "math/rand"
    "net/http"
    "os"
    "time"

    "github.com/prometheus/client_golang/prometheus"
    "github.com/prometheus/client_golang/prometheus/promhttp"
)

// Define custom metrics
var (
    // httpRequestsTotal counts total HTTP requests processed by the application
    httpRequestsTotal = prometheus.NewCounterVec(
        prometheus.CounterOpts{
            Name: "app_http_requests_total",
            Help: "Total number of HTTP requests processed by the application",
        },
        []string{"method", "status_code"},
    )

    // activeRequestsGauge tracks current in-flight requests
    activeRequestsGauge = prometheus.NewGauge(
        prometheus.GaugeOpts{
            Name: "app_active_requests",
            Help: "Current number of in-flight HTTP requests",
        },
    )

    // requestDurationHistogram measures request latency in milliseconds
    requestDurationHistogram = prometheus.NewHistogramVec(
        prometheus.HistogramOpts{
            Name:    "app_request_duration_ms",
            Help:    "Request latency in milliseconds",
            Buckets: prometheus.DefBuckets,
        },
        []string{"method"},
    )
)

func init() {
    // Register metrics with Prometheus's default registerer
    err := prometheus.Register(httpRequestsTotal)
    if err != nil {
        log.Printf("Failed to register httpRequestsTotal: %v", err)
    }
    err = prometheus.Register(activeRequestsGauge)
    if err != nil {
        log.Printf("Failed to register activeRequestsGauge: %v", err)
    }
    err = prometheus.Register(requestDurationHistogram)
    if err != nil {
        log.Printf("Failed to register requestDurationHistogram: %v", err)
    }
}

// simulateRequest simulates an HTTP request with random latency and status code
func simulateRequest(method string) {
    start := time.Now()
    activeRequestsGauge.Inc()
    defer activeRequestsGauge.Dec()

    // Simulate random latency between 50ms and 500ms
    latency := time.Duration(rand.Intn(450)+50) * time.Millisecond
    time.Sleep(latency)

    // Simulate random status code: 200 (70%), 400 (20%), 500 (10%)
    statusCode := "200"
    roll := rand.Float32()
    if roll > 0.7 && roll <= 0.9 {
        statusCode = "400"
    } else if roll > 0.9 {
        statusCode = "500"
    }

    // Record metrics
    httpRequestsTotal.WithLabelValues(method, statusCode).Inc()
    requestDurationHistogram.WithLabelValues(method).Observe(float64(latency.Milliseconds()))

    log.Printf("Simulated %s request: status=%s, latency=%dms", method, statusCode, latency.Milliseconds())
}

func main() {
    // Seed random number generator
    rand.Seed(time.Now().UnixNano())

    // Start metric simulation in background goroutine
    go func() {
        for {
            simulateRequest("GET")
            time.Sleep(time.Duration(rand.Intn(100)) * time.Millisecond)
        }
    }()

    go func() {
        for {
            simulateRequest("POST")
            time.Sleep(time.Duration(rand.Intn(150)) * time.Millisecond)
        }
    }()

    // Expose metrics endpoint on port 9091
    http.Handle("/metrics", promhttp.Handler())
    log.Println("Prometheus exporter listening on :9091")
    err := http.ListenAndServe(":9091", nil)
    if err != nil {
        log.Printf("Failed to start HTTP server: %v", err)
        os.Exit(1)
    }
}

# influxdb-custom-writer.py
# Writes custom application metrics to InfluxDB 3.0 using the InfluxDB v3 Python client.
# Requires influxdb-client-python v1.39.0 or later, Python 3.10+.
# Run: python3 influxdb-custom-writer.py
# InfluxDB 3.0 config: assumes local instance running on http://localhost:8086, bucket "local-metrics", org "local"

import time
import random
import logging
from influxdb_client import InfluxDBClient, Point
from influxdb_client.client.write_api import SYNCHRONOUS

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

# InfluxDB 3.0 connection config (update with your local instance details)
INFLUX_URL = "http://localhost:8086"
INFLUX_TOKEN = "local-admin-token-12345"  # Replace with your InfluxDB token
INFLUX_ORG = "local"
INFLUX_BUCKET = "local-metrics"

def create_influx_client():
    """Initialize and return an InfluxDB 3.0 client with error handling."""
    try:
        client = InfluxDBClient(
            url=INFLUX_URL,
            token=INFLUX_TOKEN,
            org=INFLUX_ORG,
            timeout=10_000  # 10 second timeout
        )
        # Verify connection by checking health status
        health = client.health()
        if health.status != "pass":
            logger.error(f"InfluxDB health check failed: {health.message}")
            return None
        logger.info(f"Connected to InfluxDB 3.0 at {INFLUX_URL}, status: {health.status}")
        return client
    except Exception as e:
        logger.error(f"Failed to initialize InfluxDB client: {str(e)}")
        return None

def generate_metric_points():
    """Generate a batch of metric points in InfluxDB line protocol format."""
    points = []
    current_time = time.time_ns()

    # Simulate HTTP request metrics
    for method in ["GET", "POST", "PUT"]:
        # Simulate request count
        request_count = random.randint(1, 10)
        points.append(
            Point("http_requests")
            .tag("method", method)
            .tag("status_code", random.choice(["200", "400", "500"]))
            .field("count", request_count)
            .time(current_time)
        )

        # Simulate request latency
        latency_ms = random.randint(50, 500)
        points.append(
            Point("request_latency")
            .tag("method", method)
            .field("p50", latency_ms * 0.8)
            .field("p99", latency_ms * 1.5)
            .time(current_time)
        )

    # Simulate system metrics
    points.append(
        Point("system")
        .tag("host", "local-dev-node")
        .field("cpu_usage_percent", random.uniform(5.0, 30.0))
        .field("memory_usage_percent", random.uniform(20.0, 60.0))
        .time(current_time)
    )

    return points

def write_metrics_to_influx(client, points, retry_count=3):
    """Write metric points to InfluxDB with retry logic."""
    for attempt in range(retry_count):
        try:
            write_api = client.write_api(write_options=SYNCHRONOUS)
            write_api.write(bucket=INFLUX_BUCKET, record=points)
            logger.info(f"Successfully wrote {len(points)} points to InfluxDB bucket {INFLUX_BUCKET}")
            return True
        except Exception as e:
            logger.warning(f"Attempt {attempt + 1} failed to write metrics: {str(e)}")
            if attempt < retry_count - 1:
                time.sleep(2 ** attempt)  # Exponential backoff
    logger.error(f"Failed to write metrics after {retry_count} attempts")
    return False

def main():
    # Initialize InfluxDB client
    client = create_influx_client()
    if not client:
        logger.error("Exiting due to InfluxDB client initialization failure")
        return

    # Main metric generation loop
    try:
        while True:
            points = generate_metric_points()
            write_metrics_to_influx(client, points)
            time.sleep(1)  # Generate metrics every 1 second
    except KeyboardInterrupt:
        logger.info("Received shutdown signal, closing InfluxDB client")
        client.close()
    except Exception as e:
        logger.error(f"Unexpected error in main loop: {str(e)}")
        client.close()

if __name__ == "__main__":
    main()

# metric-throughput-benchmark.py
# Benchmarks ingest throughput for Prometheus 3.0 vs InfluxDB 3.0 using identical metric payloads.
# Requires prometheus-client v0.20.0, influxdb-client-python v1.39.0, Python 3.10+.
# Run: python3 metric-throughput-benchmark.py --target prometheus (or --target influxdb)
# Benchmark config: 100 concurrent workers, 1KB payload per metric, 5 minute duration.

import argparse
import time
import random
import threading
import logging
from prometheus_client import Gauge, push_to_gateway
from influxdb_client import InfluxDBClient, Point
from influxdb_client.client.write_api import SYNCHRONOUS

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

# Benchmark config
CONCURRENT_WORKERS = 100
METRIC_DURATION_SEC = 300  # 5 minutes
METRIC_PAYLOAD_SIZE = 1024  # 1KB per metric
PROMETHEUS_PUSHGATEWAY = "http://localhost:9091"
INFLUX_URL = "http://localhost:8086"
INFLUX_TOKEN = "local-admin-token-12345"
INFLUX_ORG = "local"
INFLUX_BUCKET = "local-metrics"

# Throughput counter
throughput_counter = 0
throughput_lock = threading.Lock()

def generate_metric_payload():
    """Generate a 1KB metric payload with random dimensional labels."""
    payload = {
        "metric_name": "benchmark_metric",
        "labels": {
            "host": f"bench-host-{random.randint(1, 100)}",
            "service": random.choice(["api", "worker", "scheduler"]),
            "region": random.choice(["us-east-1", "eu-west-1", "ap-southeast-1"]),
            "random_data": "".join(random.choices("abcdefghijklmnopqrstuvwxyz0123456789", k=METRIC_PAYLOAD_SIZE - 200))
        },
        "value": random.uniform(0.0, 100.0),
        "timestamp": time.time_ns()
    }
    return payload

def prometheus_worker(worker_id, duration_sec):
    """Worker thread that pushes metrics to Prometheus Pushgateway."""
    global throughput_counter
    end_time = time.time() + duration_sec
    metric = Gauge("benchmark_metric", "Benchmark metric for throughput testing", ["host", "service", "region"])

    while time.time() < end_time:
        payload = generate_metric_payload()
        try:
            metric.labels(
                host=payload["labels"]["host"],
                service=payload["labels"]["service"],
                region=payload["labels"]["region"]
            ).set(payload["value"])

            # Push to Prometheus Pushgateway
            push_to_gateway(PROMETHEUS_PUSHGATEWAY, job="benchmark", registry=metric._registry)

            with throughput_lock:
                throughput_counter += 1
        except Exception as e:
            logger.error(f"Prometheus worker {worker_id} error: {str(e)}")
        time.sleep(0.001)  # Small delay to prevent CPU saturation

def influxdb_worker(worker_id, duration_sec):
    """Worker thread that writes metrics to InfluxDB 3.0."""
    global throughput_counter
    end_time = time.time() + duration_sec

    try:
        client = InfluxDBClient(url=INFLUX_URL, token=INFLUX_TOKEN, org=INFLUX_ORG, timeout=1000)
        write_api = client.write_api(write_options=SYNCHRONOUS)
    except Exception as e:
        logger.error(f"InfluxDB worker {worker_id} failed to initialize client: {str(e)}")
        return

    while time.time() < end_time:
        payload = generate_metric_payload()
        try:
            point = Point("benchmark_metric") \
                .tag("host", payload["labels"]["host"]) \
                .tag("service", payload["labels"]["service"]) \
                .tag("region", payload["labels"]["region"]) \
                .field("value", payload["value"]) \
                .time(payload["timestamp"])

            write_api.write(bucket=INFLUX_BUCKET, record=point)

            with throughput_lock:
                throughput_counter += 1
        except Exception as e:
            logger.error(f"InfluxDB worker {worker_id} error: {str(e)}")
        time.sleep(0.001)

    client.close()

def run_benchmark(target):
    """Run the throughput benchmark for the specified target (prometheus/influxdb)."""
    global throughput_counter
    throughput_counter = 0

    logger.info(f"Starting {target} throughput benchmark: {CONCURRENT_WORKERS} workers, {METRIC_DURATION_SEC}s duration")
    start_time = time.time()

    # Start worker threads
    threads = []
    for i in range(CONCURRENT_WORKERS):
        if target == "prometheus":
            t = threading.Thread(target=prometheus_worker, args=(i, METRIC_DURATION_SEC))
        elif target == "influxdb":
            t = threading.Thread(target=influxdb_worker, args=(i, METRIC_DURATION_SEC))
        else:
            logger.error(f"Invalid target: {target}")
            return
        t.start()
        threads.append(t)

    # Wait for all threads to complete
    for t in threads:
        t.join()

    end_time = time.time()
    total_time = end_time - start_time
    throughput = throughput_counter / total_time

    logger.info(f"Benchmark complete for {target}")
    logger.info(f"Total metrics ingested: {throughput_counter}")
    logger.info(f"Total time: {total_time:.2f}s")
    logger.info(f"Throughput: {throughput:.2f} metrics/sec")

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Benchmark metric ingest throughput")
    parser.add_argument("--target", required=True, choices=["prometheus", "influxdb"], help="Target database: prometheus or influxdb")
    args = parser.parse_args()

    run_benchmark(args.target)

# prometheus.yml relabel config to drop high-cardinality labels
scrape_configs:
  - job_name: "go-app"
    static_configs:
      - targets: ["localhost:9091"]
    metric_relabel_configs:
      - source_labels: [user_id, request_id]
        action: drop_labels
      - regex: ".*"
        source_labels: [__name__]
        target_label: __name__
        replacement: "${1}_v3"  # Add suffix to avoid name collisions

-- InfluxDB 3.0 task to roll up http_requests into 1-minute aggregates
CREATE TASK rollup_http_requests
  EVERY 1m
  AS
    INSERT INTO http_requests_rollup
    SELECT
      method,
      status_code,
      count(sum_count) AS total_requests,
      avg(avg_latency) AS avg_latency_ms,
      time_bucket(1m, time) AS time
    FROM http_requests
    WHERE time >= now() - 1h
    GROUP BY method, status_code, time_bucket(1m, time)

# Prometheus 3.0 startup flag for 7-day retention
./prometheus --config.file=prometheus.yml --storage.tsdb.retention.time=7d

# InfluxDB 3.0 CLI command to create bucket with 7-day retention
influx bucket create --name local-metrics --retention 7d --org local