Database Monitoring & Observability

# pg_stat_statements and auto_explain are shared_preload_libraries: they can
# only be loaded when the server starts, so they are passed on the command
# line here rather than enabled later. logging_collector is also restart-only,
# so it goes in the same place. Everything else in this lesson is changed on a
# running server without a restart.
docker network create monnet

docker run -d --name pgmon-demo --network monnet \
  -e POSTGRES_PASSWORD=demo -e POSTGRES_USER=demo -e POSTGRES_DB=demo \
  -p 5432:5432 postgres:16-alpine \
  -c shared_preload_libraries=pg_stat_statements,auto_explain \
  -c pg_stat_statements.track=all \
  -c logging_collector=on

# Wait for the server rather than guessing with sleep.
until docker exec pgmon-demo pg_isready -U demo -q; do sleep 1; done

# Preloading the library only makes the code available. The view itself is
# created per database, like any other extension.
docker exec pgmon-demo psql -U demo -d demo -c "CREATE EXTENSION pg_stat_statements;"
docker exec pgmon-demo psql -U demo -d demo -c "SELECT version();" -t

# A virtualenv for the Python examples further down.
python3 -m venv .venv && . .venv/bin/activate
pip install "psycopg[binary]>=3.2"

# Two tables and a workload worth measuring. Note there is deliberately NO
# index on orders.user_id: the sequential scan that causes is the slow query
# the rest of the lesson hunts down.
docker exec -i pgmon-demo psql -U demo -d demo -q \
  --pset linestyle=unicode --pset border=2 <<'SQL'
CREATE TABLE users (
    id         serial PRIMARY KEY,
    username   text NOT NULL,
    email      text NOT NULL,
    created_at timestamptz NOT NULL DEFAULT now()
);

CREATE TABLE orders (
    id         serial PRIMARY KEY,
    user_id    integer NOT NULL REFERENCES users(id),
    total      numeric(10,2) NOT NULL,
    status     text NOT NULL,
    created_at timestamptz NOT NULL DEFAULT now()
);

INSERT INTO users (username, email)
SELECT 'user' || g, 'user' || g || '@example.com'
FROM generate_series(1, 20000) AS g;

INSERT INTO orders (user_id, total, status)
SELECT (random() * 19999)::int + 1,
       (random() * 500)::numeric(10,2),
       (ARRAY['pending','paid','shipped'])[(random() * 2)::int + 1]
FROM generate_series(1, 200000);

ANALYZE users;
ANALYZE orders;

SELECT count(*) AS users FROM users;
SELECT count(*) AS orders FROM orders;
SQL

"""workload.py - traffic worth measuring, so the statistics views are not empty."""
import random

import psycopg

conn = psycopg.connect("host=localhost port=5432 dbname=demo user=demo password=demo")

# Start from a clean slate so the numbers below describe this run only.
conn.execute("SELECT pg_stat_statements_reset()")
conn.commit()

random.seed(42)

with conn.cursor() as cur:
    # Fast: primary-key lookups.
    for _ in range(2000):
        cur.execute("SELECT id, username, email FROM users WHERE id = %s",
                    (random.randint(1, 20000),))
        cur.fetchone()

    # Slow: orders.user_id has no index, so each of these scans the table.
    for _ in range(200):
        cur.execute("SELECT id, total, status FROM orders WHERE user_id = %s",
                    (random.randint(1, 20000),))
        cur.fetchall()

    # Moderate: an aggregate over all 200,000 rows.
    for _ in range(50):
        cur.execute("SELECT count(*) FROM orders WHERE status = %s", ("paid",))
        cur.fetchone()

    # Writes.
    for _ in range(100):
        cur.execute("UPDATE orders SET status = %s WHERE id = %s",
                    ("shipped", random.randint(1, 200000)))

conn.commit()
conn.close()
print("workload done")

# pg_stat_statements aggregates every execution of every query, with literals
# replaced by $1, $2, ... so that a million lookups collapse into one row.
# Rank by total_exec_time, not mean: a 2ms query called a million times costs
# far more than a 500ms report run twice a day.
docker exec -i pgmon-demo psql -U demo -d demo -q \
  --pset linestyle=unicode --pset border=2 <<'SQL'
