Posted on
• Artificial Intelligence

AI Agents for Monitoring

Author
  • User
    linuxbash
    Posts by this author
    Posts by this author

AI Agents for Monitoring: a Bash-first playbook for smarter, quieter ops

If your on-call rotation still feels like a slot machine of noisy alerts, you’re not alone. Classic monitoring stacks are great at collecting numbers and firing thresholds; they’re less great at telling you what actually matters right now, or what to do about it. Enter AI agents for monitoring: small, composable helpers you can bolt onto your existing Linux stack to summarize, prioritize, and even auto-remediate common issues—without ripping out your current tooling.

This guide shows you how to stand up two tiny agents on any Linux box using Bash and Python:

  • A log triage agent that summarizes spikes and suggests likely causes

  • A metrics anomaly agent that learns baselines and flags outliers

  • Optional safe auto-remediation with allowlisted actions

  • Webhook notifications to your chat or incident system

You’ll get actionable, explainable output and fewer false positives—while keeping control and observability in your hands.

Why AI agents for monitoring are worth your time

  • Reduce noise, keep signal: Thresholds scream when a number crosses a line. Agents can group related errors, deduplicate repeats, and explain impact in plain language.

  • Faster triage: Summaries and suggested next steps shrink mean-time-to-understand, a big part of MTTR.

  • Start small, scale safely: You can pilot AI-assisted summarization locally with guardrails, then graduate to more automation as confidence grows.

  • Works with what you have: Journald, sysstat, cron/systemd timers, webhooks—no wholesale migration needed.

What you’ll build

  • triage_agent.py: batches recent logs, detects patterns (timeouts, OOMs, 5xx storms), and optionally calls an LLM endpoint for a short summary and classification.

  • metrics_agent.py: gathers a few host metrics, learns a rolling baseline, and flags robust anomalies using median/MAD.

  • actions.sh: optional allowlisted actions (e.g., safe service restarts) with dry-run.

  • notify.sh: ships JSON events to a webhook or syslog.

Everything runs as a dedicated system user with minimal privileges. You can expand from here: more detectors, more playbooks, deeper integrations.

Prerequisites and installation (apt, dnf, zypper)

You only need standard Linux tools. Install prerequisites using your package manager.

  • Debian/Ubuntu (apt):
sudo apt update
sudo apt install -y python3 python3-venv python3-pip jq curl git coreutils sysstat
  • Fedora/RHEL/CentOS (dnf):
sudo dnf install -y python3 python3-virtualenv python3-pip jq curl git coreutils sysstat
  • openSUSE/SLES (zypper):
sudo zypper refresh
sudo zypper install -y python3 python3-virtualenv python3-pip jq curl git coreutils sysstat

Create a dedicated user and directories:

sudo useradd --system --home /opt/ai-monitor --shell /usr/sbin/nologin ai-monitor || true
sudo mkdir -p /opt/ai-monitor /var/log/ai-monitor /var/lib/ai-monitor
sudo chown -R ai-monitor:ai-monitor /opt/ai-monitor /var/log/ai-monitor /var/lib/ai-monitor

Give the agent read access to the journal (varies by distro; add both if unsure):

# For journald access
sudo usermod -aG systemd-journal ai-monitor 2>/dev/null || true
# On some Debian/Ubuntu systems logs are readable to adm group
sudo usermod -aG adm ai-monitor 2>/dev/null || true

Optional: if you later want safe service restarts from the agent, give ai-monitor limited sudo for specific units (example for nginx):

echo 'ai-monitor ALL=(root) NOPASSWD: /usr/bin/systemctl restart nginx' | sudo tee /etc/sudoers.d/ai-monitor-restarts >/dev/null
sudo chmod 440 /etc/sudoers.d/ai-monitor-restarts

Step 1 — Directory layout

sudo -u ai-monitor bash -c '
  cd /opt/ai-monitor
  touch triage_agent.py metrics_agent.py actions.sh notify.sh
  chmod +x actions.sh notify.sh
'

Step 2 — Log triage agent

This script reads log lines from stdin, groups and scores signals (errors, timeouts, OOM kills, HTTP 5xx bursts), and outputs a compact JSON incident with an optional LLM-powered summary when AI_ENDPOINT and AI_API_KEY are set. If no LLM is configured, it falls back to a rule-based summary.

Create /opt/ai-monitor/triage_agent.py:

#!/usr/bin/env python3
import os, sys, re, json, time, socket, subprocess, urllib.request

