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.
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" ``` ---
Before writing a single line of code, understand the full picture. Here is how all the pieces connect: ``` 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. ---
### 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: ``` 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. ---
### 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: ``` 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 ```text # 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" ``` ---
### 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" ``` ---
### 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 ``` ---
Imagine you are on-call at Swiggy during a peak dinner hour. Thousands of orders are coming in. Somewhere in the system,...
Before writing a single line of code, understand the full picture. Here is how all the pieces connect: Think of it like ...
What Is Prometheus and Why Are We Using It? Prometheus is a monitoring system that collects metrics from your services a...
Set Up the Project Your project will look like this: Requirements File ---...
Why Configuration Belongs in One File Every setting — the Prometheus URL, which metrics to collect, the anomaly threshol...
What This File Does This file connects to Prometheus, runs our queries, and saves the results to a CSV file. A CSV file ...
What Is Happening Here Training is the process where the Isolation Forest algorithm looks at all your collected metrics ...
Why We Need a Separate Alert Module The alert module has one job: send a structured alert to Alertmanager when the detec...
What This File Does This is the main loop that runs forever. Every 60 seconds it: Collects a single data point from Prom...
Why This Step Is Important You have built the pipeline. Now you need to verify it actually works. The best way is to del...
Why Deploy to Kubernetes Instead of Running Locally? Right now the detection script runs on your laptop. If your laptop ...
---...
❌ Training on data that includes past incidents 💥 The platform had an outage last Tuesday. You collect 24 hours of data...
Model produces too many false positives: Prometheus connection fails: Model file missing after pod restart: bash The mod...
Aligns directly with DevOps, Site Reliability (SRE), and Platform Engineering job descriptions.