AI-Driven Automated Network Monitoring & Anomaly Detection — Part 7: Deploying the End-to-End Pipeline with Docker Compose and CI/CD

⏱ 8 min read  |  ~1617 words

AI‑Driven Automated Network Monitoring & Anomaly Detection — Part 7: Deploying the End‑to‑End Pipeline with Docker Compose and CI/CD

Based on my technical understanding as a Lead Programmer Analyst (PHP, Perl, Python, Shell) and on the latest industry chatter from 2026, this article walks you through the final piece of the puzzle: turning the micro‑service ML/NLP pipeline we designed in earlier parts into a production‑ready, reproducible stack using Docker Compose, GitHub Actions, and Jenkins.

Quick recap (Parts 1‑6): We started by defining the problem space (network telemetry, packet‑level logs, and API‑traffic anomalies), then built a data‑ingestion service, a preprocessing container, a PyTorch‑based anomaly detector, an NLP risk‑assessment model, and finally a merging/alerting micro‑service. Parts 5‑6 introduced OpenSearch‑based observability and a lightweight retraining loop.

Why Docker Compose Still Matters in 2026

Even though the hype in 2026 leans heavily toward Kubernetes‑native AI agents (Claude 4.6 Opus & GPT‑5.4 Pro parallel agents), most organizations—especially those just stepping into AI‑driven DevOps—still need a low‑friction way to spin up the entire stack on a single host or a modest VM. Docker Compose gives you:

• Deterministic networking: each service gets an alias (e.g., preprocess, anomaly), making inter‑service calls trivial.
• Single‑source of truth for environment variables: the .env file lives alongside docker‑compose.yml, mirroring the way CI pipelines inject secrets.
• Fast feedback loops: docker compose up --build rebuilds only what changed—perfect for iterative model tuning.

In the real‑world example from the MDPI paper “Context‑Aware ML/NLP Pipeline for Real‑Time Anomaly Detection and Risk Assessment in Cloud API Traffic” the authors used Docker Compose for exactly this purpose (see the reference section). We’ll adopt the same structure, but we’ll also layer a CI/CD overlay that works for both GitHub Actions (the de‑facto SaaS CI) and Jenkins (the on‑premise workhorse many enterprises still run).

Folder Layout – The Blueprint

network‑monitor/
│
├─ .env                     # Global env vars (secrets are injected in CI)
├─ docker-compose.yml       # Orchestrates all services
│
├─ ingestion/
│   ├─ Dockerfile
│   └─ app.py
│
├─ preprocess/
│   ├─ Dockerfile
│   └─ preprocess.py
│
├─ anomaly/
│   ├─ Dockerfile
│   └─ model.py
│
├─ nlp/
│   ├─ Dockerfile
│   └─ risk_assess.py
│
├─ aggregator/
│   ├─ Dockerfile
│   └─ aggregator.py
│
├─ alert/
│   ├─ Dockerfile
│   └─ alert.py
│
├─ retrain/
│   ├─ Dockerfile
│   └─ retrain.py
│
└─ ci/
    ├─ github/
    │   └─ workflow.yml
    └─ jenkins/
        └─ Jenkinsfile

1️⃣ Dockerfiles – One per Service

Below are the minimal, production‑ready Dockerfiles. They share a base image (python:3.11‑slim) to keep layers consistent and reduce image bloat. Feel free to swap the base for python:3.11‑bookworm if you need extra OS packages.

Ingestion Service

# ingestion/Dockerfile
FROM python:3.11-slim

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

COPY app.py .
ENV PYTHONUNBUFFERED=1
CMD ["python", "app.py"]

Preprocess Service

# preprocess/Dockerfile
FROM python:3.11-slim

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

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

Anomaly Detection Service (PyTorch)

# anomaly/Dockerfile
FROM python:3.11-slim

