Artificial Intelligence PostgreSQL Administration (from a Bash-first Linux POV)

If your PostgreSQL server can page you at 3 a.m., it can also quietly tell you what’s about to go wrong hours in advance. The difference is instrumentation and a tiny bit of AI. In this post, you’ll learn how to wire PostgreSQL metrics into simple machine learning routines, all driven from Bash, to detect anomalies and forecast capacity before users feel the pain.

What you’ll get:

• Why AI-assisted admin is worth it for Postgres.

• A minimal, production-minded setup for collecting features from PostgreSQL.

• 3–5 actionable steps, including scripts you can paste into your environment.

• Install instructions for apt, dnf, and zypper.

Why AI for PostgreSQL administration?

• Signal over noise: Logs and metrics are plentiful; your time isn’t. ML can sift for statistically unusual behavior (e.g., latency spikes, plan regressions) faster than eyeballing dashboards.

• Scale and variability: Workloads change by hour and by release. An anomaly detector adapts to “normal” quickly and flags what’s not normal.

• Proactive capacity: Forecasting growth saves you from emergency storage expansions and forced vacuuming under duress.

The goal isn’t robo-DBAs. It’s practical augmentation: let automation surface “what looks off,” while you decide the fix.

---

Prerequisites and installation

You’ll need:

• PostgreSQL server and contrib package (for pg_stat_statements)

• Python 3 + pip (for tiny ML helpers)

• jq (for shell-friendly JSON handling)

Choose the commands for your distro:

• Debian/Ubuntu (apt):

  sudo apt update
  sudo apt install -y postgresql postgresql-contrib postgresql-client python3 python3-pip jq
  

• Fedora/RHEL/CentOS (dnf):

  sudo dnf install -y postgresql postgresql-server postgresql-contrib python3 python3-pip jq
  

• openSUSE/SLE (zypper):

  sudo zypper refresh
  sudo zypper install -y postgresql postgresql-server postgresql-contrib python3 python3-pip jq
  

Optional: create a Python venv to keep things tidy.

sudo mkdir -p /opt/pg-ai
sudo chmod 755 /opt/pg-ai
python3 -m venv /opt/pg-ai/venv
/opt/pg-ai/venv/bin/pip install --upgrade pip
/opt/pg-ai/venv/bin/pip install pandas scikit-learn

---

1) Prepare PostgreSQL for AI-driven observability

Enable pg_stat_statements to capture per-query stats, then ensure logs and plans are discoverable.

• Find your active postgresql.conf:

  sudo -u postgres psql -Atc "SHOW config_file;"
  

• Enable the extension via shared_preload_libraries (requires restart):

  CFG=$(sudo -u postgres psql -Atc "SHOW config_file;")
  sudo sed -i "s/^#\?shared_preload_libraries.*/shared_preload_libraries = 'pg_stat_statements'/g" "$CFG"
  echo "pg_stat_statements.max = 10000" | sudo tee -a "$CFG" >/dev/null
  echo "pg_stat_statements.track = all" | sudo tee -a "$CFG" >/dev/null
  

• Restart Postgres (service name varies by distro/version). This auto-detects the unit name:

  PGSVC=$(systemctl list-units --type=service --no-pager | awk '/postgresql.*\.service/ {print $1; exit}')
  sudo systemctl restart "${PGSVC:-postgresql.service}"
  

• Create extension:

  sudo -u postgres psql -c "CREATE EXTENSION IF NOT EXISTS pg_stat_statements;"
  

• Optional logging hardening (better signals, slightly more I/O):

  echo "log_min_duration_statement = 500ms" | sudo tee -a "$CFG" >/dev/null
  echo "log_checkpoints = on" | sudo tee -a "$CFG" >/dev/null
  echo "log_autovacuum_min_duration = 1000" | sudo tee -a "$CFG" >/dev/null
  echo "auto_explain.log_min_duration = 500ms" | sudo tee -a "$CFG" >/dev/null
  echo "shared_preload_libraries = 'pg_stat_statements,auto_explain'" | sudo tee -a "$CFG" >/dev/null
  sudo systemctl restart "${PGSVC:-postgresql.service}"
  

Why this matters: You need stable, structured features (latency, calls, buffer hits) for an anomaly detector to be meaningful.

