AI-Driven Automated Network Monitoring & Anomaly Detection — Part 5: Integrating the AI Model with Prometheus Alerts

AI‑Driven Automated Network Monitoring & Anomaly Detection — Part 5: Integrating the AI Model with Prometheus Alerts

In the previous installments we built a data‑ingestion pipeline that pulls raw network telemetry from a Kafka cluster, trained a PyTorch‑based anomaly detector on historical traffic, and exposed the model as a lightweight REST API behind a FastAPI gateway. We also discussed how to run the model in a GPU‑enabled container and how to scale the inference service horizontally with a Kubernetes deployment.

Today we bring everything together: we turn the model’s predictions into actionable Prometheus alerts. By exposing the anomaly scores as metrics and wiring them into Prometheus’ rule engine, we can trigger alerts that are not based on static thresholds but on learned patterns, thereby catching issues that traditional alerting would miss.

Why Prometheus‑Based Alerts Make Sense for AI‑Driven Anomaly Detection

Prometheus is the de‑facto standard for time‑series metrics in modern cloud‑native environments. Its pull‑based model, expressive query language (PromQL), and native integration with Alertmanager make it ideal for orchestrating a continuous‑monitoring pipeline. When combined with an AI model that can score every data point, we can create a “smart” alerting layer that adapts to traffic fluctuations, seasonality, and evolving network conditions.

In this section I’ll walk you through:

• Building a Prometheus exporter that forwards anomaly scores to the Prometheus server.
• Configuring Prometheus scrape targets and alerting rules that fire on high‑score events.
• Hooking into Alertmanager to enrich alerts with model‑generated context.
• Using OpenObserve to pull custom SQL queries and push anomaly streams into Prometheus.
• Performance tuning tips for low‑latency, high‑throughput deployments.

All code samples are ready to copy‑paste. Feel free to tweak them to your own environment.

1. The Prometheus Exporter: Exposing Anomaly Scores

We’ll expose a single gauge metric: network_anomaly_score. The gauge will carry a value between 0 and 1 for each metric series (e.g., router_cpu_usage{instance="router1"}). The exporter will query our FastAPI inference service once every scrape interval (default 15 s) and push the latest scores.

1.1. Python Exporter Skeleton

Below is a minimal implementation using prometheus_client and httpx for async HTTP requests. The exporter is designed to run as a Docker container behind the same Kubernetes service that hosts the inference API.

#!/usr/bin/env python3
# exporter.py

import asyncio
import os
import json
import logging
from typing import Dict

import httpx
from prometheus_client import start_http_server, Gauge

logging.basicConfig(level=logging.INFO)
log = logging.getLogger(__name__)

# Configuration
INFERENCE_URL = os.getenv("INFERENCE_URL", "http://inference-api:8000/predict")
PROMETHEUS_PORT = int(os.getenv("PROMETHEUS_PORT", 8001))
SCRAPE_INTERVAL = int(os.getenv("SCRAPE_INTERVAL", 15))

# Exported metric
anomaly_gauge = Gauge(
    "network_anomaly_score",
    "AI‑driven anomaly score for a metric series",
    ["instance", "metric"]
)

async def fetch_scores(client: httpx.AsyncClient) -> Dict:
    """
    Send a batch of metric series to the inference API and parse the JSON response.
    Expected payload:
        {"series": [{"instance": "router1", "metric": "cpu_usage", "values": [0.12, 0.15, ...]}]}
    Response:
        {"scores": [{"instance": "router1", "metric": "cpu_usage", "score": 0.87}, ...]}
    """
    payload = {
        "series": [
            # In a real deployment this would be populated from Prometheus via a remote read
            {"instance": "router1", "metric": "cpu_usage", "values": [0.12, 0.15, 0.14]},
            {"instance": "router2", "metric": "cpu_usage", "values": [0.08, 0.09, 0.07]},
        ]
    }
    resp = await client.post(INFERENCE_URL, json=payload)
    resp.raise_for_status()
    return resp.json()

async def update_metrics():
    async with httpx.AsyncClient() as client:
        while True:
            try:
                result = await fetch_scores(client)
                for item in result.get("scores", []):
                    anomaly_gauge.labels(
                        instance=item["instance"],
                        metric=item["metric"]
                    ).set(item["score"])
                log.info("Updated anomaly scores")
            except Exception as exc:
                log.exception("Failed to fetch scores: %s", exc)
            await asyncio.sleep(SCRAPE_INTERVAL)

