Database Monitoring & Observability

-- 1. Add to postgresql.conf
shared_preload_libraries = 'pg_stat_statements, auto_explain'

-- 2. Restart PostgreSQL
sudo systemctl restart postgresql

-- 3. Create extension in your database
CREATE EXTENSION pg_stat_statements;

-- Result: PostgreSQL now tracks query statistics (execution time, call count, etc.)

-- Top 10 slowest queries by total execution time
SELECT
    query,
    calls,
    total_exec_time,
    mean_exec_time,
    max_exec_time,
    stddev_exec_time
FROM pg_stat_statements
ORDER BY total_exec_time DESC
LIMIT 10;

-- Example output:
-- query                                    | calls | total_exec_time | mean_exec_time | max_exec_time
-- -----------------------------------------|-------|-----------------|----------------|---------------
-- SELECT * FROM orders WHERE user_id = $1 | 50000 | 125000ms   | 2.5ms     | 500ms
-- UPDATE products SET stock = stock - $1  | 25000 | 75000ms    | 3.0ms     | 1200ms
-- SELECT COUNT(*) FROM users              | 1000  | 50000ms    | 50ms      | 150ms

import psycopg2
import time

def monitor_slow_queries(threshold_ms=100):
    """Continuously monitor and alert on slow queries."""
    conn = psycopg2.connect("dbname=myapp user=postgres")
    cur = conn.cursor()

    while True:
        cur.execute("""
            SELECT
                pid,
                now() - query_start AS duration,
                state,
                query
            FROM pg_stat_activity
            WHERE state != 'idle'
              AND (now() - query_start) > interval '%s milliseconds'
            ORDER BY duration DESC
        """ % threshold_ms)

        slow_queries = cur.fetchall()

        if slow_queries:
            print(f"⚠️  Found {len(slow_queries)} slow queries:")
            for pid, duration, state, query in slow_queries:
                print(f"  PID {pid}: {duration} - {query[:100]}...")
        else:
            print("✓ All queries under threshold")

        time.sleep(5)  # Check every 5 seconds

-- Edit postgresql.conf

# Log queries slower than 100ms
log_min_duration_statement = 100

# Include query execution plan for slow queries
auto_explain.log_min_duration = 100
auto_explain.log_analyze = true

# Log destination
log_destination = 'csvlog'
logging_collector = on
log_directory = 'log'
log_filename = 'postgresql-%Y-%m-%d_%H%M%S.log'

# Restart PostgreSQL
sudo systemctl restart postgresql

# Result: Slow queries written to log files in CSV format

import csv
import re
from collections import defaultdict

