Build an Anomaly Detection Pipeline

Build a real AIOps anomaly detection pipeline - collect live Kubernetes metrics from Prometheus, run machine learning to detect anomalies automatically, and fire alerts when something unusual happens. No more waiting for users to report problems.

Related Concepts & TermsAlertmanager Pipe Pipeline Prometheus

Domains & Technologies

Domains

CAPSTONEAIOPSANOMALY-DETECTIONISOLATION-FORESTALERTMANAGERMACHINE-LEARNING

Technologies

PROMETHEUSPYTHONKUBERNETES

Blueprint Walkthrough

Before You Start — Read This First

Imagine you are on-call at Swiggy during a peak dinner hour. Thousands of orders are coming in. Somewhere in the system, the order-service API is getting slower — not broken, just slower. Response times that are usually 50ms are now 180ms. No error. No crash. Just slower.

A traditional alert would not fire. You set your alert at "if error rate > 5%", and the error rate is still 0%. But users are noticing. Orders are taking longer. Complaints are starting.

This is where anomaly detection changes everything. Instead of you manually setting thresholds for every possible problem, you train a machine learning model on what "normal" looks like. When the system behaves differently from normal — even if it is not technically broken — the model notices and alerts you.

This capstone builds that system from scratch.

You will:

Collect real metrics from a Kubernetes cluster using Prometheus
Write a Python pipeline that pulls those metrics automatically
Train an Isolation Forest model to learn what "normal" looks like
Run the model continuously to detect when something is unusual
Send alerts to Alertmanager when anomalies are detected
Build a simple dashboard to visualise what the model is seeing

When you finish, you will have a working AIOps system that watches your infrastructure and catches problems before users do. This is the kind of system that Razorpay, PhonePe, and Hotstar use to keep millions of transactions running smoothly.

Time to complete: 3-4 hours.

What you need before starting:

Python 3.9+ installed
A running Kubernetes cluster (minikube works perfectly)
Prometheus installed on the cluster (from the Monitoring and Logging module)
Basic Python knowledge (from the Python for Ops module)

Bash

## Verify your setup before starting
python3 --version
## Should show Python 3.9 or higher
 
kubectl get pods -n monitoring | grep prometheus
## Should show Prometheus pods Running
 
pip3 install pandas scikit-learn requests matplotlib prometheus-api-client
## Install the libraries we need
echo "✅ Ready to build"

What You Are Building

Before writing a single line of code, understand the full picture. Here is how all the pieces connect:

◈ DIAGRAM

Kubernetes Cluster
        │
        │ (metrics every 15 seconds)
        ▼
   Prometheus
        │
        │ (Python pulls metrics via HTTP API)
        ▼
  Data Collection
  (metrics.py)
        │
        │ (sends data for training)
        ▼
  ML Model Training
  (train.py)
  Isolation Forest
        │
        │ (model saved to disk)
        ▼
  Anomaly Detection
  (detect.py)
  Runs every 60 seconds
        │
        ├── Normal? → log it, continue
        │
        └── Anomaly? → send alert
                            │
                            ▼
                      Alertmanager
                            │
                            ▼
                    Slack / Email / PagerDuty

Think of it like a security guard who has memorised what the building looks like during normal hours. When something unusual happens — a door that is never used suddenly opens, a room that is always quiet becomes noisy — the guard notices immediately. The Isolation Forest model is your security guard. It memorises normal, and flags unusual.

Part 1 — Understand the Tools

What Is Prometheus and Why Are We Using It?

Prometheus is a monitoring system that collects metrics from your services and stores them as time-series data. Time-series means: a value recorded at a specific point in time, repeated over and over. For example:

◈ DIAGRAM

cpu_usage{pod="backend-abc"} = 23%  at 14:00:00
cpu_usage{pod="backend-abc"} = 25%  at 14:00:15
cpu_usage{pod="backend-abc"} = 89%  at 14:00:30  ← something changed here
cpu_usage{pod="backend-abc"} = 91%  at 14:00:45

Prometheus scrapes these metrics from your pods every 15 seconds and stores them. It also has an HTTP API that lets you query historical data — which is exactly what we need to train our ML model.

What Is Isolation Forest and Why Are We Using It?

Isolation Forest is a machine learning algorithm designed specifically for anomaly detection. Here is the intuition behind it in plain language:

Imagine you have 1000 data points representing normal CPU usage (between 20% and 40%). Then one data point comes in at 95%. If you tried to "isolate" (separate) that 95% point by randomly drawing boundaries, it would take very few boundaries to isolate it — because it is alone, far from everything else.

Normal points are packed together. They take many boundaries to isolate. Anomalies are isolated quickly. Isolation Forest counts how many boundaries it takes and uses that as the anomaly score.

Why Isolation Forest over other approaches:

Works without knowing what anomalies look like in advance
Handles high-dimensional data (many metrics at once)
Fast to train and fast to run
Does not need a labelled dataset ("this was an anomaly, this was not")

This last point is important. You do not need historical incident data to train this model. You just need data from when things were running normally.

What Is Alertmanager?

Alertmanager is the component that receives alerts from Prometheus (and from us) and routes them to the right destination — Slack, email, PagerDuty, or any webhook. We will send our anomaly alerts directly to Alertmanager so they appear in the same place as all other infrastructure alerts.

Part 2 — Project Structure

Set Up the Project

Bash