if __name__ == "__main__":
    start_http_server(PROMETHEUS_PORT)
    asyncio.run(update_metrics())

Key points:

• The exporter runs a single HTTP server on port 8001, which Prometheus will scrape.
• We use httpx.AsyncClient to keep the event loop non‑blocking even when the inference service is slow.
• The network_anomaly_score gauge is labelled by instance and metric, allowing fine‑grained alerting.
• In a production environment you would pull metric series directly from Prometheus using its remote_read API or from a local cache; here we use a static payload for illustration.

1.2. Dockerfile & Deployment

Below is a lean Dockerfile that uses the official python:3.11-slim image. The container is intended to run as a sidecar in the same pod as the inference API for low network latency.

# Dockerfile
FROM python:3.11-slim

ENV PYTHONDONTWRITEBYTECODE 1
ENV PYTHONUNBUFFERED 1

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY exporter.py .
CMD ["python", "exporter.py"]

requirements.txt:

prometheus-client==0.20.0
httpx==0.27.0

When deploying on Kubernetes, add a sidecar container to the inference pod and expose the exporter’s port. For example, in your Deployment spec:

spec:
  containers:
  - name: inference-api
    image: registry.company.com/inference-api:latest
    ports:
    - containerPort: 8000
  - name: anomaly-exporter
    image: registry.company.com/anomaly-exporter:latest
    ports:
    - containerPort: 8001

Prometheus will then scrape the exporter via a Service that selects the anomaly-exporter container.

2. Prometheus Scrape Configuration

Define a service and scrape job that targets the exporter. Add the following to your prometheus.yml:

scrape_configs:
- job_name: 'anomaly_exporter'
  static_configs:
  - targets: ['anomaly-exporter-service:8001']

Ensure the anomaly-exporter-service is a ClusterIP service that maps to the exporter container port. If you’re running locally, replace the target with the pod’s IP or localhost:8001.

3. Alerting Rules: Turning Scores into Alerts

Now that we have a time‑series of anomaly scores, we can write PromQL rules that trigger when the score crosses a threshold. The threshold is not a hard‑coded number; you can calibrate it per metric or per instance based on historical distribution.

3.1. Sample Rule File

groups:
- name: network_anomaly
  rules:
  - alert: HighAnomalyScore
    expr: network_anomaly_score{metric="cpu_usage"} > 0.75
    for: 1m
    labels:
      severity: warning
    annotations:
      summary: "Anomaly detected on {{ $labels.instance }} - CPU usage"
      description: |
        The AI model has assigned a score of {{ $value }} to the
        cpu_usage metric on {{ $labels.instance }}. Investigate
        possible hardware or software issues.
      runbook: https://company.com/runbooks/ai-anomaly

Explanation:

• The expr filters on metric="cpu_usage" but you can keep it generic if you want alerts for any metric that exceeds 0.75.
• The for: 1m clause ensures that the condition is stable for at least a minute before firing.
• Annotations can include links to runbooks, dashboards, or even the raw anomaly payload if you expose it via the exporter.

3.2. Dynamic Thresholds with record Rules

Sometimes a fixed threshold is too coarse. You can compute a moving average or percentile and use that as a dynamic threshold. For example:

- record: anomaly_score_avg
  expr: avg_over_time(network_anomaly_score[5m])
- alert: HighAnomalyScoreDynamic
  expr: network_anomaly_score > anomaly_score_avg * 1.5
  for: 30s
  labels:
    severity: critical

Here the rule fires when the current score is 1.5 times higher than the 5‑minute average, adapting to recent traffic patterns.

4. Enriching Alerts with Alertmanager Webhook

Prometheus alerting is declarative; however, when an alert fires we may want to augment the message with richer context from the AI model—such as the raw input values, confidence intervals, or even a short explanation. Alertmanager supports webhook receivers that can receive the alert payload, process it, and forward enriched alerts to downstream systems (Slack, PagerDuty, email, etc.).

4.1. Flask Webhook Receiver

The webhook receives a JSON payload from Alertmanager. We’ll parse the alert, query the inference API for the full series, and then push the enriched data to a Slack channel.

# webhook.py
import os
import json
import httpx
import logging
from flask import Flask, request, jsonify