SELECT
    left(query, 44)                    AS query,
    calls,
    round(total_exec_time::numeric, 1) AS total_ms,
    round(mean_exec_time::numeric, 3)  AS mean_ms,
    round(max_exec_time::numeric, 3)   AS max_ms,
    rows
FROM pg_stat_statements
WHERE query NOT LIKE '%pg_stat_statements%'   -- hide the monitoring query itself
ORDER BY total_exec_time DESC
LIMIT 6;
SQL

# Look again at those two UPDATE rows. Same text, different queryid, and the
# calls split 84/16 instead of showing 100.
docker exec -i pgmon-demo psql -U demo -d demo -q \
  --pset linestyle=unicode --pset border=2 <<'SQL'
SELECT queryid, calls, query
FROM pg_stat_statements
WHERE query LIKE 'UPDATE orders SET status%'
ORDER BY calls DESC;
SQL

"""param_types.py - why one SQL text can occupy two pg_stat_statements rows."""
import psycopg

conn = psycopg.connect("host=localhost port=5432 dbname=demo user=demo password=demo")

# psycopg sends the smallest integer type that fits the Python value, and the
# parameter type is part of the parse tree that pg_stat_statements hashes into
# queryid. The workload drew ids from 1..200000, so roughly 32767/200000 = 16%
# of them fit in a smallint. That is the 16 calls.
for value in (1, 32767, 32768, 200000):
    pg_type = conn.execute("SELECT pg_typeof(%s)", (value,)).fetchone()[0]
    print(f"python int {value:>6}  ->  parameter type {pg_type}")

conn.close()

"""monitor_activity.py - what is stuck right now.

pg_stat_statements is history. pg_stat_activity is a live view of backends,
which is the view you want during an incident.
"""
import time

import psycopg

# The threshold is a bound parameter, not string-formatted into the SQL.
LONG_RUNNING = """
    SELECT pid,
           now() - query_start AS running_for,
           state,
           wait_event_type,
           left(query, 60) AS query
    FROM pg_stat_activity
    WHERE state = 'active'                 -- executing, not idle
      AND backend_type = 'client backend'  -- skip autovacuum, walwriter, etc
      AND pid <> pg_backend_pid()          -- skip this monitoring query
      AND query_start IS NOT NULL
      AND now() - query_start > make_interval(secs => %s)
    ORDER BY running_for DESC
"""


def check_once(conn, threshold_seconds):
    """Print any backend running longer than threshold_seconds."""
    rows = conn.execute(LONG_RUNNING, (threshold_seconds,)).fetchall()
    if not rows:
        print(f"[ok] no query running longer than {threshold_seconds}s")
        return 0
    print(f"[warn] {len(rows)} query(s) over {threshold_seconds}s:")
    for pid, running_for, state, wait_event_type, query in rows:
        secs = running_for.total_seconds()
        waiting = wait_event_type or "-"
        print(f"   pid {pid}  {secs:5.1f}s  {state}  wait={waiting}  {query}")
    return len(rows)


def monitor(threshold_seconds=1.0, iterations=4, interval=2):
    # One connection for the whole run, closed on the way out. Opening a fresh
    # connection per check leaks a backend every cycle, which is how a
    # monitoring script ends up causing the outage it was meant to catch.
    with psycopg.connect(
        "host=localhost port=5432 dbname=demo user=demo password=demo"
    ) as conn:
        conn.autocommit = True  # never hold a transaction open while polling
        for _ in range(iterations):
            check_once(conn, threshold_seconds)
            time.sleep(interval)


if __name__ == "__main__":
    monitor()

# None of these need a restart: ALTER SYSTEM writes postgresql.auto.conf and
# pg_reload_conf() makes the server re-read it. (logging_collector DOES need a
# restart, which is why it was set at startup in the setup block above.)
docker exec -i pgmon-demo psql -U demo -d demo -q \
  --pset linestyle=unicode --pset border=2 <<'SQL'