## Create the project directory
mkdir aiops-anomaly-detection && cd aiops-anomaly-detection
 
## Create the file structure
mkdir -p src data models logs k8s
 
touch src/collect_metrics.py
touch src/train_model.py
touch src/detect_anomalies.py
touch src/alert.py
touch src/config.py
touch requirements.txt
 
echo "✅ Project structure created"

Your project will look like this:

◈ DIAGRAM

aiops-anomaly-detection/
  src/
    config.py           ← all settings in one place
    collect_metrics.py  ← pulls data from Prometheus
    train_model.py      ← trains the Isolation Forest model
    detect_anomalies.py ← runs detection every 60 seconds
    alert.py            ← sends alerts to Alertmanager
  data/                 ← collected metrics are stored here
  models/               ← trained model is saved here
  logs/                 ← detection logs go here
  k8s/                  ← Kubernetes manifests to deploy this
  requirements.txt      ← Python dependencies

Requirements File

Bash

# requirements.txt
# Pin versions for reproducible installs
 
pandas==2.1.0
scikit-learn==1.3.0
requests==2.31.0
matplotlib==3.7.2
prometheus-api-client==0.5.3
numpy==1.24.3
joblib==1.3.2
schedule==1.2.0

Bash

pip3 install -r requirements.txt
echo "✅ Dependencies installed"

Part 3 — Configuration

Why Configuration Belongs in One File

Every setting — the Prometheus URL, which metrics to collect, the anomaly threshold — should live in one place. When you want to point the system at a different Prometheus instance or add a new metric, you change one file instead of hunting through four files.

PYTHON

# src/config.py
# All settings for the anomaly detection pipeline
# Change these values to match your environment
 
import os
 
# ── Prometheus Connection ────────────────────────────────────
# The URL where Prometheus is accessible
# For minikube: run 'kubectl port-forward -n monitoring svc/prometheus 9090:9090'
# then use http://localhost:9090
PROMETHEUS_URL = os.getenv("PROMETHEUS_URL", "http://localhost:9090")
 
# ── Metrics to Collect ───────────────────────────────────────
# These are the Prometheus queries we will run
# Each one collects a different signal from the cluster
# We use averages across all pods so the model sees cluster-wide patterns
 
METRICS = {
    # Average CPU usage across all pods (0 to 1, where 1 = 100%)
    "cpu_usage": 'avg(rate(container_cpu_usage_seconds_total{container!=""}[5m]))',
 
    # Average memory usage in bytes
    "memory_usage": 'avg(container_memory_working_set_bytes{container!=""})',
 
    # HTTP request rate — requests per second across all services
    "http_request_rate": 'sum(rate(http_requests_total[5m]))',
 
    # HTTP error rate — what fraction of requests are failing
    "http_error_rate": 'sum(rate(http_requests_total{status=~"5.."}[5m])) / sum(rate(http_requests_total[5m]))',
 
    # Average response time in seconds
    "response_time_p95": 'histogram_quantile(0.95, sum(rate(http_request_duration_seconds_bucket[5m])) by (le))',
}
 
# ── Data Collection ──────────────────────────────────────────
# How far back to collect data for training (in hours)
# 24 hours of data = ~5760 data points at 15s intervals
# More data = better model, but takes longer to collect
TRAINING_DATA_HOURS = 24
 
# How often to collect a data point during detection (in seconds)
DETECTION_INTERVAL_SECONDS = 60
 
# ── Model Settings ───────────────────────────────────────────
# Contamination = what fraction of training data we expect to be anomalous
# 0.05 means "we think about 5% of our training data was slightly unusual"
# If you keep getting too many false alerts, increase this slightly
# If you are missing real anomalies, decrease it
CONTAMINATION = 0.05
 
# Where to save the trained model
MODEL_PATH = "models/isolation_forest.joblib"
 
# Where to save collected data
DATA_PATH = "data/metrics.csv"
 
# ── Alertmanager Connection ──────────────────────────────────
ALERTMANAGER_URL = os.getenv("ALERTMANAGER_URL", "http://localhost:9093")
 
# ── Logging ──────────────────────────────────────────────────
LOG_PATH = "logs/detection.log"

Part 4 — Collecting Metrics from Prometheus

What This File Does

This file connects to Prometheus, runs our queries, and saves the results to a CSV file. A CSV file (Comma-Separated Values) is like a spreadsheet in plain text format. Each row is one point in time, each column is one metric.

Why save to CSV first instead of training directly? Because collecting 24 hours of data takes time. You collect once, save to disk, and then train the model from the saved file. If something goes wrong with training, you do not have to collect all the data again.

PYTHON

# src/collect_metrics.py
# Connects to Prometheus and collects metrics for model training
 
import requests
import pandas as pd
import time
from datetime import datetime, timedelta
from src.config import PROMETHEUS_URL, METRICS, TRAINING_DATA_HOURS, DATA_PATH
 
