Anomaly Detection at Scale | The Garnet Wiki

Anomaly detection at scale identifies unexpected patterns in high-throughput, high-variability data streams, where system health, user behavior, and business metrics must be monitored continuously. It matters when a single missed anomaly can cascade into user churn, revenue loss, or infrastructure failure—when the cost of a false positive is nontrivial and the value of catching a rare, subtle shift is high. This is not ad hoc data exploration; it is the infrastructure for observing and reacting to the shape of data across thousands of time series, every minute.

Monitoring a High-Volume Metric Stream

To detect anomalies in real time across a fleet of services, one begins by ingesting metric data into a time-series database. The data arrives via a stream processor like Kafka or Pulsar, with each message formatted as a JSON object containing a timestamp, a metric name, a set of tags, and a value. A typical ingestion event looks like:

{
  "timestamp": "2024-06-12T08:15:30.450Z",
  "metric": "http.request.latency.ms",
  "tags": {
    "service": "api-gateway",
    "region": "us-west-2",
    "version": "v2.3.1"
  },
  "value": 143.2
}

The ingestion pipeline is driven by a Kafka consumer group consuming from a topic called metrics.raw. Each consumer processes messages and writes them to a PostgreSQL table metrics with the schema:

CREATE TABLE metrics (
  id BIGSERIAL PRIMARY KEY,
  metric_name TEXT NOT NULL,
  tags JSONB NOT NULL,
  value DOUBLE PRECISION NOT NULL,
  timestamp TIMESTAMP WITH TIME ZONE NOT NULL,
  created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
  INDEX idx_metric_name (metric_name),
  INDEX idx_timestamp (timestamp),
  INDEX idx_tags (tags)
);

A materialized view metrics_by_time is maintained to enable fast aggregation over time windows:

CREATE MATERIALIZED VIEW metrics_by_time AS
SELECT
  metric_name,
  tags,
  timestamp,
  value,
  AVG(value) OVER (
    PARTITION BY metric_name, tags
    ORDER BY timestamp
    ROWS BETWEEN 10 PRECEDING AND CURRENT ROW
  ) AS rolling_avg,
  STDDEV(value) OVER (
    PARTITION BY metric_name, tags
    ORDER BY timestamp
    ROWS BETWEEN 10 PRECEDING AND CURRENT ROW
  ) AS rolling_std
FROM metrics
WHERE timestamp > NOW() - INTERVAL '1 hour';

This materialized view is refreshed every 5 minutes via a cron job using systemctl:


[Unit]
Description=Refresh metrics_by_time materialized view
After=network.target

[Service]
Type=oneshot
ExecStart=/usr/bin/psql -c "REFRESH MATERIALIZED VIEW metrics_by_time"
User=postgres
Group=postgres
Environment=PGHOST=localhost

[Install]
WantedBy=multi-user.target

Detecting Anomalies Using Statistical Thresholds

With data available, anomaly detection is applied by computing a statistical baseline for each metric-tag combination and flagging values that deviate significantly. The core algorithm uses a z-score method: for each (metric_name, tags) group, compute the mean and standard deviation over a rolling window of 100 points, and flag any new point where the z-score exceeds a threshold.

A detection job runs every 1 minute via cron, reading from metrics_by_time and writing results to a table anomalies:

CREATE TABLE anomalies (
  id BIGSERIAL PRIMARY KEY,
  metric_name TEXT NOT NULL,
  tags JSONB NOT NULL,
  timestamp TIMESTAMP WITH TIME ZONE NOT NULL,
  value DOUBLE PRECISION NOT NULL,
  z_score DOUBLE PRECISION NOT NULL,
  severity TEXT NOT NULL,
  detected_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
  INDEX idx_severity (severity),
  INDEX idx_timestamp (timestamp),
  INDEX idx_metric_name (metric_name)
);

The detection script, written in Python and run via python3 detect_anomalies.py, performs the following steps:

Fetches all (metric_name, tags) groups that have at least 100 data points in the last 24 hours.
For each group, computes the mean and standard deviation over the last 100 points.
For each new point in the last 10 minutes, computes z_score = (value - mean) / std.
Flags anomalies where z_score > 3.0 and assigns severity: low (z_score > 2.0), medium (z_score > 2.5), high (z_score > 3.0).

# detect_anomalies.py
import psycopg2
from datetime import datetime, timedelta
from collections import defaultdict

conn = psycopg2.connect("host=localhost dbname=metrics user=postgres password=secret")
cursor = conn.cursor()

# Get all metric-tag combinations with sufficient history
cursor.execute("""
    SELECT DISTINCT metric_name, tags
    FROM metrics_by_time
    WHERE timestamp > NOW() - INTERVAL '24 hours'
    GROUP BY metric_name, tags
    HAVING COUNT(*) >= 100
""")