# Install system deps for torch (Ubuntu libs)
RUN apt-get update && apt-get install -y --no-install-recommends \
    libglib2.0-0 libgl1 && rm -rf /var/lib/apt/lists/*

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

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

NLP Risk‑Assessment Service (HuggingFace Transformers)

# nlp/Dockerfile
FROM python:3.11-slim

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

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

Aggregator (Merging Anomaly + NLP Scores)

# aggregator/Dockerfile
FROM python:3.11-slim

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

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

Alert Service (OpenSearch Integration)

# alert/Dockerfile
FROM python:3.11-slim

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

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

Retraining Service (Scheduled Trigger)

# retrain/Dockerfile
FROM python:3.11-slim

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

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

2️⃣ docker‑compose.yml – The Heartbeat

The following Compose file wires every micro‑service together, mounts a shared volume for raw logs, and adds an OpenSearch container for observability (mirroring the MDPI paper’s “OpenSearch‑Da” instrumentation).

# docker-compose.yml
version: "3.9"

services:
  # 1️⃣ Ingestion – pulls network telemetry from Kafka or syslog
  ingestion:
    build: ./ingestion
    container_name: ingestion
    env_file: .env
    volumes:
      - logs:/data/raw
    depends_on:
      - preprocess

  # 2️⃣ Preprocess – normalizes, filters, and enriches data
  preprocess:
    build: ./preprocess
    container_name: preprocess
    env_file: .env
    volumes:
      - logs:/data/raw
      - processed:/data/processed
    depends_on:
      - anomaly

  # 3️⃣ Anomaly – PyTorch model that outputs a score
  anomaly:
    build: ./anomaly
    container_name: anomaly
    env_file: .env
    volumes:
      - processed:/data/processed
      - anomalies:/data/anomalies
    depends_on:
      - nlp

  # 4️⃣ NLP – risk classification using a transformer
  nlp:
    build: ./nlp
    container_name: nlp
    env_file: .env
    volumes:
      - processed:/data/processed
      - nlp_out:/data/nlp
    depends_on:
      - aggregator

  # 5️⃣ Aggregator – merges scores, decides severity
  aggregator:
    build: ./aggregator
    container_name: aggregator
    env_file: .env
    volumes:
      - anomalies:/data/anomalies
      - nlp_out:/data/nlp
      - alerts:/data/alerts
    depends_on:
      - alert

  # 6️⃣ Alert – pushes alerts to OpenSearch & Slack
  alert:
    build: ./alert
    container_name: alert
    env_file: .env
    volumes:
      - alerts:/data/alerts
    depends_on:
      - opensearch

  # 7️⃣ Retraining – nightly job that pulls labeled data, retrains, and pushes new model
  retrain:
    build: ./retrain
    container_name: retrain
    env_file: .env
    volumes:
      - model:/model
    deploy:
      restart_policy:
        condition: none   # run‑once per schedule, see CI below

  # 8️⃣ OpenSearch – observability backbone
  opensearch:
    image: opensearchproject/opensearch:2.12.0
    container_name: opensearch
    environment:
      - discovery.type=single-node
      - OPENSEARCH_JAVA_OPTS=-Xms512m -Xmx512m
      - plugins.security.disabled=true
    ulimits:
      memlock:
        soft: -1
        hard: -1
    ports:
      - "9200:9200"
      - "9600:9600"
    volumes:
      - opensearch-data:/usr/share/opensearch/data

volumes:
  logs:
  processed:
  anomalies:
  nlp_out:
  alerts:
  model:
  opensearch-data:

3️⃣ Service Code Sketches – Keeping Them Minimal but Functional

Below are trimmed‑down but fully runnable snippets. In a real project you would add proper logging, error handling, and schema validation.

ingestion/app.py

import os, time, json
import paho.mqtt.client as mqtt   # Example source, replace with Kafka if needed

BROKER = os.getenv("MQTT_BROKER", "mqtt://localhost")
TOPIC  = os.getenv("MQTT_TOPIC", "network/telemetry")
OUTDIR = "/data/raw"

def on_message(client, userdata, msg):
    payload = msg.payload.decode()
    ts = int(time.time())
    fname = f"{OUTDIR}/{ts}.json"
    with open(fname, "w") as f:
        json.dump({"topic": msg.topic, "payload": payload}, f)

client = mqtt.Client()
client.on_message = on_message
client.connect(BROKER)
client.subscribe(TOPIC)
client.loop_forever()

preprocess/preprocess.py

import os, json, glob
import pandas as pd

INDIR  = "/data/raw"
OUTDIR = "/data/processed"

def normalize(record):
    # Dummy normalization – real code would parse PCAP, extract fields, etc.
    data = json.loads(record)
    return {"ts": os.path.splitext(os.path.basename(record))[0],
            "source_ip": data.get("payload", "").split()[0],
            "bytes": len(data["payload"])}

def main():
    files = glob.glob(f"{INDIR}/*.json")
    rows = [normalize(f) for f in files]
    df = pd.DataFrame(rows)
    df.to_parquet(f"{OUTDIR}/batch.parquet", index=False)

if __name__ == "__main__":
    main()

anomaly/model.py

import os, torch, pandas as pd
from torch import nn

MODEL_PATH = "/model/anomaly.pt"
INFILE = "/data/processed/batch.parquet"
OUTDIR = "/data/anomalies"

class SimpleAE(nn.Module):
    def __init__(self, dim):
        super().__init__()
        self.encoder = nn.Linear(dim, dim//2)
        self.decoder = nn.Linear(dim//2, dim)

    def forward(self, x):
        z = torch.relu(self.encoder(x))
        return torch.relu(self.decoder(z))

def load_model():
    dim = 10   # placeholder
    model = SimpleAE(dim)
    if os.path.exists(MODEL_PATH):
        model.load_state_dict(torch.load(MODEL_PATH))
    model.eval()
    return model

def infer():
    df = pd.read_parquet(INFILE)
    model = load_model()
    tensor = torch.tensor(df.values, dtype=torch.float32)
    recon = model(tensor)
    error = torch.mean((tensor - recon) ** 2, dim=1)
    df["anomaly_score"] = error.detach().numpy()
    df.to_csv(f"{OUTDIR}/scores.csv", index=False)

if __name__ == "__main__":
    infer()

nlp/risk_assess.py

import os, json, pandas as pd
from transformers import pipeline

MODEL_NAME = os.getenv("HF_MODEL", "distilbert-base-uncased-finetuned-sst-2-english")
INFILE = "/data/processed/batch.parquet"
OUTDIR = "/data/nlp"

sentiment = pipeline("sentiment-analysis", model=MODEL_NAME)

def classify(text):
    result = sentiment(text[:512])[0]   # truncate for speed
    return result["label"], result["score"]

def main():
    df = pd.read_parquet(INFILE)
    # Pretend each row has a 'payload' column with raw log text
    df["risk_label"], df["risk_score"] = zip(*df["payload"].apply(classify))
    df.to_csv(f"{OUTDIR}/risk.csv", index=False)

if __name__ == "__main__":
    main()

aggregator/aggregator.py

import os, pandas as pd

ANOM = "/data/anomalies/scores.csv"
NLP  = "/data/nlp/risk.csv"
OUT  = "/data/alerts/merged.csv"

def main():
    a = pd.read_csv(ANOM)
    n = pd.read_csv(NLP)
    merged = pd.merge(a, n, left_on="ts", right_on="ts")
    # Simple rule: if anomaly_score > 0.5 and risk_score > 0.7 → high severity
    merged["severity"] = merged.apply(
        lambda r: "high" if r.anomaly_score > 0.5 and r.risk_score > 0.7 else "medium",
        axis=1
    )
    merged.to_csv(OUT, index=False)

if __name__ == "__main__":
    main()

alert/alert.py

import os, json, pandas as pd, requests

OPENSEARCH_URL = os.getenv("OPENSEARCH_URL", "http://opensearch:9200")
INDEX = "network-anomalies"
SLACK_WEBHOOK = os.getenv("SLACK_WEBHOOK")

def push_to_opensearch(df):
    bulk = ""
    for _, row in df.iterrows():
        action = {"index": {"_index": INDEX}}
        bulk += json.dumps(action) + "\n"
        bulk += json.dumps(row.to_dict()) + "\n"
    headers = {"Content-Type": "application/x-ndjson"}
    resp = requests.post(f"{OPENSEARCH_URL}/_bulk", data=bulk, headers=headers)
    resp.raise_for_status()

def slack_notify(df):
    high = df[df.severity == "high"]
    if high.empty:
        return
    text = f"*⚠️ High‑severity anomalies detected:*\\n"
    for _, r in high.iterrows():
        text += f"`{r.ts}` – score:{r.anomaly_score:.2f}, risk:{r.risk_label}\\n"
    payload = {"text": text}
    requests.post(SLACK_WEBHOOK, json=payload)

def main():
    df = pd.read_csv("/data/alerts/merged.csv")
    push_to_opensearch(df)
    slack_notify(df)

if __name__ == "__main__":
    main()

retrain/retrain.py

import os, torch, pandas as pd
from torch import nn

DATASET = "/data/labeled/training.parquet"
MODEL_PATH = "/model/anomaly.pt"

class SimpleAE(nn.Module):
    def __init__(self, dim):
        super().__init__()
        self.encoder = nn.Linear(dim, dim//2)
        self.decoder = nn.Linear(dim//2, dim)

    def forward(self, x):
        z = torch.relu(self.encoder(x))
        return torch.relu(self.decoder(z))

def train():
    df = pd.read_parquet(DATASET)
    X = torch.tensor(df.values, dtype=torch.float32)
    model = SimpleAE(X.shape[1])
    optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)
    loss_fn = nn.MSELoss()

    for epoch in range(5):
        optimizer.zero_grad()
        recon = model(X)
        loss = loss_fn(recon, X)
        loss.backward()
        optimizer.step()
        print(f"Epoch {epoch+1} – loss: {loss.item():.4f}")

    torch.save(model.state_dict(), MODEL_PATH)
    print("✅ New model saved")

if __name__ == "__main__":
    train()

4️⃣ CI/CD – GitHub Actions + Jenkins

The pipeline has three stages:

1. Build & Test: Each Dockerfile is lint‑checked (Hadolint) and built; unit tests run inside the image.
2. Integration Test: Spin up the entire docker‑compose.yml stack on a runner, feed synthetic traffic, assert that alerts appear in OpenSearch.
3. Deploy: Push images to a private registry, then run docker compose pull && docker compose up -d on the target host.

GitHub Actions – .github/workflows/ci.yml

# ci/github/workflow.yml

name: CI / CD Pipeline

on:

  push:

    branches: [ main ]

  pull_request:

    branches: [ main ]
env:

  REGISTRY: ghcr.io/${{ github.repository_owner }}/network-monitor
jobs:

  build-test:

    runs-on: ubuntu-l

🔗 You Might Also Like

AI Safety & Ethics: What’s New in April 2026
AI Agents: What’s New in April 2026
Prompt Engineering: What’s New in April 2026




✍️ About the Author
Vijay Vinoth — Lead Programmer Analyst with expertise in PHP, Perl, Python, and Shell scripting. Passionate about AI, automation, and building scalable systems. Writing to share practical insights from real-world engineering experience.



        Note: This technical analysis reflects my independent understanding as a Lead Programmer Analyst as of April 2026.

        As AI ecosystems like Claude 4.6 Opus evolve, actual implementation may vary. Refer to official documentation for final specs.