def analyze_slow_query_log(log_file):
    """Parse and analyze PostgreSQL slow query log."""
    query_stats = defaultdict(lambda: {
        'count': 0,
        'total_time': 0,
        'max_time': 0
    })

    with open(log_file, 'r') as f:
        reader = csv.DictReader(f)
        for row in reader:
            if row.get('message', '').startswith('duration:'):
                # Extract duration
                duration_match = re.search(r'duration: ([d.]+) ms', row['message'])
                if not duration_match:
                    continue

                duration = float(duration_match.group(1))

                # Extract query (normalize parameters)
                query_match = re.search(r'statement: (.+)', row['message'])
                if not query_match:
                    continue

                query = query_match.group(1)
                # Normalize: replace numbers/strings with placeholders
                normalized = re.sub(r"'[^']*'", "'?'", query)
                normalized = re.sub(r'd+', '?', normalized)

                # Update statistics
                stats = query_stats[normalized]
                stats['count'] += 1
                stats['total_time'] += duration
                stats['max_time'] = max(stats['max_time'], duration)

    # Sort by total time
    sorted_queries = sorted(
        query_stats.items(),
        key=lambda x: x[1]['total_time'],
        reverse=True
    )

    # Print report
    print("Top 10 Slow Query Patterns:")
    for query, stats in sorted_queries[:10]:
        avg_time = stats['total_time'] / stats['count']
        print(f"
Query: {query[:100]}...")
        print(f"  Count: {stats['count']}")
        print(f"  Total: {stats['total_time']:.1f}ms")
        print(f"  Avg: {avg_time:.1f}ms")
        print(f"  Max: {stats['max_time']:.1f}ms")

# 1. Install postgres_exporter (exposes PostgreSQL metrics to Prometheus)
docker run -d \
  --name postgres-exporter \
  -p 9187:9187 \
  -e DATA_SOURCE_NAME="postgresql://postgres:password@localhost:5432/postgres?sslmode=disable" \
  prometheuscommunity/postgres-exporter

# 2. Configure Prometheus to scrape metrics (prometheus.yml)
scrape_configs:
  - job_name: 'postgresql'
    static_configs:
      - targets: ['localhost:9187']

# 3. Start Prometheus
docker run -d \
  --name prometheus \
  -p 9090:9090 \
  -v /path/to/prometheus.yml:/etc/prometheus/prometheus.yml \
  prom/prometheus

# 4. Start Grafana
docker run -d \
  --name grafana \
  -p 3000:3000 \
  grafana/grafana

# 5. Open Grafana: http://localhost:3000 (admin/admin)
# 6. Add Prometheus data source: http://prometheus:9090
# 7. Import PostgreSQL dashboard (ID: 9628)

# Result: Full-featured monitoring dashboard with:
# - QPS, connection count, cache hit rate
# - Query duration percentiles (P50, P95, P99)
# - Replication lag, deadlocks, locks
# - CPU, memory, disk I/O
# - 30-day historical data retention

# alert_rules.yml - Define alerting thresholds

groups:
  - name: postgresql_alerts
    interval: 30s
    rules:
      # High connection usage
      - alert: PostgreSQLConnectionsHigh
        expr: pg_stat_database_numbackends / pg_settings_max_connections > 0.8
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "PostgreSQL connections high ({{ $value | humanizePercentage }})"
          description: "Database {{ $labels.datname }} using >80% of max connections"

      # Low cache hit ratio
      - alert: PostgreSQLCacheHitRateLow
        expr: rate(pg_stat_database_blks_hit[5m]) /
              (rate(pg_stat_database_blks_hit[5m]) + rate(pg_stat_database_blks_read[5m])) < 0.95
        for: 10m
        labels:
          severity: warning
        annotations:
          summary: "Cache hit ratio low ({{ $value | humanizePercentage }})"
          description: "Database {{ $labels.datname }} cache hit rate <95%"

      # Replication lag
      - alert: PostgreSQLReplicationLag
        expr: pg_replication_lag > 30
        for: 2m
        labels:
          severity: critical
        annotations:
          summary: "Replication lag high ({{ $value }}s)"
          description: "Replica {{ $labels.instance }} is {{ $value }}s behind master"

      # Deadlocks detected
      - alert: PostgreSQLDeadlocks
        expr: rate(pg_stat_database_deadlocks[5m]) > 0.1
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "Deadlocks detected ({{ $value }}/sec)"
          description: "Database {{ $labels.datname }} experiencing deadlocks"

# Load rules in Prometheus
prometheus --config.file=prometheus.yml --rules.file=alert_rules.yml

import psycopg2
import requests
import time

SLACK_WEBHOOK = "https://hooks.slack.com/services/YOUR/WEBHOOK/URL"

def check_database_health():
    """Monitor database and send alerts."""
    conn = psycopg2.connect("dbname=myapp user=postgres")
    cur = conn.cursor()

    # Check 1: Connection usage
    cur.execute("""
        SELECT count(*) AS active,
               current_setting('max_connections')::int AS max_conn
        FROM pg_stat_activity
        WHERE state != 'idle'
    """)
    active, max_conn = cur.fetchone()
    usage_pct = (active / max_conn) * 100

    if usage_pct > 80:
        send_alert(
            f"⚠️  High connection usage: {active}/{max_conn} ({usage_pct:.1f}%)",
            severity="warning"
        )

    # Check 2: Cache hit ratio
    cur.execute("""
        SELECT
            sum(blks_hit) / (sum(blks_hit) + sum(blks_read)) AS cache_hit_ratio
        FROM pg_stat_database
        WHERE datname = current_database()
    """)
    cache_hit_ratio = cur.fetchone()[0]

    if cache_hit_ratio < 0.95:
        send_alert(
            f"⚠️  Low cache hit ratio: {cache_hit_ratio:.2%}",
            severity="warning"
        )

    # Check 3: Long-running queries
    cur.execute("""
        SELECT count(*)
        FROM pg_stat_activity
        WHERE state = 'active'
          AND (now() - query_start) > interval '60 seconds'
    """)
    long_queries = cur.fetchone()[0]

    if long_queries > 5:
        send_alert(
            f"🚨 {long_queries} queries running >60 seconds",
            severity="critical"
        )

def send_alert(message, severity="warning"):
    """Send alert to Slack."""
    color = {"info": "#36a64f", "warning": "#ff9900", "critical": "#ff0000"}
    requests.post(SLACK_WEBHOOK, json={
        "attachments": [{
            "color": color[severity],
            "text": message
        }]
    })

# Run every 60 seconds
while True:
    check_database_health()
    time.sleep(60)