ALTER SYSTEM SET log_destination = 'csvlog';
ALTER SYSTEM SET log_directory = 'log';
ALTER SYSTEM SET log_min_duration_statement = '100ms';   -- log slow statements
ALTER SYSTEM SET auto_explain.log_min_duration = '100ms'; -- and their plans
ALTER SYSTEM SET auto_explain.log_analyze = on;           -- with real row counts
SELECT pg_reload_conf();
SQL

# Check from a SEPARATE session. pg_reload_conf() signals the server and
# returns immediately; the session that called it is still running with the
# old values, so verifying inside the block above reports the defaults and
# makes it look like nothing happened.
docker exec pgmon-demo psql -U demo -d demo -q \
  --pset linestyle=unicode --pset border=2 -c "
SELECT name, setting, unit FROM pg_settings
WHERE name IN ('log_destination', 'log_min_duration_statement',
               'auto_explain.log_min_duration', 'auto_explain.log_analyze',
               'logging_collector')
ORDER BY name;"

# An accidental fan-out join: orders joined to itself on a non-unique key,
# which multiplies rows instead of filtering them. This is a realistic
# production mistake and it comfortably exceeds the 100ms threshold.
docker exec -i pgmon-demo psql -U demo -d demo -q <<'SQL'
\o /dev/null
SELECT a.status, count(*) FROM orders a JOIN orders b ON a.user_id = b.user_id
GROUP BY a.status;
SELECT count(*) FROM orders a JOIN orders b ON a.user_id = b.user_id
AND a.id < b.id;
SQL

# Show the newest log file and one complete record from it.
LOGF=$(docker exec pgmon-demo sh -c 'ls -t /var/lib/postgresql/data/log/*.csv | head -1')
docker exec pgmon-demo sh -c "head -c 900 $LOGF"

"""double_log.py - count the records the two settings actually produced."""
import csv

CSVLOG = "postgresql.csv"   # docker exec pgmon-demo cat <logfile> > postgresql.csv

for row in csv.reader(open(CSVLOG, newline="")):
    message = row[13]
    if not message.startswith("duration:"):
        continue
    first_line = message.split("\n")[0]
    source = "auto_explain" if "plan:" in first_line else "statement log"
    print(f"pid={row[3]:>4} line={row[6]:>2}  {source:<14} {first_line[:46]}")

"""analyze_log.py - aggregate a PostgreSQL csvlog by query, ranked by total time.

Two things make this harder than it looks:
  1. A csvlog file has NO header row, so csv.DictReader must be given
     fieldnames explicitly or it silently eats the first log record.
  2. auto_explain writes the plan into the message field, newlines and all,
     so one record spans many physical lines. Only a real CSV parser gets
     the record boundaries right: readline() and split(",") do not.
"""
import csv
import re
import sys
from collections import defaultdict

# csvlog columns for PostgreSQL 14+ (26 fields), confirmed against the running
# server rather than assumed. query_id is the last one.
CSVLOG_FIELDS = [
    "log_time", "user_name", "database_name", "process_id", "connection_from",
    "session_id", "session_line_num", "command_tag", "session_start_time",
    "virtual_transaction_id", "transaction_id", "error_severity",
    "sql_state_code", "message", "detail", "hint", "internal_query",
    "internal_query_pos", "context", "query", "query_pos", "location",
    "application_name", "backend_type", "leader_pid", "query_id",
]

DURATION_RE = re.compile(r"duration: ([\d.]+) ms")
# log_min_duration_statement writes "duration: N ms  statement: <sql>", or
# "execute <name>: <sql>" for extended-protocol queries.
STATEMENT_RE = re.compile(r"(?:statement: |execute [^:]*: )(.+)", re.S)