def query_prometheus_range(metric_name, query, start_time, end_time, step="60s"):
    """
    Query Prometheus for historical data over a time range.
 
    Parameters:
        metric_name: human-readable name (e.g. "cpu_usage")
        query: PromQL query string
        start_time: datetime object for start of range
        end_time: datetime object for end of range
        step: how often to sample (60s = one data point per minute)
 
    Returns:
        list of (timestamp, value) tuples, or empty list if query fails
    """
    url = f"{PROMETHEUS_URL}/api/v1/query_range"
 
    params = {
        "query": query,
        "start": start_time.timestamp(),
        "end": end_time.timestamp(),
        "step": step,
    }
 
    try:
        response = requests.get(url, params=params, timeout=30)
        response.raise_for_status()  # raises exception if HTTP error
 
        data = response.json()
 
        # Prometheus returns results in this format:
        # {"data": {"result": [{"values": [[timestamp, "value"], ...]}]}}
        if data["status"] != "success":
            print(f"⚠️  Prometheus query failed for {metric_name}: {data.get('error', 'unknown error')}")
            return []
 
        results = data["data"]["result"]
        if not results:
            print(f"⚠️  No data returned for {metric_name} — is this metric available in your cluster?")
            return []
 
        # Extract the time-value pairs from the first result
        # (we use avg/sum queries so there should be exactly one result)
        values = results[0]["values"]
        return [(float(ts), float(val)) for ts, val in values]
 
    except requests.exceptions.ConnectionError:
        print(f"❌ Cannot connect to Prometheus at {PROMETHEUS_URL}")
        print("   Run: kubectl port-forward -n monitoring svc/prometheus-operated 9090:9090")
        return []
    except Exception as e:
        print(f"❌ Error querying {metric_name}: {e}")
        return []
 
 
def collect_training_data():
    """
    Collect the last TRAINING_DATA_HOURS of metrics from Prometheus
    and save them to a CSV file for model training.
    """
    print(f"📊 Collecting {TRAINING_DATA_HOURS} hours of metrics from Prometheus...")
    print(f"   URL: {PROMETHEUS_URL}")
    print(f"   Metrics: {list(METRICS.keys())}")
    print()
 
    end_time = datetime.now()
    start_time = end_time - timedelta(hours=TRAINING_DATA_HOURS)
 
    print(f"   Time range: {start_time.strftime('%Y-%m-%d %H:%M')} → {end_time.strftime('%Y-%m-%d %H:%M')}")
    print()
 
    # Collect each metric separately, then merge by timestamp
    all_data = {}
 
    for metric_name, query in METRICS.items():
        print(f"   Collecting {metric_name}...", end=" ")
        values = query_prometheus_range(metric_name, query, start_time, end_time)
 
        if values:
            # Convert to dictionary: {timestamp: value}
            all_data[metric_name] = {ts: val for ts, val in values}
            print(f"✅ {len(values)} data points")
        else:
            print("⚠️  Skipped (no data)")
 
    if not all_data:
        print("\n❌ No metrics collected. Check Prometheus connection and metric names.")
        return False
 
    # Find timestamps that exist in ALL metrics
    # We need complete rows — no missing values
    common_timestamps = set.intersection(*[set(d.keys()) for d in all_data.values()])
    common_timestamps = sorted(common_timestamps)
 
    print(f"\n   Aligning {len(common_timestamps)} common timestamps across all metrics...")
 
    # Build a DataFrame — one row per timestamp, one column per metric
    rows = []
    for ts in common_timestamps:
        row = {"timestamp": datetime.fromtimestamp(ts).strftime("%Y-%m-%d %H:%M:%S")}
        for metric_name in all_data:
            row[metric_name] = all_data[metric_name].get(ts, None)
        rows.append(row)
 
    df = pd.DataFrame(rows)
 
    # Drop any rows with missing values
    original_len = len(df)
    df = df.dropna()
    if len(df) < original_len:
        print(f"   Dropped {original_len - len(df)} rows with missing values")
 
    # Save to CSV
    df.to_csv(DATA_PATH, index=False)
    print(f"\n✅ Saved {len(df)} rows to {DATA_PATH}")
    print(f"   Columns: {list(df.columns)}")
    print(f"\n   Preview of collected data:")
    print(df.head(3).to_string())
 
    return True
 
 
def collect_single_point():
    """
    Collect one data point right now — used during detection, not training.
    Returns a dictionary of {metric_name: current_value} or None if failed.
    """
    current_values = {}
 
    for metric_name, query in METRICS.items():
        url = f"{PROMETHEUS_URL}/api/v1/query"
        params = {"query": query}
 
        try:
            response = requests.get(url, params=params, timeout=10)
            data = response.json()
 
            if data["status"] == "success" and data["data"]["result"]:
                value = float(data["data"]["result"][0]["value"][1])
                current_values[metric_name] = value
            else:
                # If metric is not available, use 0 as fallback
                current_values[metric_name] = 0.0
 
        except Exception:
            current_values[metric_name] = 0.0
 
    return current_values if current_values else None
 
 
if __name__ == "__main__":
    collect_training_data()

Bash

## Run the data collection
## This will take a minute as it queries Prometheus for 24 hours of data
python3 -m src.collect_metrics
 
## Expected output:
## 📊 Collecting 24 hours of metrics from Prometheus...
##    Collecting cpu_usage... ✅ 1440 data points
##    Collecting memory_usage... ✅ 1440 data points
##    ...
## ✅ Saved 1440 rows to data/metrics.csv
 
## Check what was collected
head -5 data/metrics.csv

Part 5 — Training the Model

What Is Happening Here

Training is the process where the Isolation Forest algorithm looks at all your collected metrics and learns what "normal" looks like. It does not need labels — you do not need to mark which rows were good and which were bad. It figures out the pattern by itself.