for metric_name, tags in cursor.fetchall():
    # Load last 100 points for this group
    cursor.execute("""
        SELECT value
        FROM metrics_by_time
        WHERE metric_name = %s AND tags = %s
        ORDER BY timestamp DESC
        LIMIT 100
    """, (metric_name, tags))

    values = [row[0] for row in cursor.fetchall()]
    mean = sum(values) / len(values)
    variance = sum((v - mean) ** 2 for v in values) / len(values)
    std = variance ** 0.5

    # Detect anomalies in the last 10 minutes
    cursor.execute("""
        SELECT timestamp, value
        FROM metrics_by_time
        WHERE metric_name = %s AND tags = %s
        AND timestamp > NOW() - INTERVAL '10 minutes'
        ORDER BY timestamp
    """, (metric_name, tags))

    for timestamp, value in cursor.fetchall():
        z_score = (value - mean) / std
        severity = "high" if z_score > 3.0 else "medium" if z_score > 2.5 else "low"

        cursor.execute("""
            INSERT INTO anomalies (metric_name, tags, timestamp, value, z_score, severity)
            VALUES (%s, %s, %s, %s, %s, %s)
        """, (metric_name, tags, timestamp, value, z_score, severity))

conn.commit()
conn.close()

Handling High Cardinality: The Perils of Tag Explosion

A common failure mode is the explosion of metric-tag combinations due to fine-grained tagging. When each service instance, region, and version combination generates a unique time series, the number of (metric_name, tags) groups can easily exceed 100,000. This leads to:

Memory pressure: The detection script must keep 100 recent values per group in memory, which can exceed 1 GB when processing 50,000 groups.
Slow aggregation: A SELECT DISTINCT over 100,000 groups can take 15 seconds, making the detection job bottlenecked.
Incomplete baselines: Groups with fewer than 100 points are not detected, leading to false negatives for new services or configurations.

To mitigate, the system adds a pre-aggregation step: every 15 minutes, a job compacts raw metrics into a metrics_summary table:

CREATE TABLE metrics_summary (
  metric_name TEXT,
  tags JSONB,
  start_time TIMESTAMP WITH TIME ZONE,
  end_time TIMESTAMP WITH TIME ZONE,
  count BIGINT,
  avg_value DOUBLE PRECISION,
  min_value DOUBLE PRECISION,
  max_value DOUBLE PRECISION,
  std_value DOUBLE PRECISION,
  PRIMARY KEY (metric_name, tags, start_time)
);

This is populated by a materialized view that aggregates every 15 minutes:

CREATE MATERIALIZED VIEW metrics_summary_mv AS
SELECT
  metric_name,
  tags,
  FLOOR(EXTRACT(EPOCH FROM timestamp) / 900) * 900 AS bucket,
  COUNT(*) AS count,
  AVG(value) AS avg_value,
  MIN(value) AS min_value,
  MAX(value) AS max_value,
  STDDEV(value) AS std_value
FROM metrics
GROUP BY metric_name, tags, bucket;

The anomaly detection script now uses metrics_summary instead of metrics_by_time, reducing memory usage and enabling faster baseline calculation. However, this introduces a latency gap: anomalies are detected based on 15-minute summaries, but real-time data is available every minute. To bridge this, the system applies two-stage detection:

Fast detection: Every 1 minute, detect anomalies using the 15-minute summary data (with a 15-minute delay).
Refinement: Every 15 minutes, re-run detection using the raw metrics table, and merge the results into anomalies, ensuring no high-resolution signal is lost.

Alerting on Detected Anomalies

Once anomalies are stored, a separate system monitors for new entries and triggers alerts via HTTP. The alerting service runs every 30 seconds:

# /etc/crontab
*/30 * * * * /usr/local/bin/alerting-service --since "30 seconds ago"

The alerting service queries anomalies for entries not yet notified:

SELECT id, metric_name, tags, timestamp, value, z_score, severity
FROM anomalies
WHERE notified_at IS NULL
  AND detected_at > NOW() - INTERVAL '30 seconds'
ORDER BY detected_at DESC
LIMIT 100;

For each anomaly, it sends a JSON payload to an alerting endpoint:

POST /api/alerts
{
  "title": "High anomaly in http.request.latency.ms",
  "message": "Latency in api-gateway (us-west-2, v2.3.1) spiked to 217.4 ms (z=4.1)",
  "severity": "high",
  "details": {
    "metric": "http.request.latency.ms",
    "tags": {
      "service": "api-gateway",
      "region": "us-west-2",
      "version": "v2.3.1"
    },
    "value": 217.4,
    "z_score": 4.1,
    "timestamp": "2024-06-12T08:23:15.000Z"
  },
  "source": "anomaly-detection"
}

The alerting service uses curl to send notifications to Slack, PagerDuty, and email:

#!/bin/bash
# /usr/local/bin/alerting-service
set -euo pipefail

since=$(date -u -d "30 seconds ago" +"%Y-%m-%dT%H:%M:%SZ")

# Query for new anomalies
response=$(curl -s -H "Accept: application/json" \
  "http://metrics-db:5432/db/anomalies?since=$since&notified=false")