def analyze(log_file):
    stats = defaultdict(lambda: {"count": 0, "total": 0.0, "max": 0.0, "sample": ""})

    with open(log_file, newline="", encoding="utf-8") as f:
        for row in csv.DictReader(f, fieldnames=CSVLOG_FIELDS):
            message = row["message"] or ""
            match = DURATION_RE.search(message)
            if not match:
                continue

            # Count each execution ONCE. With log_min_duration_statement and
            # auto_explain.log_min_duration both armed, a slow query produces
            # two records. Keeping only the statement records avoids counting
            # every call twice.
            statement = STATEMENT_RE.search(message)
            if not statement:
                continue

            duration = float(match.group(1))
            sample = " ".join(statement.group(1).split())

            # Group by query_id when the server computed one: it is the same
            # identifier pg_stat_statements uses, so no fragile regex
            # normalisation of literals is needed.
            key = row["query_id"] or sample

            entry = stats[key]
            entry["count"] += 1
            entry["total"] += duration
            entry["max"] = max(entry["max"], duration)
            entry["sample"] = entry["sample"] or sample

    ranked = sorted(stats.values(), key=lambda e: e["total"], reverse=True)

    print(f"{'calls':>5} {'total_ms':>9} {'avg_ms':>8} {'max_ms':>8}  query")
    for entry in ranked[:10]:
        avg = entry["total"] / entry["count"]
        print(f"{entry['count']:>5} {entry['total']:>9.1f} {avg:>8.1f} "
              f"{entry['max']:>8.1f}  {entry['sample'][:44]}")


if __name__ == "__main__":
    analyze(sys.argv[1])

# All four containers must share one user-defined network. Docker's default
# bridge network does NOT resolve container names, so every "connect by name"
# step below would fail with "no such host" without this.
docker network create monnet 2>/dev/null || true   # already made in setup

# 1. postgres_exporter turns PostgreSQL statistics into Prometheus metrics.
#    It reaches the database by container name, never localhost: inside a
#    container, localhost is that container.
docker run -d --name pgexporter --network monnet -p 9187:9187 \
  -e DATA_SOURCE_NAME="postgresql://demo:demo@pgmon-demo:5432/demo?sslmode=disable" \
  quay.io/prometheuscommunity/postgres-exporter

# 2. Prometheus scrapes the exporter. Same rule: target the container name.
#    rule_files is how alert rules are loaded (there is no --rules.file flag).
cat > prometheus.yml <<'YAML'
global:
  scrape_interval: 15s

rule_files:
  - /etc/prometheus/alert_rules.yml

scrape_configs:
  - job_name: 'postgresql'
    static_configs:
      - targets: ['pgexporter:9187']
YAML

# rule_files names a file that must EXIST, even if it has no rules yet. Skip
# this and Docker bind-mounts a directory over the missing path, and
# Prometheus exits at once with "alert_rules.yml: is a directory". The
# alerting section later replaces this placeholder with real rules.
cat > alert_rules.yml <<'YAML'
groups: []
YAML

docker run -d --name prom-demo --network monnet -p 9090:9090 \
  -v $PWD/prometheus.yml:/etc/prometheus/prometheus.yml \
  -v $PWD/alert_rules.yml:/etc/prometheus/alert_rules.yml \
  prom/prometheus

# 3. Grafana. Port 3001 on the host because a dev server usually owns 3000.
docker run -d --name grafana-demo --network monnet -p 3001:3000 grafana/grafana

# 4. Confirm the chain works before opening any UI. A just-started Prometheus
#    has no targets until its first scrape, so wait for one instead of racing
#    it: querying immediately returns an empty list, or no JSON at all if the
#    server is not listening yet.
until curl -s localhost:9090/api/v1/targets | grep -q '"health":"up"'; do sleep 2; done

curl -s localhost:9090/api/v1/targets \
  | python3 -c "import json,sys; [print(' ', t['scrapeUrl'], '->', t['health']) \
      for t in json.load(sys.stdin)['data']['activeTargets']]"

# Grafana on http://localhost:3001 (admin/admin). Add Prometheus as a data
# source with URL http://prom-demo:9090, then import dashboard ID 9628
# ("PostgreSQL Database").