After training, the model is saved to disk using joblib. Joblib is a Python library that serialises Python objects to files. Serialise means "convert to a format that can be saved and loaded later." This way you train once and run detection many times without retraining every time.

PYTHON

# src/train_model.py
# Trains an Isolation Forest model on the collected metrics
# and saves it to disk for use during detection
 
import pandas as pd
import numpy as np
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler
import joblib
import os
from src.config import DATA_PATH, MODEL_PATH, CONTAMINATION
 
def load_and_prepare_data():
    """
    Load the CSV data and prepare it for model training.
 
    Preparation steps:
    1. Drop the timestamp column (the model does not need to know when)
    2. Handle any remaining missing values
    3. Scale the features so CPU usage and memory bytes are comparable
    """
 
    if not os.path.exists(DATA_PATH):
        print(f"❌ Data file not found at {DATA_PATH}")
        print("   Run collect_metrics.py first to collect training data")
        return None, None, None
 
    print(f"📂 Loading data from {DATA_PATH}...")
    df = pd.read_csv(DATA_PATH)
 
    print(f"   Shape: {df.shape[0]} rows × {df.shape[1]} columns")
    print(f"   Columns: {list(df.columns)}")
 
    # Drop the timestamp column — it is not a feature, it is an identifier
    feature_columns = [col for col in df.columns if col != "timestamp"]
    X = df[feature_columns].copy()
 
    # Fill any remaining NaN values with column median
    # Median is better than mean here because a spike anomaly in training
    # data would distort the mean but not the median
    for col in X.columns:
        if X[col].isnull().any():
            median_val = X[col].median()
            X[col] = X[col].fillna(median_val)
            print(f"   Filled NaN in {col} with median {median_val:.4f}")
 
    # Scale the features
    # Without scaling, memory_usage (in billions of bytes) would dominate
    # the model completely and cpu_usage (0-1) would be ignored
    # StandardScaler converts each feature to: (value - mean) / std_deviation
    # After scaling, all features have mean=0 and std=1
    scaler = StandardScaler()
    X_scaled = scaler.fit_transform(X)
 
    print(f"\n   Feature statistics after scaling:")
    for i, col in enumerate(feature_columns):
        original_mean = df[col].mean()
        original_std = df[col].std()
        print(f"   {col}: mean={original_mean:.4f}, std={original_std:.4f}")
 
    return X_scaled, scaler, feature_columns
 
 
def train_and_save_model(X_scaled, scaler, feature_columns):
    """
    Train the Isolation Forest model and save it along with the scaler.
 
    Why save the scaler too?
    During detection, incoming data must be scaled the SAME WAY as training data.
    If we trained on data scaled with mean=100MB and std=50MB,
    we must scale detection data the same way.
    The scaler remembers the training mean and std — we must save it.
    """
 
    print(f"\n🤖 Training Isolation Forest model...")
    print(f"   Contamination: {CONTAMINATION} ({CONTAMINATION*100:.0f}% of training data expected to be unusual)")
    print(f"   Training samples: {X_scaled.shape[0]}")
    print(f"   Features: {X_scaled.shape[1]}")
 
    # Create and train the model
    # n_estimators=100: build 100 trees (more = more accurate but slower)
    # random_state=42: makes results reproducible (same seed = same model)
    model = IsolationForest(
        n_estimators=100,
        contamination=CONTAMINATION,
        random_state=42,
        n_jobs=-1,  # use all CPU cores for faster training
    )
 
    model.fit(X_scaled)
 
    # Get anomaly scores for the training data
    # score_samples returns negative values: more negative = more anomalous
    scores = model.score_samples(X_scaled)
    predictions = model.predict(X_scaled)
 
    # Isolation Forest labels: 1 = normal, -1 = anomaly
    n_anomalies = (predictions == -1).sum()
    n_normal = (predictions == 1).sum()
 
    print(f"\n   Training results:")
    print(f"   Normal points:  {n_normal} ({n_normal/len(predictions)*100:.1f}%)")
    print(f"   Anomaly points: {n_anomalies} ({n_anomalies/len(predictions)*100:.1f}%)")
    print(f"   Anomaly score range: {scores.min():.4f} to {scores.max():.4f}")
    print(f"   (more negative = more anomalous)")
 
    # Save the model and scaler together in a dictionary
    # This ensures they are always used together
    model_package = {
        "model": model,
        "scaler": scaler,
        "feature_columns": feature_columns,
        "training_samples": X_scaled.shape[0],
        "contamination": CONTAMINATION,
    }
 
    os.makedirs(os.path.dirname(MODEL_PATH), exist_ok=True)
    joblib.dump(model_package, MODEL_PATH)
 
    print(f"\n✅ Model saved to {MODEL_PATH}")
    print(f"   File size: {os.path.getsize(MODEL_PATH) / 1024:.1f} KB")
 
    return model, scores
 
 
