How I added identity, fine-grained authorization, image scanning and self-healing to an event-driven ML pipeline running on AWS EC2 — without slowing it down.*
Most "real-time AIOps" demos stop at the happy path: a producer pushes events, a model scores them, a dashboard lights up. The part nobody shows you is the boring, important stuff — who is allowed to call what, whether the containers you shipped are full of CVEs, and what happens at 3 a.m. when a service falls over.
I built an event-driven pipeline that ingests a live data stream, scores every event with a machine-learning consumer, and persists the results — and then I spent most of my effort making it secure and self-healing. This post walks through how, with the patterns and code you can reuse.
Here's the shape of the system:
Producer ──▶ Kafka topic ──▶ ML consumer ──▶ PostgreSQL ──▶ Dashboard (FastAPI) │ ▲ └────────── auto-heal topic ◀── health watcher ┘ Six containers on an isolated Docker bridge network: ZooKeeper, Kafka, PostgreSQL, Keycloak, the FastAPI API/dashboard, and the ML worker. It processes a batch of records every few seconds with sub-second end-to-end latency and high delivery reliability across multiple partitions. But the interesting engineering is in the three layers I wrapped around it.
1. Identity at the edge: Keycloak OIDC
The first rule of a data pipeline that touches real decisions: no anonymous calls. Every request to the API has to carry a valid, verifiable token.
I ran Keycloak as the identity provider (one of the six containers) and issued OIDC/JWT tokens. The FastAPI gateway validates the token on every request — signature, issuer, audience and expiry — before any business logic runs.
# auth.py — validate Keycloak-issued JWTs on every request from fastapi import Depends, HTTPException, status from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials import jwt # PyJWT from jwt import PyJWKClient JWKS_URL = "http://keycloak:8080/realms/aiops/protocol/openid-connect/certs" ISSUER = "http://keycloak:8080/realms/aiops" AUDIENCE = "aiops-api" _jwk_client = PyJWKClient(JWKS_URL) bearer = HTTPBearer() def current_user(creds: HTTPAuthorizationCredentials = Depends(bearer)) -> dict: token = creds.credentials try: signing_key = _jwk_client.get_signing_key_from_jwt(token).key claims = jwt.decode( token, signing_key, algorithms=["RS256"], audience=AUDIENCE, issuer=ISSUER, ) except jwt.PyJWTError as exc: raise HTTPException(status.HTTP_401_UNAUTHORIZED, f"invalid token: {exc}") return claims Fetching the JWKS once and caching it means validation is a fast local signature check — no network round-trip per request, so authentication adds negligible latency to the hot path.
2. Fine-grained authorization: OPA + Rego
Authentication tells you who someone is. It doesn't tell you what they're allowed to do. Baking that logic into application code means every rule change is a redeploy — and a security review nightmare.
Instead I externalized authorization to Open Policy Agent (OPA). The API asks OPA a simple question — "can this user perform this action on this resource?" — and OPA answers from policy written in Rego, versioned separately from the app.
# authz.rego — role-based access for the pipeline API package aiops.authz default allow := false # Operators can read scored events and trigger remediations allow if { input.role == "operator" input.action in {"read:events", "trigger:remediation"} } # Analysts are read-only allow if { input.role == "analyst" input.action == "read:events" } # Only admins can change model configuration allow if { input.role == "admin" } # the API consults OPA before serving a protected action import httpx async def authorize(role: str, action: str) -> bool: payload = {"input": {"role": role, "action": action}} async with httpx.AsyncClient() as client: r = await client.post("http://opa:8181/v1/data/aiops/authz/allow", json=payload) return r.json().get("result", False) The payoff: security policy becomes data, not code. I can tighten a rule, add a role, or audit exactly who can trigger a remediation — by reading one Rego file — without touching the pipeline.
3. Shipping clean images: Trivy in CI
A pipeline is only as trustworthy as the containers it runs. Before any image reached the environment, I scanned it with Trivy and failed the build on HIGH/CRITICAL vulnerabilities.
# .github/workflows/ci.yml — gate the build on image CVEs - name: Scan image with Trivy uses: aquasecurity/trivy-action@0.28.0 with: image-ref: ${{ env.IMAGE }}:${{ github.sha }} format: table exit-code: '1' # fail the pipeline on findings severity: CRITICAL,HIGH ignore-unfixed: true This moves security left: a vulnerable base image never gets the chance to run, and the feedback lands in the pull request instead of in production.
4. The part I'm proudest of: a self-healing loop
Monitoring tells you something broke. Self-healing does something about it. I added a lightweight health watcher that checks each service and, when one degrades, publishes a structured remediation event to a dedicated Kafka topic. A consumer subscribes to that topic and takes the recovery action — so the system responds to its own failures through the same event-driven backbone it uses for data.
# health_watcher.py — publish remediation events when a service degrades import json, time from kafka import KafkaProducer producer = KafkaProducer( bootstrap_servers="kafka:9092", value_serializer=lambda v: json.dumps(v).encode(), ) def watch(services: dict, interval: int = 10): while True: for name, check in services.items(): if not check(): # e.g. HTTP /health probe or consumer-lag threshold producer.send("remediation", { "service": name, "action": "restart", "ts": time.time(), }) time.sleep(interval) Using Kafka as the remediation channel (rather than a direct call) gave me three things for free: an audit trail of every incident and action, decoupling between detection and response, and the ability to add new responders later without touching the watcher.
What I'd tell someone building this
- Put identity at the gateway, authorization in policy. Keycloak + OPA/Rego kept auth logic out of the application and made it reviewable.
- Cache your JWKS. Token validation should never add a network hop to a real-time path.
- Fail the build, not production. Trivy in CI is a few lines and catches a whole class of problems early.
- Make remediation an event. Routing recovery through Kafka turned "monitoring" into "self-healing" and gave me an audit log as a side effect.
- Isolate the network. Running everything on a dedicated Docker bridge network meant services talked to each other, not to the world.
The lesson that stuck with me: in a real-time system, security and reliability can't be a layer you add at the end — they have to ride the same rails as the data.
I'm Muhammad Raees, an AIOps / DevSecOps engineer in Islamabad, Pakistan. I build and secure real-time, AI-driven cloud platforms on AWS and Azure. Code and more projects are on my GitHub and portfolio.
