AI‑Powered Log Anomaly Detection & Monitoring

In this article, we will build an AI‑powered log anomaly detection and monitoring pipeline using Flask, Prometheus, Grafana, and IsolationForest from scikit‑learn. This setup allows us to expose metrics from an API, detect anomalies in log data, and visualize them in Grafana dashboards for real‑time observability.

Step #1: System Preparation

Update system packages.

sudo apt update && sudo apt upgrade -y

Install essential tools for compiling, downloading, and managing software.

sudo apt install -y build-essential wget curl git unzip software-properties-common

Add Python 3.12 repository.

sudo add-apt-repository ppa:deadsnakes/ppa -y

Install Python 3.12, virtual environment support, and development headers.

sudo apt install -y python3.12 python3.12-venv python3.12-dev

Verifie the installation.

python3.12 --version

Step #2: Create Project Structure

Create a working directory log-anomaly-monitor and navigate to it.

mkdir ~/log-anomaly-monitor
cd ~/log-anomaly-monitor

Add subfolders for API code, anomaly detector, and logs.

mkdir api anomaly_detector logs

Create an isolated Python environment. Then activate it so dependencies don’t conflict with system Python.

python3.12 -m venv venv
source venv/bin/activate

Then Install Dependencies.

pip install flask prometheus_client scikit-learn numpy pandas requests

• Flask → API framework.
• prometheus_client → Expose metrics.
• scikit-learn → IsolationForest anomaly detection.
• numpy/pandas → Data handling.
• requests → Simulate traffic.

Step #3: Create Flask API

Create a app.py in api directory.

nano api/app.py

add the following content in it.

from flask import Flask, request, jsonify
from prometheus_client import Counter, Histogram, Gauge, generate_latest
import numpy as np
from anomaly_detector.detector import LogAnomalyDetector

app = Flask(__name__)

# Prometheus metrics
REQUEST_COUNT = Counter("api_requests_total", "Total API Requests", ["endpoint"])
LATENCY = Histogram("api_latency_seconds", "Request latency", ["endpoint"])
ANOMALY_SCORE = Gauge("api_anomaly_score", "Latest anomaly score")

# Initialize detector with baseline data
detector = LogAnomalyDetector()
baseline = np.random.normal(0, 1, (100, 3))
detector.fit(baseline)

@app.route("/predict", methods=["POST"])
def predict():
    REQUEST_COUNT.labels(endpoint="/predict").inc()
    with LATENCY.labels(endpoint="/predict").time():
        data = request.json
        values = np.array(data.get("values", [])).reshape(1, -1)
        prediction, scores = detector.predict(values)
        score = scores[0]
        ANOMALY_SCORE.set(score)
        return jsonify({"prediction": int(prediction[0]), "anomaly_score": float(score)})

@app.route("/metrics")
def metrics():
    return generate_latest()

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

Explanation:

• Defines Flask API with /predict and /metrics.
• REQUEST_COUNT → counts API calls.
• LATENCY → measures request latency.
• ANOMALY_SCORE → exposes anomaly detection score.
• Uses IsolationForest to detect anomalies.
• /metrics endpoint exposes Prometheus metrics.

Create a detector.py.

nano anomaly_detector/detector.py

add the following content in it.

import numpy as np
from sklearn.ensemble import IsolationForest

class LogAnomalyDetector:
    def __init__(self):
        self.model = IsolationForest(n_estimators=100, contamination=0.1, random_state=42)

    def fit(self, data):
        self.model.fit(data)

    def predict(self, data):
        return self.model.predict(data), self.model.decision_function(data)

Explanation (points):

• Wraps IsolationForest in a class.
• fit() → trains model on baseline data.
• predict() → returns anomaly prediction (1 normal, -1 anomaly) and score.

Run the API. Run the API from project root so imports work.

python -m api.app

Starts Flask on port 5000.

Test the api. Send requests to /predict and check metrics.

