DevOps Infrastructure AI Agent

Why This Matters

DevOps teams face relentless pressure to maintain uptime, manage infrastructure, and respond to incidents—often simultaneously. An AI-powered DevOps agent automates routine tasks like Terraform operations, monitoring, and incident response, reducing MTTR (Mean Time To Resolution) by 40-60% while maintaining strict audit trails and safety controls.

Real-World Analogy

Think of a DevOps Agent as an air traffic controller for your infrastructure. Just as an air traffic controller monitors thousands of flights, detects anomalies, and coordinates emergency responses, this agent monitors system metrics, detects incidents, and orchestrates remediation—all while maintaining a complete audit trail of every action taken.

What is a DevOps Agent?

DevOps agents automate infrastructure management, monitoring, incident response, and cost optimization. They bridge the gap between development and operations by providing intelligent automation for routine tasks and decision support for complex situations. Key capabilities: IaC management (Terraform), real-time monitoring and alerting, automated incident response, log analysis, and cloud cost optimization.

Project Overview

We will build a DevOps agent that:

Manages Terraform infrastructure state
Monitors system metrics and generates alerts
Analyzes logs for anomalies and errors
Automates incident response runbooks
Optimizes cloud resource utilization
Generates infrastructure reports

Expected outcome: An agent that automates 60%+ of routine DevOps tasks.

Architecture

Tools & Setup

Tool	Version	Purpose
Python	3.11+	Core language
subprocess	stdlib	Terraform CLI
httpx	0.27+	API calls
openai	1.0+	LLM backbone
pydantic	2.0+	Data models

Step 1: Environment Setup

python -m venv venv
source venv/bin/activate
pip install httpx openai pydantic
export OPENAI_API_KEY="sk-your-key"
export PROMETHEUS_URL="http://localhost:9090"

Step 2: Terraform Manager

import subprocess
import json
import logging
from typing import Any, Dict, List, Optional

logger = logging.getLogger(__name__)


class TerraformManager:
    """Manage Terraform infrastructure operations with safety controls."""

    def __init__(self, working_dir: str = "./terraform", timeout: int = 300):
        self.working_dir = working_dir
        self.timeout = timeout

    def _run(self, command: str) -> Dict[str, Any]:
        try:
            result = subprocess.run(
                command,
                shell=True,
                cwd=self.working_dir,
                capture_output=True,
                text=True,
                timeout=self.timeout,
            )
            return {
                "success": result.returncode == 0,
                "stdout": result.stdout,
                "stderr": result.stderr,
                "return_code": result.returncode,
            }
        except subprocess.TimeoutExpired:
            logger.error(f"Terraform command timed out: {command}")
            return {"success": False, "stdout": "", "stderr": "Command timed out", "return_code": -1}

    def plan(self) -> Dict[str, Any]:
        result = self._run("terraform plan -json")
        if result["success"]:
            try:
                plan_data = json.loads(result["stdout"])
                return {
                    "success": True,
                    "changes": plan_data.get("changes", {}),
                    "resource_changes": plan_data.get("resource_changes", []),
                }
            except json.JSONDecodeError:
                return {"success": True, "raw_output": result["stdout"]}
        return result

    def apply(self, auto_approve: bool = False) -> Dict[str, Any]:
        flag = "-auto-approve" if auto_approve else ""
        return self._run(f"terraform apply {flag} -json")

    def destroy(self, auto_approve: bool = False) -> Dict[str, Any]:
        flag = "-auto-approve" if auto_approve else ""
        return self._run(f"terraform destroy {flag} -json")

    def state_list(self) -> List[str]:
        result = self._run("terraform state list")
        if result["success"]:
            return [line.strip() for line in result["stdout"].split("\n") if line.strip()]
        return []

    def validate(self) -> Dict[str, Any]:
        return self._run("terraform validate -json")

    def cost_estimate(self) -> Dict[str, Any]:
        result = self._run("terraform plan -json")
        if result["success"]:
            try:
                plan = json.loads(result["stdout"])
                resources = plan.get("resource_changes", [])
                return {
                    "total_resources": len(resources),
                    "to_add": sum(1 for r in resources if r.get("change", {}).get("actions", []) == ["create"]),
                    "to_change": sum(1 for r in resources if "update" in r.get("change", {}).get("actions", [])),
                    "to_destroy": sum(1 for r in resources if "destroy" in r.get("change", {}).get("actions", [])),
                }
            except (json.JSONDecodeError, KeyError):
                pass
        return {"error": "Could not estimate costs"}