def visualise_training_data():
    """
    Create a simple chart showing the training data and anomaly scores.
    Saves to data/training_visualisation.png
    """
    try:
        import matplotlib.pyplot as plt
        import matplotlib.dates as mdates
 
        df = pd.read_csv(DATA_PATH)
        df["timestamp"] = pd.to_datetime(df["timestamp"])
 
        fig, axes = plt.subplots(len(df.columns) - 1, 1,
                                 figsize=(14, 3 * (len(df.columns) - 1)))
        fig.suptitle("Training Data — Metrics Over Time", fontsize=14, fontweight="bold")
 
        feature_cols = [col for col in df.columns if col != "timestamp"]
        for i, col in enumerate(feature_cols):
            ax = axes[i] if len(feature_cols) > 1 else axes
            ax.plot(df["timestamp"], df[col], linewidth=0.8, color="#F97316")
            ax.set_ylabel(col, fontsize=9)
            ax.set_xlabel("")
            ax.xaxis.set_major_formatter(mdates.DateFormatter("%H:%M"))
            ax.grid(True, alpha=0.3)
 
        plt.tight_layout()
        plt.savefig("data/training_visualisation.png", dpi=150, bbox_inches="tight")
        print("📊 Training visualisation saved to data/training_visualisation.png")
 
    except Exception as e:
        print(f"⚠️  Could not create visualisation: {e}")
 
 
if __name__ == "__main__":
    X_scaled, scaler, feature_columns = load_and_prepare_data()
 
    if X_scaled is not None:
        model, scores = train_and_save_model(X_scaled, scaler, feature_columns)
        visualise_training_data()
        print("\n🎯 Model is ready. Run detect_anomalies.py to start detection.")

Bash

## Train the model
python3 -m src.train_model
 
## Expected output:
## 📂 Loading data from data/metrics.csv...
##    Shape: 1440 rows × 6 columns
## 🤖 Training Isolation Forest model...
##    Training samples: 1440
##    Normal points:  1368 (95.0%)
##    Anomaly points: 72 (5.0%)
## ✅ Model saved to models/isolation_forest.joblib
 
## Verify the model file exists
ls -lh models/

Part 6 — Sending Alerts

Why We Need a Separate Alert Module

The alert module has one job: send a structured alert to Alertmanager when the detection module finds an anomaly. Keeping it separate means the detection code does not need to know anything about how alerts are sent. If you later want to send alerts to Slack directly instead of Alertmanager, you only change this file.

PYTHON

# src/alert.py
# Sends anomaly alerts to Alertmanager
 
import requests
import json
from datetime import datetime
from src.config import ALERTMANAGER_URL
 
def send_anomaly_alert(anomaly_score, metric_values, detection_time=None):
    """
    Send an anomaly alert to Alertmanager.
 
    Alertmanager expects a specific JSON format.
    Each alert has:
    - labels: key-value pairs that identify the alert (used for routing and grouping)
    - annotations: human-readable information about the alert
    - startsAt: when the anomaly was detected
 
    Parameters:
        anomaly_score: float, how anomalous this point is (more negative = worse)
        metric_values: dict of {metric_name: current_value}
        detection_time: datetime when anomaly was detected (defaults to now)
    """
 
    if detection_time is None:
        detection_time = datetime.now()
 
    # Format the metric values for the alert description
    metric_summary = " | ".join([
        f"{k}: {v:.4f}" for k, v in metric_values.items()
    ])
 
    # Alertmanager alert format
    # The "fingerprint" is what Alertmanager uses to deduplicate alerts
    # We use the alert name + hour so repeated anomalies in the same hour
    # do not create hundreds of duplicate notifications
    alert = [
        {
            "labels": {
                "alertname": "AnomalyDetected",
                "severity": "warning",
                "source": "aiops-anomaly-detector",
                "environment": "production",
            },
            "annotations": {
                "summary": "Anomaly detected in cluster metrics",
                "description": (
                    f"The anomaly detection model flagged unusual behavior. "
                    f"Anomaly score: {anomaly_score:.4f} "
                    f"(threshold: more negative than -0.1 is anomalous). "
                    f"Current metrics: {metric_summary}"
                ),
                "runbook_url": "https://your-backstage.internal/docs/runbooks/anomaly-detection",
            },
            "startsAt": detection_time.isoformat() + "Z",
        }
    ]
 
    try:
        response = requests.post(
            f"{ALERTMANAGER_URL}/api/v1/alerts",
            data=json.dumps(alert),
            headers={"Content-Type": "application/json"},
            timeout=10,
        )
 
        if response.status_code == 200:
            print(f"   🔔 Alert sent to Alertmanager successfully")
            return True
        else:
            print(f"   ⚠️  Alertmanager returned status {response.status_code}: {response.text}")
            return False
 
    except requests.exceptions.ConnectionError:
        print(f"   ⚠️  Cannot reach Alertmanager at {ALERTMANAGER_URL}")
        print(f"   Run: kubectl port-forward -n monitoring svc/alertmanager 9093:9093")
        return False
    except Exception as e:
        print(f"   ⚠️  Alert failed: {e}")
        return False
 
 
def send_recovery_alert(metric_values):
    """
    Send a recovery notification when metrics return to normal.
    This resolves the warning in Alertmanager so it does not show as ongoing.
    """
 
    metric_summary = " | ".join([
        f"{k}: {v:.4f}" for k, v in metric_values.items()
    ])
 
    alert = [
        {
            "labels": {
                "alertname": "AnomalyDetected",
                "severity": "warning",
                "source": "aiops-anomaly-detector",
                "environment": "production",
            },
            "annotations": {
                "summary": "Anomaly resolved — metrics returned to normal",
                "description": f"Current metrics are within normal range: {metric_summary}",
            },
            "endsAt": datetime.now().isoformat() + "Z",
        }
    ]
 
    try:
        requests.post(
            f"{ALERTMANAGER_URL}/api/v1/alerts",
            data=json.dumps(alert),
            headers={"Content-Type": "application/json"},
            timeout=10,
        )
        print(f"   ✅ Recovery alert sent")
    except Exception:
        pass  # Recovery alerts are best-effort