MAX_LINES = int(os.getenv("TRIAGE_MAX_LINES", "800"))
BATCH_SECONDS = int(os.getenv("TRIAGE_BATCH_SECONDS", "60"))
HOST = socket.gethostname()

# Simple signal detectors
PATTERNS = {
    "oom_kill": re.compile(r"Out of memory|Killed process \d+|oom-killer", re.I),
    "timeout": re.compile(r"timeout|timed out|deadline exceeded", re.I),
    "conn_reset": re.compile(r"connection reset|broken pipe", re.I),
    "auth_fail": re.compile(r"failed password|authentication failure", re.I),
    "segfault": re.compile(r"segfault|segmentation fault", re.I),
    "http_5xx": re.compile(r"\s5\d{2}\s"),
    "db_error": re.compile(r"(deadlock|too many connections|lock wait timeout|read-only)* (mysql|postgres|pg|sqlite)", re.I),
}

def parse_program(line):
    # Try "prog[pid]:" or "prog:" patterns
    m = re.search(r"\s([a-zA-Z0-9_\-\.]+)(?:\[\d+\])?:", line)
    return m.group(1) if m else "unknown"

def score_lines(lines):
    counts = {k:0 for k in PATTERNS}
    by_prog = {}
    errs = 0
    for ln in lines:
        prog = parse_program(ln)
        by_prog[prog] = by_prog.get(prog, 0) + 1
        if re.search(r"\b(error|fail|crit|alert|panic)\b", ln, re.I):
            errs += 1
        for k, rx in PATTERNS.items():
            if rx.search(ln):
                counts[k] += 1
    top_prog = sorted(by_prog.items(), key=lambda x: x[1], reverse=True)[:3]
    return counts, top_prog, errs

def heuristic_summary(counts, top_prog, total):
    signals = sorted([(v,k) for k,v in counts.items() if v>0], reverse=True)
    if not signals and total == 0:
        return "No notable signals in this window.", "none", 0.1
    top = [k for _,k in signals[:3]]
    prog_str = ", ".join([f"{p}({c})" for p,c in top_prog]) if top_prog else "unknown"
    if "oom_kill" in top:
        return f"OOM activity detected; processes killed. Top programs: {prog_str}.", "resource_exhaustion", 0.8
    if "http_5xx" in top and any(p for p,_ in top_prog if "nginx" in p or "apache" in p):
        return f"HTTP 5xx surge likely from web tier. Top programs: {prog_str}.", "service_error", 0.7
    if "timeout" in top or "conn_reset" in top:
        return f"Network timeouts/connection resets observed. Top programs: {prog_str}.", "network_instability", 0.6
    if "auth_fail" in top:
        return f"Spike in authentication failures. Top programs: {prog_str}.", "security_auth", 0.6
    if "segfault" in top:
        return f"Segmentation faults seen. Top programs: {prog_str}.", "crash", 0.7
    return f"Elevated errors across {prog_str}.", "generic_error", 0.5

def llm_summarize(text, default_summary, default_type, default_conf):
    endpoint = os.getenv("AI_ENDPOINT", "").strip()
    api_key = os.getenv("AI_API_KEY", "").strip()
    model = os.getenv("AI_MODEL", "gpt-4o-mini")
    if not endpoint or not api_key:
        return default_summary, default_type, default_conf
    try:
        prompt = ("Summarize the root cause and impact in 2-3 sentences; return a JSON object with keys "
                  "'summary', 'incident_type' (one of: service_error, resource_exhaustion, network_instability, security_auth, crash, generic_error, none), "
                  "'confidence' (0..1), and 'suggested_action' (like restart_service:<name> or investigate:<hint>). "
                  "Logs:\n" + text[:12000])
        req = {
            "model": model,
            "messages": [
                {"role":"system", "content":"You are a reliable SRE assistant."},
                {"role":"user", "content":prompt}
            ],
            "temperature": 0.2
        }
        data = json.dumps(req).encode("utf-8")
        r = urllib.request.Request(endpoint, data=data, headers={
            "Content-Type":"application/json",
            "Authorization": f"Bearer {api_key}"
        })
        with urllib.request.urlopen(r, timeout=20) as resp:
            out = resp.read().decode("utf-8", "ignore")
        # Try to extract JSON either from tool output or direct content
        j = None
        try:
            obj = json.loads(out)
            # OpenAI-compatible: choices[0].message.content
            content = obj.get("choices",[{}])[0].get("message",{}).get("content","")
            j = json.loads(content) if content.strip().startswith("{") else None
        except Exception:
            j = None
        if j and isinstance(j, dict):
            return j.get("summary", default_summary), j.get("incident_type", default_type), float(j.get("confidence", default_conf)), j.get("suggested_action","")
    except Exception:
        pass
    return default_summary, default_type, default_conf, ""

