apiVersion: eksctl.io/v1alpha5

kind: ClusterConfig

metadata:

name: pulseshield

region: us-east-1

version: "1.30"

iam:

withOIDC: true # required for IRSA

managedNodeGroups:

- name: workers

instanceType: m5.large

minSize: 2

desiredCapacity: 3

maxSize: 6

addons:

- name: vpc-cni

- name: coredns

- name: kube-proxy

- name: aws-ebs-csi-driver # needed only if Postgres runs in-cluster

const express = require('express');
const { Pool } = require('pg');
const promClient = require('prom-client');
const winston = require('winston');

const app = express();
app.use(express.json());

const logger = winston.createLogger({
level: 'info',
format: winston.format.json(),
transports: [new winston.transports.Console()]
});

const register = new promClient.Registry();
promClient.collectDefaultMetrics({ register });

const httpRequestDuration = new promClient.Histogram({
name: 'http_request_duration_seconds',
help: 'Duration of HTTP requests in seconds',
labelNames: ['method', 'route', 'status'],
registers: [register]
});

const httpRequestTotal = new promClient.Counter({
name: 'http_requests_total',
help: 'Total HTTP requests',
labelNames: ['method', 'route', 'status'],
registers: [register]
});

const pool = new Pool({
host: process.env.DB_HOST || 'postgres',
port: process.env.DB_PORT || 5432,
database: process.env.DB_NAME || 'orders',
user: process.env.DB_USER || 'postgres',
password: process.env.DB_PASSWORD || 'postgres',
ssl: process.env.DB_SSL === 'false' ? false : { rejectUnauthorized: false }
});

pool.query(`
CREATE TABLE IF NOT EXISTS orders (
id SERIAL PRIMARY KEY,
customer_name VARCHAR(255),
product VARCHAR(255),
quantity INT,
created_at TIMESTAMP DEFAULT NOW()
)
`).catch(err => logger.error('DB init error:', err));

app.use((req, res, next) => {
const end = httpRequestDuration.startTimer();
res.on('finish', () => {
end({ method: req.method, route: req.route?.path || req.path, status: res.statusCode });
httpRequestTotal.inc({ method: req.method, route: req.route?.path || req.path, status: res.statusCode });
});
next();
});

app.post('/orders', async (req, res) => {
try {
const { customer_name, product, quantity } = req.body;
const result = await pool.query(
'INSERT INTO orders (customer_name, product, quantity) VALUES ($1, $2, $3) RETURNING *',
[customer_name, product, quantity]
);
logger.info('Order created', { orderId: result.rows[0].id });
res.status(201).json(result.rows[0]);
} catch (err) {
logger.error('Order creation failed', { error: err.message });
res.status(500).json({ error: 'Internal server error' });
}
});

app.get('/orders/:id', async (req, res) => {
try {
const result = await pool.query('SELECT * FROM orders WHERE id = $1', [req.params.id]);
if (result.rows.length === 0) return res.status(404).json({ error: 'Order not found' });
res.json(result.rows[0]);
} catch (err) {
logger.error('Order fetch failed', { error: err.message });
res.status(500).json({ error: 'Internal server error' });
}
});

app.get('/health', (req, res) => {
res.json({ status: 'healthy', service: 'order-service' });
});

app.get('/metrics', async (req, res) => {
res.set('Content-Type', register.contentType);
res.end(await register.metrics());
});

const PORT = process.env.PORT || 3000;
app.listen(PORT, () => logger.info(`Order service running on port ${PORT}`));

from fastapi import FastAPI, HTTPException
from prometheus_client import Counter, Histogram, generate_latest, CONTENT_TYPE_LATEST
from fastapi.responses import Response
import logging
import os

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

app = FastAPI(title="Inventory Service")

inventory_db = {
"laptop": {"stock": 50, "price": 999.99},
"phone": {"stock": 100, "price": 699.99},
"tablet": {"stock": 75, "price": 499.99}
}