Part 7 — Running Continuous Detection

What This File Does

This is the main loop that runs forever. Every 60 seconds it:

Collects a single data point from Prometheus
Scales it the same way the training data was scaled
Asks the model: "Is this normal or anomalous?"
If anomalous: sends an alert and logs the event
If normal: logs it and continues

PYTHON

# src/detect_anomalies.py
# The main detection loop — runs continuously, checking for anomalies
 
import joblib
import numpy as np
import pandas as pd
import time
import logging
import os
from datetime import datetime
from src.config import MODEL_PATH, DETECTION_INTERVAL_SECONDS, LOG_PATH
from src.collect_metrics import collect_single_point
from src.alert import send_anomaly_alert, send_recovery_alert
 
# Set up logging
# This writes both to the terminal and to a log file
os.makedirs(os.path.dirname(LOG_PATH), exist_ok=True)
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[
        logging.FileHandler(LOG_PATH),
        logging.StreamHandler(),  # also print to terminal
    ],
)
logger = logging.getLogger(__name__)
 
 
def load_model():
    """
    Load the trained model and scaler from disk.
    Exits if the model file does not exist — you must train first.
    """
    if not os.path.exists(MODEL_PATH):
        logger.error(f"Model not found at {MODEL_PATH}")
        logger.error("Run train_model.py first to train the model")
        exit(1)
 
    logger.info(f"Loading model from {MODEL_PATH}...")
    package = joblib.load(MODEL_PATH)
 
    model = package["model"]
    scaler = package["scaler"]
    feature_columns = package["feature_columns"]
 
    logger.info(f"Model loaded — trained on {package['training_samples']} samples")
    logger.info(f"Features: {feature_columns}")
 
    return model, scaler, feature_columns
 
 
def detect_once(model, scaler, feature_columns):
    """
    Run one detection cycle:
    1. Collect current metric values from Prometheus
    2. Scale them using the training scaler
    3. Get the anomaly score from the model
    4. Return (is_anomaly, score, metric_values)
    """
 
    # Collect current values
    metric_values = collect_single_point()
    if metric_values is None:
        logger.warning("Could not collect metrics — skipping this cycle")
        return None, None, None
 
    # Build a single-row DataFrame with columns in the same order as training
    # Order matters because the scaler expects columns in training order
    row = {col: metric_values.get(col, 0.0) for col in feature_columns}
    X = pd.DataFrame([row])[feature_columns]
 
    # Scale using the training scaler
    X_scaled = scaler.transform(X)
 
    # Get the anomaly score
    # score_samples returns an array of one value (since we have one row)
    score = model.score_samples(X_scaled)[0]
 
    # Get the prediction: 1 = normal, -1 = anomaly
    prediction = model.predict(X_scaled)[0]
    is_anomaly = prediction == -1
 
    return is_anomaly, score, metric_values
 
 
def run_detection_loop():
    """
    The main detection loop.
    Runs forever, checking for anomalies every DETECTION_INTERVAL_SECONDS.
    Sends alerts when anomalies are detected and recovery notices when resolved.
    """
 
    model, scaler, feature_columns = load_model()
 
    logger.info("=" * 60)
    logger.info("🚀 AIOps Anomaly Detection Started")
    logger.info(f"   Checking every {DETECTION_INTERVAL_SECONDS} seconds")
    logger.info(f"   Prometheus: {os.getenv('PROMETHEUS_URL', 'http://localhost:9090')}")
    logger.info(f"   Alertmanager: {os.getenv('ALERTMANAGER_URL', 'http://localhost:9093')}")
    logger.info("=" * 60)
 
    # Track consecutive anomalies to avoid alert storms
    # Only send an alert after 2 consecutive anomalous readings
    # This reduces false positives from brief spikes
    consecutive_anomalies = 0
    ANOMALY_THRESHOLD = 2
 
    # Track whether we are currently in an anomaly state
    currently_alerting = False
 
    while True:
        try:
            cycle_start = datetime.now()
            logger.info(f"--- Detection cycle at {cycle_start.strftime('%H:%M:%S')} ---")
 
            is_anomaly, score, metric_values = detect_once(model, scaler, feature_columns)
 
            if is_anomaly is None:
                # Metrics collection failed, try again next cycle
                time.sleep(DETECTION_INTERVAL_SECONDS)
                continue
 
            # Log the current values and score
            logger.info(f"   Anomaly score: {score:.4f} ({'⚠️ ANOMALOUS' if is_anomaly else '✅ NORMAL'})")
            for metric, value in metric_values.items():
                logger.info(f"   {metric}: {value:.6f}")
 
            if is_anomaly:
                consecutive_anomalies += 1
                logger.warning(f"   Anomaly detected ({consecutive_anomalies}/{ANOMALY_THRESHOLD} consecutive)")
 
                if consecutive_anomalies >= ANOMALY_THRESHOLD and not currently_alerting:
                    logger.warning("🚨 ALERT: Sending anomaly notification to Alertmanager")
                    send_anomaly_alert(score, metric_values, cycle_start)
                    currently_alerting = True
 
            else:
                if currently_alerting:
                    logger.info("✅ Metrics returned to normal — sending recovery notification")
                    send_recovery_alert(metric_values)
                    currently_alerting = False
 
                consecutive_anomalies = 0
 
        except KeyboardInterrupt:
            logger.info("\n⛔ Detection stopped by user")
            break
        except Exception as e:
            logger.error(f"Unexpected error in detection cycle: {e}")
 
        # Wait for the next cycle
        time.sleep(DETECTION_INTERVAL_SECONDS)
 
 