def main():
    start = time.time()
    buf = []
    for ln in sys.stdin:
        buf.append(ln.rstrip("\n"))
        if len(buf) >= MAX_LINES:
            break
        if time.time() - start >= BATCH_SECONDS:
            break
    total = len(buf)
    counts, top_prog, errs = score_lines(buf)
    default_summary, incident_type, confidence = heuristic_summary(counts, top_prog, total)
    suggested_action = ""
    joined = "\n".join(buf)
    ret = llm_summarize(joined, default_summary, incident_type, confidence)
    if isinstance(ret, tuple) and len(ret) == 4:
        summary, incident_type, confidence, suggested_action = ret
    else:
        summary, incident_type, confidence = ret
    event = {
        "kind": "log_triage",
        "host": HOST,
        "total_lines": total,
        "errors": errs,
        "top_programs": [{"program":p,"count":c} for p,c in top_prog],
        "counts": counts,
        "incident_type": incident_type,
        "confidence": round(confidence,2),
        "summary": summary,
        "suggested_action": suggested_action,
        "ts": int(time.time())
    }
    print(json.dumps(event, ensure_ascii=False))
if __name__ == "__main__":
    main()

How to run (ad-hoc):

# Last 2 minutes of logs into the triage agent
journalctl --since "2 min ago" --no-pager -o short | /usr/bin/python3 /opt/ai-monitor/triage_agent.py

Optional LLM config (use your endpoint/token; can be OpenAI-compatible or self-hosted):

export AI_ENDPOINT="https://api.openai.com/v1/chat/completions"
export AI_API_KEY="sk-..."
export AI_MODEL="gpt-4o-mini"

Tip: If you don’t set AI_ENDPOINT/AI_API_KEY, the heuristic summary still works.

Step 3 — Metrics anomaly agent (MAD-based, no external deps)

This script collects a few host metrics, learns baselines, and uses a robust z-score (Median Absolute Deviation) to flag spikes or drops. State is persisted under /var/lib/ai-monitor/state.json.

Create /opt/ai-monitor/metrics_agent.py:

#!/usr/bin/env python3
import os, sys, json, time, socket, shutil
from statistics import median

HOST = socket.gethostname()
STATE_FILE = os.getenv("AI_MONITOR_STATE", "/var/lib/ai-monitor/state.json")
WINDOW = int(os.getenv("AI_MONITOR_WINDOW", "288"))  # ~24h if every 5m
THRESH = float(os.getenv("AI_MONITOR_Z", "6.0"))     # robust z threshold

def read_mem_available():
    avail = total = 0
    with open("/proc/meminfo") as f:
        for ln in f:
            if ln.startswith("MemAvailable:"):
                avail = int(ln.split()[1]) * 1024
            if ln.startswith("MemTotal:"):
                total = int(ln.split()[1]) * 1024
    pct_free = (avail/total)*100 if total>0 else 0
    return round(100-pct_free,2)  # memory used percent

def read_load1():
    with open("/proc/loadavg") as f:
        return float(f.read().split()[0])

def read_disk_used_pct(path="/"):
    t = shutil.disk_usage(path)
    used = t.used / t.total * 100 if t.total>0 else 0
    return round(used,2)

def robust_z(values, x):
    if len(values) < 10:
        return 0.0
    m = median(values)
    abs_dev = [abs(v - m) for v in values]
    mad = median(abs_dev) or 1e-6
    return 0.6745 * (x - m) / mad

def load_state():
    try:
        with open(STATE_FILE) as f:
            return json.load(f)
    except Exception:
        return {"load1": [], "mem_used_pct": [], "disk_used_pct_root": []}

def save_state(st):
    tmp = STATE_FILE + ".tmp"
    with open(tmp, "w") as f:
        json.dump(st, f)
    os.replace(tmp, STATE_FILE)