Step 3: Monitoring and Incident Detection

import httpx
from typing import Any, Dict, List
from datetime import datetime, timedelta


class PrometheusClient:
    """Async Prometheus API client for metrics and alerts."""

    def __init__(self, base_url: str = "http://localhost:9090"):
        self.base_url = base_url

    async def query(self, promql: str) -> Dict[str, Any]:
        async with httpx.AsyncClient() as client:
            response = await client.get(f"{self.base_url}/api/v1/query", params={"query": promql})
            return response.json()

    async def get_cpu_usage(self, instance: str = ".*") -> float:
        result = await self.query(
            f'100 - (avg(rate(node_cpu_seconds_total{{mode="idle", instance=~"{instance}"}}[5m])) * 100)'
        )
        if result.get("data", {}).get("result"):
            return float(result["data"]["result"][0]["value"][1])
        return 0.0

    async def get_memory_usage(self, instance: str = ".*") -> float:
        result = await self.query(
            f'(1 - node_memory_MemAvailable_bytes / node_memory_MemTotal_bytes){{instance=~"{instance}"}} * 100'
        )
        if result.get("data", {}).get("result"):
            return float(result["data"]["result"][0]["value"][1])
        return 0.0

    async def get_disk_usage(self, instance: str = ".*") -> float:
        result = await self.query(
            f'(1 - node_filesystem_avail_bytes{{mountpoint="/",instance=~"{instance}"}} / node_filesystem_size_bytes{{mountpoint="/",instance=~"{instance}"}}) * 100'
        )
        if result.get("data", {}).get("result"):
            return float(result["data"]["result"][0]["value"][1])
        return 0.0

    async def get_alerts(self) -> List[Dict[str, Any]]:
        async with httpx.AsyncClient() as client:
            response = await client.get(f"{self.base_url}/api/v1/alerts")
            data = response.json()
        return [
            {
                "name": alert.get("labels", {}).get("alertname", "Unknown"),
                "severity": alert.get("labels", {}).get("severity", "unknown"),
                "instance": alert.get("labels", {}).get("instance", "unknown"),
                "description": alert.get("annotations", {}).get("description", ""),
                "active_at": alert.get("activeAt", ""),
            }
            for alert in data.get("data", {}).get("alerts", [])
        ]

Step 4: Incident Detector and Responder

from openai import AsyncOpenAI
import json
import subprocess
import logging
from typing import Any, Dict

logger = logging.getLogger(__name__)


class IncidentDetector:
    """Analyze infrastructure alerts and metrics using LLM."""

    def __init__(self, model: str = "gpt-4o"):
        self.client = AsyncOpenAI()
        self.model = model

    async def analyze_incident(self, alerts: List[Dict], metrics: Dict) -> Dict[str, Any]:
        response = await self.client.chat.completions.create(
            model=self.model,
            messages=[
                {
                    "role": "system",
                    "content": """Analyze infrastructure alerts and metrics.
                    Determine severity, root cause hypothesis, and recommended actions.
                    Return JSON:
                    {
                        "severity": "critical|high|medium|low",
                        "root_cause_hypothesis": "likely cause",
                        "impact": "description of impact",
                        "recommended_actions": ["ordered list of actions"],
                        "escalation_needed": true/false
                    }""",
                },
                {
                    "role": "user",
                    "content": f"Alerts:\n{json.dumps(alerts, indent=2)}\n\nMetrics:\n{json.dumps(metrics, indent=2)}",
                },
            ],
            temperature=0.0,
        )
        try:
            return json.loads(response.choices[0].message.content)
        except (json.JSONDecodeError, IndexError):
            return {"severity": "unknown", "root_cause_hypothesis": "Unable to determine", "escalation_needed": True}