# Grafana runs in a container but you log into it through the browser, at
# http://localhost:3001 (port 3001 on the host maps to 3000 in the container).
# The first login is admin / admin, and Grafana then asks you to choose a new
# password. That prompt is where this usually goes wrong: pick a password,
# forget it, and admin/admin no longer works.

# Locked out? Reset it from inside the container. Two things to know: the
# command is "grafana cli", two words, because the old "grafana-cli" binary
# is gone from the current image and gives "executable file not found in
# $PATH"; and it prints Grafana's entire startup log to stdout before the
# one line you care about, hence the grep.
docker exec grafana-demo grafana cli admin reset-admin-password admin \
  | grep -F 'Admin password'

# Prefer not to be asked at all? Set the password when you create the
# container and Grafana skips the change-password prompt entirely:
#   docker run -d --name grafana-demo --network monnet -p 3001:3000 \
#     -e GF_SECURITY_ADMIN_PASSWORD=admin grafana/grafana

# This container has no volume, so nothing is persisted. Deleting and
# recreating it is also a perfectly good reset:
#   docker rm -f grafana-demo && docker run -d --name grafana-demo ...

# WHERE: your own shell, on the Docker host, not inside any container. It
# calls the Prometheus HTTP API through the port the stack block published
# (-p 9090:9090), so all it needs locally is curl and python3.
# HOW: paste the whole block in one go. The queries call the q() helper
# defined here, so running them in a different shell gives "q: command not
# found".
#
# The single most useful habit in Prometheus: before trusting an expression in
# an alert, ask whether it matches anything at all. An expression that returns
# no series is not an alert that never fires by luck, it is an alert that
# CANNOT fire.
q() { n=$(curl -s --data-urlencode "query=$1" localhost:9090/api/v1/query \
        | python3 -c 'import json,sys; print(len(json.load(sys.stdin)["data"]["result"]))'); \
      printf '%2s series  <-  %s\n' "$n" "$1"; }

q 'pg_stat_database_numbackends / pg_settings_max_connections'
q 'sum(pg_stat_database_numbackends) / on() pg_settings_max_connections'
q 'pg_replication_lag'
q 'pg_replication_lag_seconds'
q 'pg_up'

# Replace the "groups: []" placeholder written in the stack block with the
# rules above, then RESTART Prometheus rather than sending it a reload signal.
# Rewriting a file usually replaces its inode, and a bind mount of a single
# file follows the inode, not the name: the container keeps reading the old
# one, and a SIGHUP reload cheerfully reports success while loading nothing.
docker restart prom-demo

# Wait for it to come back, then confirm the rules are really loaded.
until curl -s localhost:9090/api/v1/rules >/dev/null 2>&1; do sleep 2; done
curl -s localhost:9090/api/v1/rules \
  | python3 -c "import json,sys; print(' rules loaded:', sum(len(g['rules']) \
      for g in json.load(sys.stdin)['data']['groups']))"

# promtool validates YAML structure and PromQL grammar. Run it in CI.
# But notice what it says about the ORIGINAL rules, the ones using
# pg_replication_lag and the unmatched division:
promtool check rules alert_rules.yml