def main():
    st = load_state()
    now = int(time.time())
    metrics = {
        "load1": read_load1(),
        "mem_used_pct": read_mem_available(),
        "disk_used_pct_root": read_disk_used_pct("/")
    }
    events = []
    for k, val in metrics.items():
        z = robust_z(st.get(k, []), val)
        if abs(z) >= THRESH:
            events.append({
                "kind":"metric_anomaly",
                "host": HOST,
                "metric": k,
                "value": val,
                "robust_z": round(z,2),
                "ts": now,
                "summary": f"Anomalous {k}={val} (robust_z={z:.2f})",
                "incident_type": "resource_exhaustion" if z>0 else "sudden_drop"
            })
        # update state
        series = (st.get(k, []) + [val])[-WINDOW:]
        st[k] = series
    save_state(st)
    for e in events:
        print(json.dumps(e))
if __name__ == "__main__":
    main()

Run ad-hoc:

/usr/bin/python3 /opt/ai-monitor/metrics_agent.py

Step 4 — Safe action runner and notifier

Only run allowlisted actions, default to dry-run, and log everything. Start conservative: suggest actions first, then enable automation.

Create /opt/ai-monitor/actions.sh:

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

# Comma-separated safelist: e.g. AI_MONITOR_ALLOW_RESTART="nginx,redis-server"
ALLOW_RESTART="${AI_MONITOR_ALLOW_RESTART:-}"
DRY_RUN="${AI_MONITOR_DRY_RUN:-1}"   # 1 = simulate by default

is_allowed() {
  local svc="$1"
  [[ ",${ALLOW_RESTART}," == *",$svc,"* ]]
}

while IFS= read -r line; do
  [[ -z "$line" ]] && continue
  # Pass through non-JSON lines
  if ! echo "$line" | jq . >/dev/null 2>&1; then
    echo "$line"
    continue
  fi

  kind=$(echo "$line" | jq -r '.kind // empty')
  suggested=$(echo "$line" | jq -r '.suggested_action // empty')
  host=$(echo "$line" | jq -r '.host // "unknown"')

  if [[ "$kind" == "log_triage" && "$suggested" == restart_service:* ]]; then
    svc="${suggested#restart_service:}"
    if [[ -n "$svc" ]] && is_allowed "$svc"; then
      if [[ "$DRY_RUN" == "1" ]]; then
        echo "$line" | jq --arg a "Would restart $svc (dry-run)" '. + {action_taken:$a}'
      else
        # Use sudo if needed, constrain via sudoers
        if sudo /usr/bin/systemctl restart "$svc"; then
          echo "$line" | jq --arg a "Restarted $svc" '. + {action_taken:$a}'
        else
          echo "$line" | jq --arg a "Failed to restart $svc" '. + {action_error:$a}'
        fi
      fi
      continue
    fi
  fi

  # Default: no action
  echo "$line" | jq '. + {action_taken:"none"}'
done

Create /opt/ai-monitor/notify.sh:

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

WEBHOOK="${MONITOR_WEBHOOK_URL:-}"
CHANNEL="${MONITOR_CHANNEL:-ai-monitor}"
HOST="$(hostname)"

post_webhook() {
  local payload="$1"
  curl -fsS -m 10 -H "Content-Type: application/json" -X POST -d "$payload" "$WEBHOOK" >/dev/null
}

while IFS= read -r line; do
  [[ -z "$line" ]] && continue

  if [[ -n "$WEBHOOK" ]]; then
    # Wrap line as JSON if it's not valid JSON
    if echo "$line" | jq . >/dev/null 2>&1; then
      payload=$(jq -cn --arg ch "$CHANNEL" --arg h "$HOST" --argjson ev "$line" \
        '{channel:$ch, host:$h, event:$ev}')
      post_webhook "$payload" || logger -t ai-monitor "Webhook failed"
    else
      payload=$(jq -cn --arg ch "$CHANNEL" --arg h "$HOST" --arg msg "$line" \
        '{channel:$ch, host:$h, message:$msg}')
      post_webhook "$payload" || logger -t ai-monitor "Webhook failed"
    fi
  else
    # Fallback to syslog
    logger -t ai-monitor -- "$line"
  fi
done

Environment knobs you can set (system-wide in /etc/default/ai-monitor or per-unit):

  • AI_MONITOR_DRY_RUN=1

  • AI_MONITOR_ALLOW_RESTART="nginx,redis-server"

  • MONITOR_WEBHOOK_URL="https://your.webhook/endpoint"

Step 5 — Wire it up with systemd timers

Create unit and timer files.