http_requests_total = Counter('http_requests_total', 'Total HTTP requests', ['method', 'endpoint', 'status'])
http_request_duration = Histogram('http_request_duration_seconds', 'HTTP request duration', ['method', 'endpoint'])

@app.middleware("http")
async def metrics_middleware(request, call_next):
with http_request_duration.labels(method=request.method, endpoint=request.url.path).time():
response = await call_next(request)
http_requests_total.labels(method=request.method, endpoint=request.url.path, status=response.status_code).inc()
return response

@app.get("/inventory/{product}")
async def get_inventory(product: str):
if product not in inventory_db:
logger.warning(f"Product not found: {product}")
raise HTTPException(status_code=404, detail="Product not found")
logger.info(f"Inventory checked for: {product}")
return inventory_db[product]

@app.post("/inventory/{product}/reduce")
async def reduce_inventory(product: str, quantity: int):
if product not in inventory_db:
raise HTTPException(status_code=404, detail="Product not found")
if inventory_db[product]["stock"] < quantity:
raise HTTPException(status_code=400, detail="Insufficient stock")
inventory_db[product]["stock"] -= quantity
logger.info(f"Reduced {quantity} units of {product}")
return {"product": product, "remaining_stock": inventory_db[product]["stock"]}

@app.get("/health")
async def health():
return {"status": "healthy", "service": "inventory-service"}

@app.get("/metrics")
async def metrics():
return Response(content=generate_latest(), media_type=CONTENT_TYPE_LATEST)

if __name__ == "__main__":
import uvicorn
port = int(os.getenv("PORT", 8000))
uvicorn.run(app, host="0.0.0.0", port=port)

def monitoring_loop():
logger.info("AI Agent monitoring started")
while True:
try:
anomalies = detect_anomalies()
if anomalies:
logger.info(f"Detected {len(anomalies)} anomalies")
for anomaly in anomalies:
metrics_snapshot = query_prometheus(f'{{job="{anomaly["service"]}"}}')
ai_response = analyze_with_ai(anomaly, metrics_snapshot)
action = ai_response.get("action", "alert")
reasoning = ai_response.get("reasoning", "No reasoning provided")
remediation_result = execute_remediation(action, anomaly['service'])
message = build_slack_message(anomaly, ai_response, remediation_result)
send_slack_notification(message)
log_incident({
'service': anomaly['service'],
'anomaly_type': anomaly['type'],
'metrics': anomaly,
'ai_diagnosis': reasoning,
'action': remediation_result,
'status': 'resolved' if action != 'alert' else 'alerted',
'slack_sent': True
})
else:
logger.info("No anomalies detected")
time.sleep(60)
except Exception as e:
logger.error(f"Monitoring loop error: {e}")
time.sleep(60)

def analyze_with_ai(anomaly, metrics_snapshot):
if not anthropic_client:
return {
"action": "alert",
"severity": "medium",
"summary": "AI not configured",
"root_cause": "N/A",
"impact": "N/A",
"reasoning": "AI not configured",
"remediation_steps": ["Check AI agent configuration"]
}

prompt = f"""You are a senior DevOps AI agent monitoring a Kubernetes microservices platform called PulseShield AI.

An anomaly has been detected. Analyze it thoroughly and respond with a detailed incident report.

Anomaly Details:
{json.dumps(anomaly, indent=2)}

Metrics Snapshot:
{json.dumps(metrics_snapshot, indent=2)}

Provide your response as JSON only with this exact structure:
{{
"action": "restart|scale|alert",
"severity": "low|medium|high|critical",
"summary": "one sentence summary of the issue",
"root_cause": "likely root cause of the anomaly",
"impact": "what is the impact on users and the system",
"reasoning": "detailed explanation of your analysis",
"remediation_steps": ["step 1", "step 2", "step 3"],
"scale_replicas": 3
}}"""

