Infrastructure Agent for DevOps

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.

The key capabilities are: infrastructure-as-code management (Terraform), real-time monitoring and alerting, automated incident response and remediation, log analysis and root cause identification, and cloud cost optimization.

These agents reduce MTTR (Mean Time To Resolution) by automating detection, diagnosis, and common remediation tasks while escalating complex issues with full context.

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.

Difficulty: Advanced (requires understanding of cloud infrastructure, IaC, and monitoring)

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: Project Structure

Architecture Diagram

devops-agent/
├── terraform/
│   ├── __init__.py
│   └── manager.py
├── monitoring/
│   ├── __init__.py
│   └── prometheus.py
├── incidents/
│   ├── __init__.py
│   ├── detector.py
│   └── responder.py
├── costs/
│   ├── __init__.py
│   └── optimizer.py
├── agent.py
└── main.py

Step 3: Terraform Manager

# terraform/manager.py
import subprocess
import json
from typing import Dict, List

class TerraformManager:
    def __init__(self, working_dir: str = "./terraform"):
        self.working_dir = working_dir

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

    def plan(self) -> Dict:
        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:
        flag = "-auto-approve" if auto_approve else ""
        return self._run(f"terraform apply {flag} -json")

    def destroy(self, auto_approve: bool = False) -> Dict:
        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:
        return self._run("terraform validate -json")

    def format_check(self) -> Dict:
        return self._run("terraform fmt -check -recursive")

    def cost_estimate(self) -> Dict:
        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:
                pass
        return {"error": "Could not estimate costs"}

Step 4: Monitoring and Alerting

# monitoring/prometheus.py
import httpx
from typing import Dict, List
from datetime import datetime, timedelta

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

    def query(self, promql: str) -> Dict:
        response = httpx.get(f"{self.base_url}/api/v1/query", params={"query": promql})
        return response.json()

    def query_range(
        self, promql: str, start: datetime, end: datetime, step: str = "60s"
    ) -> Dict:
        response = httpx.get(
            f"{self.base_url}/api/v1/query_range",
            params={"query": promql, "start": start.isoformat(), "end": end.isoformat(), "step": step},
        )
        return response.json()

    def get_cpu_usage(self, instance: str = ".*") -> float:
        result = 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

    def get_memory_usage(self, instance: str = ".*") -> float:
        result = 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

    def get_disk_usage(self, instance: str = ".*") -> float:
        result = 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

    def get_alerts(self) -> List[Dict]:
        response = httpx.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", [])
        ]

# incidents/detector.py
from openai import OpenAI
from typing import Dict, List
import json

class IncidentDetector:
    def __init__(self, model: str = "gpt-4-turbo-preview"):
        self.client = OpenAI()
        self.model = model

    def analyze_incident(self, alerts: List[Dict], metrics: Dict) -> Dict:
        response = 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:
            return {"severity": "unknown", "root_cause_hypothesis": "Unable to determine"}

# incidents/responder.py
from typing import Dict
import subprocess

class IncidentResponder:
    def execute_runbook(self, runbook_name: str, context: Dict) -> Dict:
        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:
            return action(context)
        return {"success": False, "error": f"Unknown runbook: {runbook_name}"}

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

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

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

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

Step 5: Complete Agent

# agent.py
from terraform.manager import TerraformManager
from monitoring.prometheus import PrometheusClient
from incidents.detector import IncidentDetector
from incidents.responder import IncidentResponder
from typing import Dict, List

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

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

    def handle_incident(self) -> Dict:
        alerts = self.monitoring.get_alerts()
        metrics = self.infrastructure_status()
        analysis = 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:
        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):

Intuition: 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.

Intuition: Composite health score across all infrastructure dimensions.

Testing & Evaluation

import pytest
from terraform.manager import TerraformManager
from monitoring.prometheus import PrometheusClient

def test_terraform_validate():
    manager = TerraformManager()
    result = manager.validate()
    assert "success" in result

def test_monitoring():
    client = PrometheusClient()
    alerts = client.get_alerts()
    assert isinstance(alerts, list)

Performance Metrics

Metric	Value	Notes
Terraform Plan	10-60s	Depends on infrastructure size
Metric Query	100-500ms	Prometheus
Incident Analysis	3-8s	GPT-4 with context
Auto-Remediation	30s-5min	Depends on action
Health Check Interval	60s	Configurable

Deployment

# main.py
from agent import DevOpsAgent
import json

def main():
    agent = DevOpsAgent()
    print("DevOps Agent Ready\n")
    while True:
        cmd = input("Command (status/incident/terraform/quit): ").strip()
        if cmd == "quit":
            break
        elif cmd == "status":
            status = agent.infrastructure_status()
            print(json.dumps(status, indent=2, default=str))

if __name__ == "__main__":
    main()

Real-World Use Cases

SRE Operations: Automated incident response and runbook execution
Cloud Management: Resource provisioning and optimization
Security Operations: Vulnerability detection and remediation
Cost Management: Cloud spend optimization
Compliance: Infrastructure audit and reporting

Common Pitfalls & Solutions

Pitfall	Solution
False positive alerts	Tune thresholds, implement deduplication
Runbook failures	Add rollback mechanisms
State file conflicts	Use remote state with locking
Cost overruns	Implement budget alerts
Configuration drift	Regular terraform plan checks

Summary with Key Takeaways

Terraform integration enables safe infrastructure changes
Real-time monitoring with Prometheus provides instant visibility
Automated incident response reduces MTTR significantly
LLM-powered analysis provides human-like incident diagnosis
Always implement rollback mechanisms for automated actions

Infrastructure Agent for DevOps

Infrastructure Agent for DevOps

What is a DevOps Agent?

Project Overview

Architecture

Tools & Setup

Step 1: Environment Setup

Step 2: Project Structure

Step 3: Terraform Manager

Step 4: Monitoring and Alerting

Step 5: Complete Agent

Mathematical Foundation

Testing & Evaluation

Performance Metrics

Deployment

Real-World Use Cases

Common Pitfalls & Solutions

Summary with Key Takeaways

Need Expert AI Agents Help?