sudo tee /etc/systemd/system/ai-monitor-triage.service >/dev/null <<'EOF'
[Unit]
Description=AI log triage (batch recent logs)
After=network-online.target

[Service]
Type=oneshot
User=ai-monitor
Group=ai-monitor
Environment="AI_MONITOR_DRY_RUN=1"
# Optional LLM:
# Environment="AI_ENDPOINT=https://api.openai.com/v1/chat/completions"
# Environment="AI_API_KEY=sk-..."
# Environment="AI_MODEL=gpt-4o-mini"
# Allow service restarts (set sudoers if enabling) and safelist:
# Environment="AI_MONITOR_ALLOW_RESTART=nginx,redis-server"
# Notifications:
# Environment="MONITOR_WEBHOOK_URL=https://hooks.example.com/..."
ExecStart=/bin/bash -lc 'journalctl --since "2 min ago" --no-pager -o short \
  | /usr/bin/python3 /opt/ai-monitor/triage_agent.py \
  | /bin/bash /opt/ai-monitor/actions.sh \
  | /bin/bash /opt/ai-monitor/notify.sh'
StandardOutput=journal
StandardError=journal
EOF

sudo tee /etc/systemd/system/ai-monitor-triage.timer >/dev/null <<'EOF'
[Unit]
Description=Run AI log triage every 2 minutes

[Timer]
OnCalendar=*:0/2
Persistent=true

[Install]
WantedBy=timers.target
EOF

Metrics anomaly every 5 minutes:

sudo tee /etc/systemd/system/ai-monitor-metrics.service >/dev/null <<'EOF'
[Unit]
Description=AI metrics anomaly agent
After=network-online.target

[Service]
Type=oneshot
User=ai-monitor
Group=ai-monitor
Environment="AI_MONITOR_WINDOW=288"
Environment="AI_MONITOR_Z=6.0"
ExecStart=/bin/bash -lc '/usr/bin/python3 /opt/ai-monitor/metrics_agent.py \
  | /bin/bash /opt/ai-monitor/actions.sh \
  | /bin/bash /opt/ai-monitor/notify.sh'
StandardOutput=journal
StandardError=journal
EOF

sudo tee /etc/systemd/system/ai-monitor-metrics.timer >/dev/null <<'EOF'
[Unit]
Description=Run AI metrics anomaly agent every 5 minutes

[Timer]
OnCalendar=*:0/5
Persistent=true

[Install]
WantedBy=timers.target
EOF

Enable and start:

sudo systemctl daemon-reload
sudo systemctl enable --now ai-monitor-triage.timer ai-monitor-metrics.timer

Check status and logs:

systemctl list-timers | grep ai-monitor
journalctl -u ai-monitor-triage.service -n 50 -e
journalctl -u ai-monitor-metrics.service -n 50 -e

Real-world style examples

  • Web tier error storm: Nginx starts returning 502/504 after a deploy. Triage agent groups the 5xx surge, flags probable service_error with nginx atop top_programs, and suggests restart_service:nginx. With DRY_RUN=1 you get a preview; add nginx to AI_MONITOR_ALLOW_RESTART to let the agent self-heal next time.

  • Sudden memory pressure: A runaway process fills memory. Metrics agent detects a high robust_z on mem_used_pct and emits an incident_type=resource_exhaustion. Your webhook receives a compact JSON; your runbook or on-call can act before OOM kills hit.

  • SSH brute-force: auth_fail spikes. The triage agent classifies security_auth, sending you a short summary rather than a hundred identical alerts.

Safety and operability tips

  • Start with DRY_RUN and summaries only. Flip to automation selectively for well-understood playbooks.

  • Whitelist exact actions and services. Avoid wildcards.

  • Log everything: inputs, decisions, and actions.

  • Rate-limit actions (e.g., only one restart per 30 minutes per service).

  • Least privilege: the ai-monitor user should only have the sudo commands it truly needs.

  • Treat the LLM as advisory: keep heuristics and guardrails in place even when summaries come from AI.

Your next step

  • Deploy the timers, watch a few cycles, and tune thresholds and allowlists.

  • Add detectors: disk IO latency (iostat), TCP retransmits (ss), app-specific logs.

  • Enrich notifications with runbook URLs and dashboards.

  • If you run an internal LLM, point AI_ENDPOINT at it for privacy and resilience.

AI agents don’t replace your monitoring—they make it calmer and more actionable. Start with the two tiny helpers above, and iterate towards a smarter, safer on-call.

Related posts