app = Flask(__name__)
log = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)

INFERENCE_URL = os.getenv("INFERENCE_URL", "http://inference-api:8000/predict_full")
SLACK_WEBHOOK = os.getenv("SLACK_WEBHOOK")

@app.route("/alert", methods=["POST"])
def alert_handler():
    payload = request.json
    alerts = payload.get("alerts", [])
    enriched = []

    for alert in alerts:
        labels = alert.get("labels", {})
        instance = labels.get("instance")
        metric = labels.get("metric")
        # Retrieve full series from inference API
        series_payload = {"instance": instance, "metric": metric}
        try:
            resp = httpx.post(INFERENCE_URL, json=series_payload)
            resp.raise_for_status()
            series_data = resp.json()
            explanation = series_data.get("explanation", "No explanation available")
        except Exception as exc:
            log.exception("Failed to fetch series: %s", exc)
            explanation = str(exc)

        enriched.append({
            "original": alert,
            "explanation": explanation
        })

    # Push to Slack
    if SLACK_WEBHOOK:
        for item in enriched:
            message = {
                "text": f"*{item['original']['labels']['alertname']}*\n"
                        f"Instance: {item['original']['labels']['instance']}\n"
                        f"Score: {item['original']['status']}\n"
                        f"Explanation: {item['explanation']}"
            }
            httpx.post(SLACK_WEBHOOK, json=message)

    return jsonify({"status": "processed", "count": len(enriched)})

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

Key points:

• The webhook calls /predict_full on the inference API, which should return the raw input values, the score, and a brief explanation (e.g., “Spike in outbound traffic due to a DDoS attack”).
• We then post a formatted message to Slack. You can replace this with any other notification channel.

4.2. Alertmanager Configuration

Add the webhook to Alertmanager’s receivers section and reference it in the route for the network_anomaly group:

receivers:
- name: 'slack-notifications'
  webhook_configs:
  - url: 'http://alert-webhook-service:5000/alert'
    send_resolved: true

route:
  group_by: ['alertname', 'instance']
  receiver: 'slack-notifications'
  routes:
  - match:
      alertname: 'HighAnomalyScore'
    receiver: 'slack-notifications'

Make sure the webhook service is reachable from Alertmanager; in Kubernetes, expose it via a ClusterIP service.

5. Pulling Anomalies from OpenObserve into Prometheus

OpenObserve provides a powerful _anomalies stream that contains raw anomaly detections produced by its internal ML models. By querying this stream with custom SQL, you can push those results into Prometheus as metrics.

5.1. Example Query

Suppose you want to fetch the top 10 anomalies for the past hour:

SELECT
  instance,
  metric,
  score,
  anomaly_timestamp
FROM _anomalies
WHERE anomaly_timestamp >= now() - interval '1 hour'
ORDER BY score DESC
LIMIT 10;

Execute this query via OpenObserve’s REST API or CLI and parse the JSON response.

5.2. Pushing to Prometheus with remote_write

Prometheus supports remote_write to send metrics to external systems. OpenObserve can act as a remote write target if you configure it to expose a /api/v1/write endpoint. The exporter can then forward the anomalies as InstantVector samples.

remote_write:
- url: https://openobserve.company.com/api/v1/write
  headers:
    Authorization: Bearer <YOUR_TOKEN>

In the exporter, after fetching anomalies from OpenObserve, you can write them to Prometheus via the pushgateway or by directly calling remote_write with the remote_write client library. For simplicity, here we use the pushgateway:

from prometheus_client import CollectorRegistry, Gauge, push_to_gateway

registry = CollectorRegistry()
gauge = Gauge('openobserve_anomaly_score', 'Score from OpenObserve', ['instance', 'metric'], registry=registry)

# Suppose anomalies is a list of dicts from OpenObserve
for anomaly in anomalies:
    gauge.labels(
        instance=anomaly["instance"],
        metric=anomaly["metric"]
    ).set(anomaly["score"])

push_to_gateway('pushgateway.company.com:9091', job='openobserve_anomalies', registry=registry)

This approach keeps the exporter stateless and allows Prometheus to scrape the pushgateway for the latest values.

6. Performance & Reliability Considerations

When integrating AI inference with Prometheus, latency and throughput become critical. Below are some best practices:

Batch Requests:

---

Originally published at https://artificial-inteligence.phptutorial.co.in