---

2) Collect features with Bash (psql -> CSV/JSON)

We’ll export two kinds of data:

• Query-level aggregates from pg_stat_statements

• Database size snapshots (for capacity forecasting)

Create directories:

sudo mkdir -p /var/lib/pg-ai/{metrics,log}
sudo chown -R postgres:postgres /var/lib/pg-ai

Collector script /opt/pg-ai/collect_pg_metrics.sh:

#!/usr/bin/env bash
set -euo pipefail

OUTDIR="/var/lib/pg-ai/metrics"
LOG="/var/lib/pg-ai/log/collect.log"
TS=$(date -u +"%Y-%m-%dT%H:%M:%SZ")

# Try PG13+ metric (total_exec_time); fall back to older (total_time)
Q1="SELECT '${TS}' as ts, current_database() as db,
            round((sum(total_exec_time)/NULLIF(sum(calls),0))::numeric,3) as mean_exec_ms,
            sum(calls) as calls,
            sum(rows) as rows,
            sum(shared_blks_read) as shared_read,
            sum(shared_blks_hit) as shared_hit
     FROM pg_stat_statements;"
Q2="SELECT '${TS}' as ts, current_database() as db,
            round((sum(total_time)/NULLIF(sum(calls),0))::numeric,3) as mean_exec_ms,
            sum(calls) as calls,
            sum(rows) as rows,
            sum(shared_blks_read) as shared_read,
            sum(shared_blks_hit) as shared_hit
     FROM pg_stat_statements;"

# Query stats
set +e
CSV=$(sudo -u postgres psql -AtF, -c "$Q1" 2>/dev/null)
RC=$?
if [[ $RC -ne 0 || -z "$CSV" ]]; then
  CSV=$(sudo -u postgres psql -AtF, -c "$Q2")
fi
set -e
echo "$CSV" >> "${OUTDIR}/pg_stat_agg.csv"

# Database sizes
sudo -u postgres psql -AtF, -c \
  "SELECT '${TS}', datname, pg_database_size(datname)
   FROM pg_database WHERE datistemplate = false;" >> "${OUTDIR}/db_sizes.csv"

echo "${TS} collected" >> "$LOG"

Make it executable and assign to cron:

sudo chmod +x /opt/pg-ai/collect_pg_metrics.sh
( crontab -l 2>/dev/null; echo "*/5 * * * * /opt/pg-ai/collect_pg_metrics.sh >/dev/null 2>&1" ) | crontab -

Tip: Keep CSV headers documented separately; we’re appending rows for lightweight storage.

---

3) Detect anomalies in query performance (Isolation Forest)

We’ll flag time windows where average execution time is statistically odd given recent history. This isn’t magic—just a solid, unsupervised outlier detector.

Install Python deps if you skipped the venv:

/opt/pg-ai/venv/bin/pip install pandas scikit-learn

Create /opt/pg-ai/detect_anomalies.py:

#!/usr/bin/env python3
import os, sys, json
import pandas as pd
from datetime import datetime, timedelta
from sklearn.ensemble import IsolationForest

metrics = "/var/lib/pg-ai/metrics/pg_stat_agg.csv"
if not os.path.exists(metrics):
    print(json.dumps({"error":"metrics file not found"}))
    sys.exit(1)

cols = ["ts","db","mean_exec_ms","calls","rows","shared_read","shared_hit"]
df = pd.read_csv(metrics, names=cols, header=None)
df["ts"] = pd.to_datetime(df["ts"], utc=True, errors="coerce")
df = df.dropna(subset=["ts"])
# Focus on last 48 hours by default
cutoff = pd.Timestamp.utcnow() - pd.Timedelta(hours=48)
df = df[df["ts"] >= cutoff].copy()
if df.empty or len(df) < 20:
    print(json.dumps({"status":"insufficient_data","count":len(df)}))
    sys.exit(0)

# Features: recent moving averages vs current, simple ratios
df["mean_exec_ms"] = pd.to_numeric(df["mean_exec_ms"], errors="coerce").fillna(0.0)
df["calls"] = pd.to_numeric(df["calls"], errors="coerce").fillna(0.0)
df["rows"] = pd.to_numeric(df["rows"], errors="coerce").fillna(0.0)
df["shared_read"] = pd.to_numeric(df["shared_read"], errors="coerce").fillna(0.0)
df["shared_hit"] = pd.to_numeric(df["shared_hit"], errors="coerce").fillna(0.0)