class IncidentResponder:
    """Execute remediation runbooks with safety controls."""

    def __init__(self, cooldown_seconds: int = 300):
        self.cooldown_seconds = cooldown_seconds
        self.last_execution: Dict[str, float] = {}

    def execute_runbook(self, runbook_name: str, context: Dict[str, Any]) -> Dict[str, Any]:
        import time
        now = time.time()
        if runbook_name in self.last_execution:
            elapsed = now - self.last_execution[runbook_name]
            if elapsed < self.cooldown_seconds:
                return {"success": False, "error": f"Cooldown active, retry in {int(self.cooldown_seconds - elapsed)}s"}

        actions = {
            "restart_service": self._restart_service,
            "scale_up": self._scale_up,
            "clear_cache": self._clear_cache,
            "rollback": self._rollback,
        }
        action = actions.get(runbook_name)
        if action:
            self.last_execution[runbook_name] = now
            return action(context)
        return {"success": False, "error": f"Unknown runbook: {runbook_name}"}

    def _restart_service(self, context: Dict[str, Any]) -> Dict[str, Any]:
        service = context.get("service", "nginx")
        try:
            result = subprocess.run(
                ["sudo", "systemctl", "restart", service],
                capture_output=True, text=True, timeout=30,
            )
            return {"success": result.returncode == 0, "output": result.stdout}
        except Exception as e:
            return {"success": False, "error": str(e)}

    def _scale_up(self, context: Dict[str, Any]) -> Dict[str, Any]:
        return {"success": True, "message": f"Scaling {context.get('service', 'unknown')} up"}

    def _clear_cache(self, context: Dict[str, Any]) -> Dict[str, Any]:
        return {"success": True, "message": "Cache cleared"}

    def _rollback(self, context: Dict[str, Any]) -> Dict[str, Any]:
        return {"success": True, "message": f"Rolling back to {context.get('version', 'previous')}"}

Step 5: Complete Agent

from typing import Any, Dict, List, Optional


class DevOpsAgent:
    """Orchestrate DevOps operations across infrastructure."""

    def __init__(
        self,
        prometheus_url: str = "http://localhost:9090",
        model: str = "gpt-4o",
    ):
        self.terraform = TerraformManager()
        self.monitoring = PrometheusClient(prometheus_url)
        self.detector = IncidentDetector(model)
        self.responder = IncidentResponder()

    async def infrastructure_status(self) -> Dict[str, Any]:
        return {
            "cpu_usage": await self.monitoring.get_cpu_usage(),
            "memory_usage": await self.monitoring.get_memory_usage(),
            "disk_usage": await self.monitoring.get_disk_usage(),
            "active_alerts": await self.monitoring.get_alerts(),
        }

    async def handle_incident(self) -> Dict[str, Any]:
        alerts = await self.monitoring.get_alerts()
        metrics = await self.infrastructure_status()
        analysis = await self.detector.analyze_incident(alerts, metrics)
        if analysis.get("recommended_actions"):
            for action in analysis["recommended_actions"][:1]:
                self.responder.execute_runbook(action, {"service": "unknown"})
        return {
            "alerts_count": len(alerts),
            "analysis": analysis,
            "status": "handled" if not analysis.get("escalation_needed") else "escalated",
        }

    def terraform_plan_review(self) -> Dict[str, Any]:
        plan = self.terraform.plan()
        if plan.get("success"):
            resources = plan.get("resource_changes", [])
            return {
                "total_changes": len(resources),
                "additions": sum(1 for r in resources if "create" in r.get("change", {}).get("actions", [])),
                "modifications": sum(1 for r in resources if "update" in r.get("change", {}).get("actions", [])),
                "deletions": sum(1 for r in resources if "destroy" in r.get("change", {}).get("actions", [])),
            }
        return plan

Mathematical Foundation

MTTR (Mean Time To Resolution):

Average time from incident detection to resolution. Lower MTTR indicates faster recovery.

Infrastructure Health Score:

Where , , are CPU, memory, disk utilization (inverted), and is alert penalty. Composite health score from 0 (critical) to 1 (healthy).

SLA Availability:

Target: 99.9% (8.76 hours downtime/year) for most production systems.

Performance Considerations

Metric	Latency	Cost	Accuracy
Terraform plan	5-30s	Free (local)	Exact
Prometheus query	<500ms	Free	Exact
Incident analysis	3-8s	$0.02-0.05	High
Runbook execution	1-10s	Free (local)	High
Full health check	2-5s	$0.01	High

Security Considerations

Never use auto_approve with terraform destroy in production
Implement RBAC for infrastructure operations
Use remote state with DynamoDB locking to prevent conflicts
Audit all automated actions with append-only logs
Store secrets in Vault or AWS Secrets Manager, not in code
Validate Terraform plans before applying changes
Implement cooldown periods to prevent remediation loops

Interview Q&A

Q1: What safety mechanisms prevent the Terraform agent from making destructive changes?

The agent uses terraform plan -json to preview changes before applying. Destructive operations require explicit auto_approve=False. The LLM reviews the plan and flags destructive changes. In production, implement confirmation steps and ChatOps approval.

Q2: How does the agent handle Terraform state file conflicts?

Use remote state backends (S3 + DynamoDB) with state locking. DynamoDB provides optimistic locking—only one terraform apply can proceed at a time. Check lock status before operations and handle state locked errors gracefully.