if __name__ == "__main__":
    run_detection_loop()

Bash

## Start continuous detection
## Keep this running in a terminal — it will check every 60 seconds
python3 -m src.detect_anomalies
 
## Expected output:
## 2024-01-15 14:23:01 [INFO] Loading model from models/isolation_forest.joblib...
## 2024-01-15 14:23:01 [INFO] 🚀 AIOps Anomaly Detection Started
## 2024-01-15 14:24:01 [INFO] --- Detection cycle at 14:24:01 ---
## 2024-01-15 14:24:01 [INFO]    Anomaly score: -0.0823 (✅ NORMAL)
## 2024-01-15 14:24:01 [INFO]    cpu_usage: 0.234100
## ...

Part 8 — Inject a Real Anomaly and Watch It Get Detected

Why This Step Is Important

You have built the pipeline. Now you need to verify it actually works. The best way is to deliberately cause a spike in CPU usage and watch the model detect it.

Bash

## Terminal 1: Run detection (keep this open)
python3 -m src.detect_anomalies
 
## Terminal 2: Inject CPU spike using a stress test
## This runs a CPU-intensive workload inside Kubernetes
kubectl run cpu-stress \
  --image=containerstack/cpustress \
  --restart=Never \
  -- --cpu 4 --timeout 120s
## This runs 4 CPU threads for 120 seconds — should cause a noticeable spike
 
## Watch the detection output in Terminal 1
## Within 2-3 minutes you should see:
## ⚠️  Anomaly detected (1/2 consecutive)
## ⚠️  Anomaly detected (2/2 consecutive)
## 🚨 ALERT: Sending anomaly notification to Alertmanager
 
## After 120 seconds the stress pod stops
## Watch for the recovery message:
## ✅ Metrics returned to normal — sending recovery notification
 
## Clean up the stress pod
kubectl delete pod cpu-stress

Part 9 — Deploy to Kubernetes

Why Deploy to Kubernetes Instead of Running Locally?

Right now the detection script runs on your laptop. If your laptop closes, detection stops. In production, you want this running continuously in the cluster itself.

YAML

# k8s/deployment.yaml
# Deploys the anomaly detector as a Kubernetes Deployment
 
apiVersion: apps/v1
kind: Deployment
metadata:
  name: anomaly-detector
  namespace: monitoring
  labels:
    app: anomaly-detector
spec:
  replicas: 1   ## only one replica — running two would send duplicate alerts
  selector:
    matchLabels:
      app: anomaly-detector
  template:
    metadata:
      labels:
        app: anomaly-detector
    spec:
      containers:
      - name: detector
        ## Build and push your image before applying this
        ## docker build -t your-registry/anomaly-detector:v1.0.0 .
        ## docker push your-registry/anomaly-detector:v1.0.0
        image: your-registry/anomaly-detector:v1.0.0
        env:
        - name: PROMETHEUS_URL
          ## Use the internal cluster DNS — no port-forward needed inside the cluster
          value: "http://prometheus-operated.monitoring.svc:9090"
        - name: ALERTMANAGER_URL
          value: "http://alertmanager-operated.monitoring.svc:9093"
        resources:
          requests:
            cpu: "100m"
            memory: "256Mi"
          limits:
            cpu: "500m"
            memory: "512Mi"
        ## Mount the trained model into the container
        volumeMounts:
        - name: model-volume
          mountPath: /app/models
        - name: data-volume
          mountPath: /app/data
      volumes:
      ## In production, use a PersistentVolumeClaim so the model
      ## survives pod restarts. For now we use emptyDir for simplicity.
      - name: model-volume
        emptyDir: {}
      - name: data-volume
        emptyDir: {}

Dockerfile

# Dockerfile
# Build the anomaly detector image
 
FROM python:3.11-slim
 
WORKDIR /app
 
## Install dependencies first (Docker layer caching)
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
 
## Copy source code
COPY src/ ./src/
 
## Copy the trained model and data
COPY models/ ./models/
COPY data/ ./data/
 
## Run the detector
CMD ["python3", "-m", "src.detect_anomalies"]

Bash

## Build and push the image
docker build -t your-registry/anomaly-detector:v1.0.0 .
docker push your-registry/anomaly-detector:v1.0.0
 
## Deploy to Kubernetes
kubectl apply -f k8s/deployment.yaml
 
## Verify it is running
kubectl get pods -n monitoring | grep anomaly-detector
 
## Watch the logs
kubectl logs -n monitoring deployment/anomaly-detector -f

Production Checklist

Bash

## ─── 1. Data was collected successfully ─────────────────────
ls -lh data/metrics.csv
## File should exist and be non-empty (at least a few KB)
 
wc -l data/metrics.csv
## Should show at least 500 lines (500 data points = ~8 hours minimum)
 
## ─── 2. Model was trained and saved ─────────────────────────
ls -lh models/isolation_forest.joblib
## File should exist and be at least 100KB
 
python3 -c "
import joblib
pkg = joblib.load('models/isolation_forest.joblib')
print('Features:', pkg['feature_columns'])
print('Training samples:', pkg['training_samples'])
print('✅ Model loads correctly')
"
 