# Parse JSON and send alerts
echo "$response" | jq -c '.[]' | while read alert; do
  title=$(echo "$alert" | jq -r '.title')
  message=$(echo "$alert" | jq -r '.message')
  severity=$(echo "$alert" | jq -r '.severity')
  details=$(echo "$alert" | jq -c '.details')

  # Send to Slack
  curl -X POST -H "Content-Type: application/json" \
    -d '{
      "text": "'"$title"'",
      "attachments": [{
        "title": "Anomaly Details",
        "fields": [
          {"title": "Metric", "value": "'"$(echo "$details" | jq -r '.metric')"'", "short": true},
          {"title": "Tags", "value": "'"$(echo "$details" | jq -r '.tags | to_entries | map(.key + \": \". + .value) | join(\", \")')"'", "short": true},
          {"title": "Value", "value": "'"$(echo "$details" | jq -r '.value')"'", "short": true},
          {"title": "Z-Score", "value": "'"$(echo "$details" | jq -r '.z_score')"'", "short": true}
        ],
        "color": "'$(case "$severity" in high) echo "danger" ;; medium) echo "warning" ;; low) echo "info" ;; esac)'"
      }]
    }' \
    https://hooks.slack.com/services/ABC123/DEF456/GHI789

  # Send to PagerDuty
  curl -X POST -H "Content-Type: application/json" \
    -d '{
      "service_key": "abc123def456ghi789",
      "event_type": "trigger",
      "description": "'"$message"'",
      "details": {
        "metric": "'$(echo "$details" | jq -r '.metric')'",
        "tags": $(echo "$details" | jq -c '.tags'),
        "value": $(echo "$details" | jq -r '.value'),
        "z_score": $(echo "$details" | jq -r '.z_score')
      }
    }' \
    https://events.pagerduty.com/v2/enqueue

  # Mark as notified
  alert_id=$(echo "$alert" | jq -r '.id')
  psql -c "UPDATE anomalies SET notified_at = NOW() WHERE id = $alert_id"
done

Common Pitfalls and Silent Failures

Z-score threshold misalignment: A z_score > 3.0 is assumed to mean "3 standard deviations", but the standard deviation is computed over the last 100 points, not the entire series. A sudden shift in the data—say, a new version rollout—can cause the baseline to lag, so a 3-sigma spike appears only after 100 points, introducing a 10-minute delay.
Missing anomalies due to time zone drift: The detection job uses NOW() but the timestamp field is in UTC. If the database server is in a different time zone, and the detection script runs on a different host, a 1-hour lag in timestamp can cause anomalies to be missed or duplicated.
Inconsistent materialized view refresh: If metrics_by_time is refreshed every 5 minutes but the detection script runs every 1 minute, the first 4 minutes of data are missing from the baseline. This leads to false negatives in the first detection cycle.
Memory overflow: When the number of metric-tag groups exceeds 50,000, the Python script may fail due to OOM (out-of-memory), especially under high load. The error manifests silently: the script logs “MemoryError: Unable to allocate 100 MiB for buffer” but continues, missing several hours of anomaly detection.
Incomplete alerting: If the alerting service fails to update notified_at, anomalies are sent multiple times. A UNIQUE INDEX on anomalies(id) prevents duplicate alerts, but the notified_at column is not indexed, leading to slow UPDATE queries.

Scaling the Pipeline

To scale beyond a single server, the system uses a fan-out architecture: each detection job runs on a separate worker, with Kafka as the coordination point. The detect_anomalies.py script is containerized with Docker and deployed via kubectl:

# deploy/detection-worker.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: anomaly-detection-worker
spec:
  replicas: 10
  selector:
    matchLabels:
      app: anomaly-detection
  template:
    metadata:
      labels:
        app: anomaly-detection
    spec:
      containers:
      - name: detector
        image: metrics/anomaly-detector:latest
        env:
        - name: DATABASE_URL
          value: "postgresql://postgres:secret@metrics-db:5432/metrics"
        - name: DETECTION_INTERVAL
          value: "60"
        - name: LOG_LEVEL
          value: "INFO"
        ports:
        - containerPort: 8080

Workers consume from a Kafka topic anomalies.detected, and each writes a unique worker_id to a worker_status table. The system tracks progress and handles worker failures by checking for missing worker_id entries.

The entire pipeline is monitored via systemctl and journalctl:

# Check worker status
journalctl -u anomaly-detection-worker --since "1 hour ago" | grep -E "ERROR|WARNING"

# Check for missed detection cycles
psql -c "
SELECT
  COUNT(*) AS missing_jobs,
  MIN(timestamp) AS earliest_missing
FROM (
  SELECT
    'detect_anomalies.py' AS job,
    (EXTRACT(EPOCH FROM NOW()) / 60)::int AS expected_cycle,
    (EXTRACT(EPOCH FROM detected_at) / 60)::int AS actual_cycle
  FROM anomalies
  WHERE detected_at > NOW() - INTERVAL '1 hour'
  GROUP BY actual_cycle
  HAVING COUNT(*) < 10
) t
WHERE expected_cycle != actual_cycle"

This architecture enables anomaly detection that is not only accurate and timely, but resilient, observable, and self-healing—ready for the demands of a modern, distributed system.

This page was rewritten on 17 October 2026. It replaced a templated version that was not specific to its own title. How these pages are written and checked. If anything here is wrong, tell us at [email protected] and we will correct it.