Q3: What is the difference between MTTR and MTTD?

MTTD measures time from incident occurrence to detection. MTTR measures detection to resolution. The agent reduces both: monitoring reduces MTTD, automated remediation reduces MTTR. Both are critical SRE metrics.

Q4: How do you prevent false positive alerts from triggering unnecessary remediation?

Implement alert correlation—only trigger remediation when multiple related alerts fire. Use the LLM to analyze alert patterns and distinguish real incidents from transient issues. Add cooldown periods between remediation attempts.

Q5: How would you implement blue-green deployments with this agent?

Extend Terraform to support two identical environments. Route traffic to active, deploy to inactive, validate health checks, then switch. Rollback is instant by switching back. The LLM generates deployment plans and validates each step.

Q6: How does the agent handle multi-cloud infrastructure?

Abstract cloud providers behind a unified interface. Use Terraform providers for each cloud (AWS, GCP, Azure). Monitoring aggregates metrics from all clouds into a single view. The incident detector analyzes cross-cloud dependencies.

Q7: What metrics should the infrastructure health score track?

CPU utilization, memory usage, disk usage, network throughput, error rates, and active alerts. Each normalized to 0-1 (1 = healthy). Weights customized per infrastructure type (e.g., database servers weight memory higher).

Q8: How do you audit all automated actions taken by the agent?

Implement comprehensive audit logs: timestamp, action type, target resource, before/after state, trigger (agent or human), and outcome. Store in append-only database. Generate compliance reports from the audit trail.

Common Pitfalls & Solutions

Pitfall	Solution
False positive alerts	Tune thresholds; implement alert correlation; add cooldowns
Runbook failures	Add rollback mechanisms; test runbooks in staging
State file conflicts	Use remote state with DynamoDB locking
Cost overruns	Implement budget alerts; review terraform cost estimates
Configuration drift	Regular `terraform plan` checks; drift detection
Alert fatigue	Prioritize by severity; aggregate related alerts
Unauthorized changes	Implement RBAC; audit all terraform operations
Multi-environment complexity	Use workspaces; separate state files per environment

Knowledge Check

Q1: What does MTTR measure? A) Time from incident to detection B) Time from detection to resolution C) Total downtime D) Time between deployments

Answer

B) Time from detection to resolution.

Q2: In the health score formula, why are utilization metrics inverted? A) Simpler formula B) Higher utilization means less healthy C) Prometheus formatting D) Terraform requires it

Answer

B) Higher utilization means less healthy, so inversion makes higher = healthier.

Q3: What is the recommended backend for Terraform state in production? A) Local file B) S3 + DynamoDB with state locking C) Git repository D) PostgreSQL

Answer

B) S3 + DynamoDB with state locking for durable storage and concurrent access control.

Q4: What does a 99.9% SLA availability target allow in annual downtime? A) 8.76 hours B) 87.6 hours C) 876 hours D) No downtime

Answer

A) 8.76 hours ().

Q5: Why should terraform destroy never use auto_approve in production? A) Slower B) Requires manual confirmation to prevent accidental deletion C) Not supported D) Increases cost

Answer

B) Manual confirmation prevents accidental infrastructure deletion.

Q6: What is the primary benefit of alert correlation? A) Reduces API costs B) Prevents false positive remediation C) Increases alert volume D) Simplifies config

Answer

B) Prevents false positive remediation by requiring multiple related alerts.

Summary with Key Takeaways

Terraform integration enables safe infrastructure changes with plan-before-apply workflow
Real-time monitoring with Prometheus provides instant visibility into system health
Automated incident response reduces MTTR significantly while maintaining audit trails
LLM-powered analysis provides human-like incident diagnosis and root cause identification
Always implement rollback mechanisms for automated actions—never blindly remediate
Remote state with locking prevents concurrent modification conflicts
The health score provides a single number for overall infrastructure status

DevOps Infrastructure AI Agent

DevOps Infrastructure AI Agent

Why This Matters

Real-World Analogy

What is a DevOps Agent?

Project Overview

Architecture

Tools & Setup

Step 1: Environment Setup

Step 2: Terraform Manager

Step 3: Monitoring and Incident Detection

Step 4: Incident Detector and Responder

Step 5: Complete Agent

Mathematical Foundation

Performance Considerations

Security Considerations

Interview Q&A

Common Pitfalls & Solutions

Knowledge Check

Summary with Key Takeaways

Need Expert AI Agents Help?