## ─── 3. Single detection cycle works ────────────────────────
python3 -c "
from src.collect_metrics import collect_single_point
values = collect_single_point()
print('Current metrics:', values)
if values:
    print('✅ Can collect metrics from Prometheus')
else:
    print('❌ Cannot collect metrics — check Prometheus connection')
"
 
## ─── 4. Alert reaches Alertmanager ──────────────────────────
## Port-forward Alertmanager UI and check
kubectl port-forward -n monitoring svc/alertmanager-operated 9093:9093 &
## Open http://localhost:9093 in browser
## After injecting a CPU spike, the AnomalyDetected alert should appear
 
## ─── 5. Recovery detection works ─────────────────────────────
## After the stress test ends, check the logs for:
grep "returned to normal" logs/detection.log
## Should show the recovery message
 
## ─── 6. No false positives during normal operation ──────────
grep "ANOMALOUS" logs/detection.log | wc -l
## Count false positives over 1 hour of normal operation
## Should be 0 or at most 1-2
 
echo "✅ Production checklist complete"

Common Production Mistakes

❌ Training on data that includes past incidents 💥 The platform had an outage last Tuesday. You collect 24 hours of data that includes that outage. The model learns that high CPU (90%+) is "normal" because it appeared in training data. During the next incident, the model does not alert because it has seen this pattern before and considers it within normal range. ✅ Before collecting training data, check Grafana for the past 24-48 hours. Make sure you are collecting from a period of genuinely normal operation. If you had an incident recently, collect from a different 24-hour window. Add a comment in config.py noting when the training data was collected.

❌ Setting contamination too low and getting alert storms 💥 You set contamination to 0.001 (0.1%). The model is extremely strict. Every minor CPU blip triggers an alert. During a busy Monday morning, the detector sends 47 alerts in two hours. Engineers start ignoring all anomaly alerts because they are always firing for nothing. The boy who cried wolf problem — when a real anomaly happens, nobody pays attention. ✅ Start with contamination=0.05 (5%) and tune from experience. Track false positive rate over one week. If you get more than 2-3 false alerts per day, increase contamination. If you miss real anomalies, decrease it. The goal is fewer than 1 false alert per day in normal operation.

❌ Never retraining the model 💥 You trained the model in January on a cluster with 10 pods. By June the cluster has grown to 80 pods. What was an anomalous CPU reading in January (50%) is now perfectly normal during peak hours. The model keeps alerting on normal behaviour because it has not seen the new normal. Engineers disable the system entirely because it has become useless. ✅ Retrain the model monthly or after any significant infrastructure change (major scaling, new services deployed, architecture changes). Automate retraining with a Kubernetes CronJob that runs collect_metrics and train_model on a schedule.

❌ Only monitoring CPU and memory 💥 A database connection pool fills up completely. Queries start queuing. Response times go from 50ms to 8 seconds. Users see timeouts. But CPU is normal (queries are waiting, not executing). Memory is normal. The anomaly detector sees nothing wrong and does not alert. ✅ Include application-level metrics in your model — HTTP error rate, response time percentiles, database connection pool utilisation, cache hit rate. These catch the failures that infrastructure metrics miss.

Debugging Playbook

Model produces too many false positives:

Bash

## Step 1: Check what the anomaly scores look like during normal operation
python3 -c "
import joblib, numpy as np
from src.collect_metrics import collect_single_point
from sklearn.preprocessing import StandardScaler
 
pkg = joblib.load('models/isolation_forest.joblib')
model = pkg['model']
scaler = pkg['scaler']
feature_columns = pkg['feature_columns']
 
import pandas as pd
metrics = collect_single_point()
row = {col: metrics.get(col, 0.0) for col in feature_columns}
X = pd.DataFrame([row])[feature_columns]
X_scaled = scaler.transform(X)
score = model.score_samples(X_scaled)[0]
print(f'Current score: {score:.4f}')
print(f'Score < -0.1 triggers alert')
"
 
## Step 2: If scores are consistently around -0.08 to -0.12 during normal ops,
## your contamination setting is too low — increase it
## Edit config.py: CONTAMINATION = 0.08
 
## Step 3: Retrain with the new setting
python3 -m src.train_model

Prometheus connection fails:

Bash

## Step 1: Verify Prometheus is accessible
curl -s http://localhost:9090/api/v1/query?query=up | jq '.status'
## Should return: "success"
 
## If it fails, start port-forward:
kubectl port-forward -n monitoring svc/prometheus-operated 9090:9090 &
 
## Step 2: Verify your specific metrics exist
curl -s "http://localhost:9090/api/v1/query?query=avg(rate(container_cpu_usage_seconds_total{container!=\"\"}[5m]))" | jq '.data.result | length'
## Should return a number > 0
## If 0, the metric does not exist in your cluster
## Try a simpler metric first: ?query=up

Model file missing after pod restart:

Bash

## The model is lost because emptyDir volumes are not persistent
## Solution: use a PersistentVolumeClaim
 
kubectl apply -f - << 'EOF'
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
  name: anomaly-detector-models
  namespace: monitoring
spec:
  accessModes:
    - ReadWriteOnce
  resources:
    requests:
      storage: 1Gi

Videos & Guides

No external walkthroughs, video tutorials, or reference files attached to this blueprint yet.