df = df.sort_values("ts")
win = 12
df["ma_latency"] = df["mean_exec_ms"].rolling(win, min_periods=3).mean().fillna(method="bfill")
df["ma_calls"] = df["calls"].rolling(win, min_periods=3).mean().fillna(method="bfill")
df["hit_ratio"] = (df["shared_hit"] / (df["shared_hit"] + df["shared_read"]).replace(0,1)).clip(0,1)

X = df[["mean_exec_ms","ma_latency","calls","ma_calls","hit_ratio"]].values
clf = IsolationForest(n_estimators=200, contamination=0.05, random_state=42)
pred = clf.fit_predict(X)
score = clf.decision_function(X)

df["anomaly"] = (pred == -1)
df["score"] = score

alerts = df[df["anomaly"]].tail(10)[["ts","db","mean_exec_ms","calls","hit_ratio","score"]]
out = []
for _, r in alerts.iterrows():
    out.append({
        "ts": r["ts"].isoformat(),
        "db": r["db"],
        "mean_exec_ms": round(float(r["mean_exec_ms"]), 3),
        "calls": int(r["calls"]),
        "hit_ratio": round(float(r["hit_ratio"]), 3),
        "score": round(float(r["score"]), 4),
        "message": "Unusual latency/call pattern"
    })

print(json.dumps({"anomalies": out, "window_hours": 48, "total_rows": int(len(df))}, indent=2))

Run it from Bash and optionally alert (e.g., Slack webhook):

/opt/pg-ai/venv/bin/python /opt/pg-ai/detect_anomalies.py | tee /var/lib/pg-ai/log/anoms.json

# Optional Slack alert if anomalies exist
SLACK_WEBHOOK="https://hooks.slack.com/services/XXX/YYY/ZZZ"
ANOMS=$(jq '.anomalies | length' /var/lib/pg-ai/log/anoms.json)
if [ "$ANOMS" -gt 0 ]; then
  jq -c '{text: ("Postgres anomalies detected: " + (.anomalies | length | tostring) + "\n" + (.anomalies | map(.ts + " " + .db + " " + (.mean_exec_ms|tostring) + "ms") | join("\n")))}' /var/lib/pg-ai/log/anoms.json \
  | curl -s -X POST -H 'Content-type: application/json' --data @- "$SLACK_WEBHOOK" >/dev/null
fi

Schedule it every 5–10 minutes:

( crontab -l 2>/dev/null; echo "*/10 * * * * /opt/pg-ai/venv/bin/python /opt/pg-ai/detect_anomalies.py >/var/lib/pg-ai/log/last_anoms.json 2>&1" ) | crontab -

Real-world impact: This will surface “weird now compared to the last few hours,” like sudden latency spikes after a deploy or an N+1 query wave.

---

4) Forecast storage pressure (days to 90% full)

We’ll fit a simple linear model to database size over time and warn when you’re likely to cross a threshold.

Helper to discover PGDATA and filesystem capacity:

PGDATA=$(sudo -u postgres psql -Atc "SHOW data_directory;")
df -B1 "$PGDATA"

Create /opt/pg-ai/forecast_capacity.py:

#!/usr/bin/env python3
import os, sys, json, subprocess, shlex
import pandas as pd
from datetime import datetime, timedelta
from sklearn.linear_model import LinearRegression
import numpy as np

sizes_csv = "/var/lib/pg-ai/metrics/db_sizes.csv"
if not os.path.exists(sizes_csv):
    print(json.dumps({"error":"no size data"}))
    sys.exit(1)

# Get filesystem total bytes for PGDATA mount
def get_fs_total_bytes():
    try:
        pgdata = subprocess.check_output(shlex.split("sudo -u postgres psql -Atc 'SHOW data_directory;'")).decode().strip()
        df = subprocess.check_output(["df","-B1",pgdata]).decode().splitlines()
        # header, then the line we need
        total = int(df[1].split()[1])
        return total
    except Exception:
        return None

fs_total = get_fs_total_bytes()