curl -X POST http://:5000/predict \
-H "Content-Type: application/json" \
-d '{"values":[0.5,-1.2,3.3]}'

• Sends test data to /predict.
• Returns anomaly score and prediction.

curl http://:5000/metrics | grep api_anomaly_score

Step #4: Install Prometheus

In another tab go to project directory and start python env. Then install prometheus.

cd ~/log-anomaly-monitor
source venv/bin/activate
wget https://github.com/prometheus/prometheus/releases/download/v2.55.0/prometheus-2.55.0.linux-amd64.tar.gz

extract the binary file.

tar xvf prometheus-2.55.0.linux-amd64.tar.gz

Move it to prometheus directory.

mv prometheus-2.55.0.linux-amd64 prometheus

Edit config file.

nano prometheus/prometheus.yml

add following code under the scrap config.

  - job_name: 'flask-api'
    static_configs:
      - targets: ['localhost:5000']

Run Prometheus.

cd prometheus
./prometheus --config.file=prometheus.yml

Access: http://<EC2-IP>:9090

Go to status > Targets. You will see api is running and up. Prometheus is scraping metrics from it.

Step #5: Install Grafana

Go to project directory. Download the Grafana.

cd ~/log-anomaly-monitor
wget https://dl.grafana.com/oss/release/grafana-11.2.0.linux-amd64.tar.gz

Extract the binary.

tar -zxvf grafana-11.2.0.linux-amd64.tar.gz

Move it to grafana.

mv grafana-v11.2.0 grafana

Go to grafana/bin directory.

cd grafana/bin

Start the Grafana server.

./grafana-server

Access Grafana: http://<EC2-IP>:3000 (default login: admin/admin).

Step #6: Visualize Data in Grafana

Go to Connections > Data sources.

Click on Add data source.

Select Prometheus as a data source.

Enter the prometheus url here. http://<EC2-IP>:9090

Click on Save and test. You will see th Successfully quaried the Prometheus Api message. Now go to + icon at top-right corner and click on New dashboard.

Click on Add visualization.

Select prometheus as a data source.

Create Grafana Dashboard. create panels with following quaries.

• api_requests_total → request count.
  • Counts every request made to the /predict endpoint.
  • Useful for tracking API usage and traffic patterns.
  • Visualization: Time series or Stat panel to show request volume.
• api_latency_seconds_sum → latency trends.
  • api_latency_seconds_sum → total accumulated latency.
  • api_latency_seconds_count → number of requests.
  • Dividing them gives average latency per request.
  • Visualization: Time series to see latency trends over time.
• api_anomaly_score → anomaly detection gauge/time series.
  • Shows the latest anomaly score emitted by the IsolationForest model.
  • Normal requests → scores closer to 0 or positive.
  • Anomalous requests → negative scores, flagged as outliers.
  • Visualization: Gauge (to show current anomaly score) or Time series (to see anomaly spikes).

Combined Overview. You can also create a multi‑panel dashboard:

• Top row → Request count and latency.
• Bottom row → Anomaly score gauge + anomaly score time series.

This layout gives both performance metrics and anomaly detection insights at a glance.

Why This Dashboard Matters

• Developers → See if API is healthy and performant.
• Ops teams → Detect anomalies before they impact users.
• Data scientists → Validate anomaly detection models in production.

It’s a single pane of glass for monitoring both system health and AI‑driven anomaly detection.

Conclusion:

In conclusion, this project demonstrates how combining Flask, Prometheus, Grafana, and IsolationForest creates a powerful end‑to‑end pipeline for anomaly detection and monitoring. With the API exposing metrics, Prometheus scraping them, and Grafana visualizing both performance and anomaly scores, you gain real‑time observability into system health and unusual behavior. This setup is lightweight, extensible, and production‑ready, making it an effective foundation for modern cloud monitoring solutions.

Related Articles:

Building and Testing Go Project on GitHub Actions

Reference:

Grafana Official Page