# Then ask Prometheus which rules it actually loaded and how they evaluate.
curl -s localhost:9090/api/v1/rules | python3 -c "
import json, sys
for group in json.load(sys.stdin)['data']['groups']:
    for rule in group['rules']:
        print(f\"  {rule['name']:<28} health={rule['health']}  state={rule['state']}\")"

#!/bin/sh
# verify_alert.sh - an alert you have never seen fire is a guess.
alerts() {
  curl -s localhost:9090/api/v1/alerts | python3 -c '
import json, sys
a = json.load(sys.stdin)["data"]["alerts"]
parts = [x["labels"]["alertname"] + "=" + x["state"] for x in a]
print("  " + (", ".join(parts) or "none active"))'
}
pg_up() {
  curl -s --data-urlencode 'query=pg_up' localhost:9090/api/v1/query | python3 -c '
import json, sys
r = json.load(sys.stdin)["data"]["result"]
print("  pg_up =", r[0]["value"][1] if r else "no series")'
}

echo "before:"; pg_up; alerts

docker stop pgmon-demo >/dev/null
until [ "$(curl -s --data-urlencode 'query=pg_up' localhost:9090/api/v1/query \
  | python3 -c 'import json,sys; r=json.load(sys.stdin)["data"]["result"]; print(r[0]["value"][1] if r else "")')" = "0" ]; do sleep 2; done
echo "database stopped:"; pg_up; alerts

# The "for: 1m" clause keeps the alert pending until the condition has held
# that long. This is what stops a single failed scrape from paging anyone.
until [ "$(curl -s localhost:9090/api/v1/alerts | python3 -c \
  'import json,sys; a=json.load(sys.stdin)["data"]["alerts"]; print(a[0]["state"] if a else "")')" = "firing" ]; do sleep 5; done
echo "after the for: 1m clause:"; alerts

docker start pgmon-demo >/dev/null
until docker exec pgmon-demo pg_isready -U demo -q; do sleep 1; done
echo "database restarted."

"""health_check.py - a dependency-free health check, for teams without Prometheus."""
import os

import psycopg

DSN = "host=localhost port=5432 dbname=demo user=demo password=demo"

# NULLIF guards the denominator: on a freshly reset database blks_hit and
# blks_read are both 0, and 0/0 raises division_by_zero instead of alerting.
CHECKS = {
    "connection_usage": """
        SELECT count(*)::float / current_setting('max_connections')::int
        FROM pg_stat_activity
        WHERE backend_type = 'client backend'
    """,
    "cache_hit_ratio": """
        SELECT sum(blks_hit)::numeric
               / nullif(sum(blks_hit) + sum(blks_read), 0)
        FROM pg_stat_database
        WHERE datname = current_database()
    """,
    "long_running_queries": """
        SELECT count(*)
        FROM pg_stat_activity
        WHERE state = 'active'
          AND backend_type = 'client backend'
          AND pid <> pg_backend_pid()
          AND now() - query_start > interval '60 seconds'
    """,
    "deadlocks": """
        SELECT deadlocks FROM pg_stat_database
        WHERE datname = current_database()
    """,
}


def send_alert(message, severity):
    """Route an alert. Prints unless SLACK_WEBHOOK_URL is set."""
    webhook = os.environ.get("SLACK_WEBHOOK_URL")
    if not webhook:
        print(f"  [{severity.upper():<8}] {message}")
        return
    import requests  # imported here so the script runs without it installed

    color = {"info": "#36a64f", "warning": "#ff9900", "critical": "#ff0000"}
    requests.post(
        webhook,
        json={"attachments": [{"color": color[severity], "text": message}]},
        timeout=5,
    )


def check_database_health():
    # One connection, closed by the context manager even if a check raises.
    with psycopg.connect(DSN) as conn:
        conn.autocommit = True
        results = {
            name: conn.execute(sql).fetchone()[0] for name, sql in CHECKS.items()
        }

    usage = results["connection_usage"]
    if usage > 0.8:
        send_alert(f"Connection usage {usage:.1%} of max_connections", "critical")
    else:
        print(f"  [ok      ] connection usage {usage:.1%}")

    ratio = results["cache_hit_ratio"]
    if ratio is None:
        print("  [ok      ] cache hit ratio n/a (no reads yet)")
    elif ratio < 0.95:
        send_alert(f"Cache hit ratio {ratio:.2%} (<95%)", "warning")
    else:
        print(f"  [ok      ] cache hit ratio {ratio:.2%}")

    long_running = results["long_running_queries"]
    if long_running > 5:
        send_alert(f"{long_running} queries running >60s", "critical")
    else:
        print(f"  [ok      ] {long_running} queries running >60s")

    deadlocks = results["deadlocks"]
    if deadlocks > 0:
        send_alert(f"{deadlocks} deadlocks since stats reset", "warning")
    else:
        print(f"  [ok      ] {deadlocks} deadlocks since stats reset")


if __name__ == "__main__":
    check_database_health()