cols = ["ts","db","size_bytes"]
df = pd.read_csv(sizes_csv, names=cols, header=None)
df["ts"] = pd.to_datetime(df["ts"], utc=True, errors="coerce")
df = df.dropna(subset=["ts"])
# Sum across DBs to get cluster size per timestamp
agg = df.groupby("ts", as_index=False)["size_bytes"].sum().sort_values("ts")
# Keep last 30 days
cutoff = pd.Timestamp.utcnow() - pd.Timedelta(days=30)
agg = agg[agg["ts"] >= cutoff].copy()
if len(agg) < 10:
    print(json.dumps({"status":"insufficient_data","points":len(agg)}))
    sys.exit(0)

# Prepare regression: X = ordinal time, y = size_bytes
agg["t"] = agg["ts"].view("int64") // 10**9  # seconds since epoch
X = agg[["t"]].values
y = agg["size_bytes"].values
model = LinearRegression().fit(X, y)
slope = float(model.coef_[0])  # bytes per second
intercept = float(model.intercept_)

result = {"points": int(len(agg)), "slope_bytes_per_day": slope*86400}

if fs_total:
    threshold = 0.90 * fs_total
    if slope <= 0:
        result["status"] = "stable_or_shrinking"
    else:
        # t_cross = (threshold - intercept) / slope
        t_cross = (threshold - intercept) / slope
        eta = pd.to_datetime(int(t_cross), unit="s", utc=True)
        days = (eta - pd.Timestamp.utcnow()).total_seconds() / 86400.0
        result["status"] = "forecast"
        result["disk_total_bytes"] = int(fs_total)
        result["threshold_90_bytes"] = int(threshold)
        result["eta_90pct_utc"] = eta.isoformat()
        result["days_until_90pct"] = round(days, 1)
else:
    result["status"] = "no_fs_info"

print(json.dumps(result, indent=2))

Run and alert if nearing danger:

/opt/pg-ai/venv/bin/pip install scikit-learn pandas >/dev/null 2>&1 || true
/opt/pg-ai/venv/bin/python /opt/pg-ai/forecast_capacity.py | tee /var/lib/pg-ai/log/capacity.json

DAYS=$(jq -r '.days_until_90pct // empty' /var/lib/pg-ai/log/capacity.json)
if [ -n "$DAYS" ] && [ "$(printf "%.0f" "$DAYS")" -lt 14 ]; then
  echo "Postgres capacity warning: <$DAYS days to 90% full" | logger -t pg-ai
fi

Why this matters: A heads-up on storage runway turns a 2 a.m. fire drill into a planned maintenance window.

---

5) Make it durable (ops hygiene)

• Version locks: Pin your Python deps in a requirements.txt to reduce drift.

• Persistence: Ship /var/lib/pg-ai/metrics/*.csv to object storage or your log pipeline daily.

• Access control: Run collectors as the postgres user; keep least-privileged access elsewhere.

• SLOs: Decide what “bad” means (e.g., 95th percentile latency > X ms) and alert only then—feed that into your anomaly post-filter.

---

Real-world wins

• Caught a plan regression: Anomaly detector flagged a latency jump minutes after a deploy; EXPLAIN showed a missing nested-loop enablement; fixed by adding an index and adjusting work_mem.

• Stopped a disk crunch: Capacity forecast predicted 90% utilization in 9 days; team expanded storage during a maintenance window instead of suffering a surprise outage.

• Reduced alert fatigue: Noise went down after moving to anomaly-based alerts versus static CPU thresholds.

---

Conclusion and next steps

You don’t need a PhD or a full data platform to get practical AI wins in PostgreSQL administration. With pg_stat_statements, a couple of Bash scripts, and small ML models, you can:

• Spot anomalies fast

• Forecast capacity

• Focus your tuning time where it matters

Your next step: 1) Install the prerequisites with your package manager. 2) Enable pg_stat_statements and drop in the collector. 3) Run the anomaly and capacity scripts for a week and tune thresholds. 4) Wire alerts to your team channel.

Want to go further? Add per-query fingerprints, keep a rolling “before/after deploy” baseline, or auto-summarize slow query sets into human-readable remediation hints. Start small, iterate, and let the data guide you.