try:
message = anthropic_client.messages.create(
model=CLAUDE_MODEL,
max_tokens=2048,
messages=[{"role": "user", "content": prompt}]
)
response_text = message.content[0].text
start = response_text.find("{")
end = response_text.rfind("}") + 1
if start == -1 or end == 0:
raise ValueError(f"No JSON in response: {response_text[:200]}")
return json.loads(response_text[start:end], strict=False)
except Exception as e:
logger.error(f"AI analysis failed: {e}")
return {
"action": "alert",
"severity": "high",
"summary": f"Anomaly detected on {anomaly.get('service', 'unknown')}",
"root_cause": f"Metric value {anomaly.get('value')} exceeded threshold {anomaly.get('threshold')}",
"impact": "Service degradation detected. Users may experience errors or slow responses.",
"reasoning": f"Automated detection triggered. AI analysis unavailable: {str(e)}",
"remediation_steps": [
f"Check logs: kubectl logs -l app={anomaly.get('service')} -n pulseshield",
f"Check pods: kubectl get pods -n pulseshield",
f"Restart: kubectl rollout restart deployment/{anomaly.get('service')} -n pulseshield",
"Monitor metrics in Grafana dashboard",
"Escalate to on-call engineer if issue persists"
]
}

def execute_remediation(action, service, namespace="pulseshield"):
try:
if action == "restart":
k8s_apps.patch_namespaced_deployment(
name=service,
namespace=namespace,
body={"spec": {"template": {"metadata": {"annotations": {"kubectl.kubernetes.io/restartedAt": datetime.now().isoformat()}}}}}
)
logger.info(f"Restarted deployment: {service}")
return "restarted"
elif action == "scale":
deployment = k8s_apps.read_namespaced_deployment(service, namespace)
current_replicas = deployment.spec.replicas
new_replicas = current_replicas + 2
k8s_apps.patch_namespaced_deployment_scale(
name=service,
namespace=namespace,
body={"spec": {"replicas": new_replicas}}
)
logger.info(f"Scaled {service} from {current_replicas} to {new_replicas}")
return f"scaled to {new_replicas}"
return "alert_only"
except Exception as e:
logger.error(f"Remediation failed: {e}")
return f"failed: {e}"

apiVersion: v1
kind: ServiceAccount
metadata:
name: ai-agent-sa
namespace: {{ .Values.global.namespace }}
annotations:
# EKS / IRSA: binds this ServiceAccount to an IAM role.
# Only needed if the agent calls AWS APIs (e.g. AMP, Secrets Manager).
# Kubernetes-API-only remediation works without it.
eks.amazonaws.com/role-arn: arn:aws:iam::<ACCOUNT_ID>:role/pulseshield-ai-agent

---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
name: ai-agent-role
rules:
- apiGroups: ["apps"]
resources: ["deployments", "deployments/scale"]
verbs: ["get", "list", "patch", "update"]
- apiGroups: [""]
resources: ["pods"]
verbs: ["get", "list", "delete"]

---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
name: ai-agent-binding
subjects:
- kind: ServiceAccount
name: ai-agent-sa
namespace: {{ .Values.global.namespace }}
roleRef:
kind: ClusterRole
name: ai-agent-role
apiGroup: rbac.authorization.k8s.io

apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
name: pulseshield-gateway
namespace: pulseshield
annotations:
kubernetes.io/ingress.class: alb
alb.ingress.kubernetes.io/scheme: internet-facing
alb.ingress.kubernetes.io/target-type: ip
spec:
rules:
- http:
paths:
- path: /
pathType: Prefix
backend:
service:
name: nginx-gateway
port:
number: 80

CREATE TABLE IF NOT EXISTS incidents (
id SERIAL PRIMARY KEY,
timestamp TIMESTAMP DEFAULT NOW(),
service VARCHAR(255),
anomaly_type VARCHAR(255),
metrics_snapshot TEXT, -- full Prometheus response: reproducible
ai_diagnosis TEXT, -- Claude's reasoning, not just the action
remediation_action VARCHAR(255),
status VARCHAR(50),
slack_sent BOOLEAN DEFAULT FALSE
);