From Functional Infrastructure to AI-Orchestrated Infrastructure Enablement: How Responsibilities Are Changing

# Terraform: conventional DevOps provisioning
module "eks_cluster" {
  source          = "terraform-aws-modules/eks/aws"
  version         = "~> 20.0"
  cluster_name    = "platform-prod"
  cluster_version = "1.29"
  vpc_id          = module.vpc.vpc_id
  subnet_ids      = module.vpc.private_subnets

  eks_managed_node_groups = {
    general = {
      min_size     = 2
      max_size     = 10
      desired_size = 3
      instance_types = ["m6i.large"]
    }
  }
}

# .github/workflows/deploy.yml — conventional DevOps CI/CD
name: Deploy to EKS
on:
  push:
    branches: [main]

jobs:
  deploy:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Configure AWS credentials
        uses: aws-actions/configure-aws-credentials@v4
        with:
          role-to-assume: ${{ secrets.AWS_ROLE_ARN }}
          aws-region: us-east-1
      - name: Build and push Docker image
        run: |
          docker build -t $ECR_REGISTRY/$ECR_REPOSITORY:$GITHUB_SHA .
          docker push $ECR_REGISTRY/$ECR_REPOSITORY:$GITHUB_SHA          
      - name: Deploy to EKS
        run: |
          aws eks update-kubeconfig --name platform-prod --region us-east-1
          helm upgrade --install myapp ./charts/myapp \
            --set image.tag=$GITHUB_SHA \
            --namespace production          

# scripts/rotate_secrets.py — conventional DevOps scripting
import boto3
import json

def rotate_rds_password(secret_name: str, region: str = "us-east-1") -> None:
    """Rotate an RDS password stored in AWS Secrets Manager."""
    client = boto3.client("secretsmanager", region_name=region)
    client.rotate_secret(SecretId=secret_name)
    print(f"Rotation triggered for {secret_name}. Monitor CloudWatch for status.")

if __name__ == "__main__":
    rotate_rds_password("prod/rds/primary")

# Kyverno policy — hand-authored by a security engineer
apiVersion: kyverno.io/v1
kind: ClusterPolicy
metadata:
  name: require-pod-resource-limits
spec:
  validationFailureAction: enforce
  rules:
    - name: check-resource-limits
      match:
        resources:
          kinds: [Pod]
      validate:
        message: "Resource limits are required for all containers."
        pattern:
          spec:
            containers:
              - resources:
                  limits:
                    memory: "?*"
                    cpu: "?*"

# ai_provisioning_agent.py — AI-Orchestrated provisioning
import boto3
import json
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_aws import ChatBedrock
from langchain.tools import tool
from langchain_core.prompts import ChatPromptTemplate

llm = ChatBedrock(
    model_id="anthropic.claude-3-5-sonnet-20241022-v2:0",
    region_name="us-east-1",
)

@tool
def validate_terraform(hcl_code: str) -> str:
    """Run terraform validate and tflint on generated HCL code."""
    import subprocess
    result = subprocess.run(
        ["terraform", "validate"],
        input=hcl_code, capture_output=True, text=True
    )
    return result.stdout if result.returncode == 0 else result.stderr

@tool
def open_github_pr(title: str, body: str, branch: str, files: dict) -> str:
    """Open a GitHub pull request with the generated infrastructure code."""
    import requests
    # Implementation calls GitHub REST API
    return f"PR opened: https://github.com/org/infra/pull/42"

@tool
def check_aws_service_quotas(service: str, region: str) -> str:
    """Check AWS service quotas before provisioning to prevent failures."""
    client = boto3.client("service-quotas", region_name=region)
    response = client.list_service_quotas(ServiceCode=service)
    quotas = {q["QuotaName"]: q["Value"] for q in response["Quotas"]}
    return json.dumps(quotas)

# The agent reasons over tools to fulfill the infrastructure request
agent = create_tool_calling_agent(
    llm=llm,
    tools=[validate_terraform, open_github_pr, check_aws_service_quotas],
    prompt=ChatPromptTemplate.from_messages([
        ("system", "You are an expert AWS infrastructure engineer. "
                   "Generate secure, cost-optimized Terraform code, validate it, "
                   "check quotas, and open a PR. Follow company standards: "
                   "use existing VPC/subnet IDs, tag all resources, enable encryption."),
        ("human", "{input}"),
        ("placeholder", "{agent_scratchpad}"),
    ]),
)

executor = AgentExecutor(agent=agent, tools=[validate_terraform, open_github_pr, check_aws_service_quotas])

result = executor.invoke({
    "input": "Provision a new EKS cluster named 'ml-workloads-prod' in us-east-1 "
             "with GPU node groups for ML inference, auto-scaling from 0 to 20 nodes, "
             "and cost optimization via Karpenter."
})
print(result["output"])

# .github/workflows/ai-assisted-deploy.yml — AI-Orchestrated CI/CD
name: AI-Orchestrated Deploy
on:
  push:
    branches: [main]
  workflow_dispatch:

jobs:
  ai-plan:
    runs-on: ubuntu-latest
    outputs:
      deploy_plan: ${{ steps.plan.outputs.plan }}
      risk_level: ${{ steps.plan.outputs.risk_level }}
    steps:
      - uses: actions/checkout@v4
        with:
          fetch-depth: 0  # Full history for change analysis

      - name: AI Change Analysis
        id: plan
        uses: actions/github-script@v7
        with:
          script: |
            const { execSync } = require('child_process');
            const diff = execSync('git diff HEAD~1 HEAD -- terraform/').toString();

            // Call Amazon Bedrock via AWS SDK to analyze the diff
            const { BedrockRuntimeClient, InvokeModelCommand } = require("@aws-sdk/client-bedrock-runtime");
            const client = new BedrockRuntimeClient({ region: "us-east-1" });

            const response = await client.send(new InvokeModelCommand({
              modelId: "anthropic.claude-3-5-sonnet-20241022-v2:0",
              contentType: "application/json",
              body: JSON.stringify({
                anthropic_version: "bedrock-2023-05-31",
                max_tokens: 1024,
                messages: [{
                  role: "user",
                  content: `Analyze this Terraform diff and return a JSON object with:
                            - risk_level: "low" | "medium" | "high"
                            - summary: one-sentence description
                            - concerns: array of potential issues
                            Diff:\n${diff}`
                }]
              })
            }));

            const analysis = JSON.parse(JSON.parse(new TextDecoder().decode(response.body)).content[0].text);
            core.setOutput('plan', JSON.stringify(analysis));
            core.setOutput('risk_level', analysis.risk_level);            

  deploy:
    needs: ai-plan
    runs-on: ubuntu-latest
    environment: ${{ needs.ai-plan.outputs.risk_level == 'high' && 'production-gated' || 'production' }}
    steps:
      - uses: actions/checkout@v4
      - name: Configure AWS credentials
        uses: aws-actions/configure-aws-credentials@v4
        with:
          role-to-assume: ${{ secrets.AWS_ROLE_ARN }}
          aws-region: us-east-1
      - name: Terraform Apply
        run: terraform apply -auto-approve
      - name: Post-Deploy Validation
        run: python scripts/ai_smoke_test.py --env production

  ai-failure-triage:
    if: failure()
    needs: [ai-plan, deploy]
    runs-on: ubuntu-latest
    steps:
      - name: AI Triage & Auto-Fix
        uses: actions/github-script@v7
        with:
          github-token: ${{ secrets.GITHUB_TOKEN }}
          script: |
            // Fetch failure logs, send to Bedrock, get remediation steps,
            // open a PR with proposed fix or post to Slack with analysis
            console.log("AI triage agent analyzing failure...");            

# agents/incident_responder.py — Agentic runbook
"""
Incident response agent that autonomously diagnoses and remediates
common platform incidents using AWS, kubectl, and GitHub tools.
"""
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_aws import ChatBedrock
from langchain.tools import tool
from langchain_core.prompts import ChatPromptTemplate
import boto3
import subprocess
import json

llm = ChatBedrock(
    model_id="anthropic.claude-3-5-sonnet-20241022-v2:0",
    region_name="us-east-1",
    model_kwargs={"temperature": 0},
)

@tool
def get_cloudwatch_alarms(region: str = "us-east-1") -> str:
    """Retrieve all currently firing CloudWatch alarms."""
    cw = boto3.client("cloudwatch", region_name=region)
    alarms = cw.describe_alarms(StateValue="ALARM")
    return json.dumps([
        {"name": a["AlarmName"], "reason": a["StateReason"]}
        for a in alarms["MetricAlarms"]
    ])

@tool
def get_pod_logs(namespace: str, pod_selector: str, lines: int = 100) -> str:
    """Get recent logs from Kubernetes pods matching a label selector."""
    result = subprocess.run(
        ["kubectl", "logs", "-n", namespace, "-l", pod_selector,
         "--tail", str(lines), "--prefix"],
        capture_output=True, text=True
    )
    return result.stdout or result.stderr

@tool
def scale_deployment(namespace: str, deployment: str, replicas: int) -> str:
    """Scale a Kubernetes deployment to the specified replica count."""
    result = subprocess.run(
        ["kubectl", "scale", "deployment", deployment,
         "-n", namespace, f"--replicas={replicas}"],
        capture_output=True, text=True
    )
    return result.stdout

@tool
def get_rds_performance_insights(db_instance_id: str) -> str:
    """Retrieve RDS Performance Insights data for the last 5 minutes."""
    pi = boto3.client("pi", region_name="us-east-1")
    import datetime
    end = datetime.datetime.utcnow()
    start = end - datetime.timedelta(minutes=5)
    response = pi.get_resource_metrics(
        ServiceType="RDS",
        Identifier=f"db:{db_instance_id}",
        MetricQueries=[{"Metric": "db.load.avg"}],
        StartTime=start,
        EndTime=end,
        PeriodInSeconds=60,
    )
    return json.dumps(response.get("MetricList", []), default=str)

@tool
def post_to_slack(channel: str, message: str) -> str:
    """Post a message to a Slack channel via webhook."""
    import requests, os
    webhook = os.environ["SLACK_WEBHOOK_URL"]
    requests.post(webhook, json={"channel": channel, "text": message})
    return "Posted to Slack"

incident_agent = AgentExecutor(
    agent=create_tool_calling_agent(
        llm=llm,
        tools=[get_cloudwatch_alarms, get_pod_logs, scale_deployment,
               get_rds_performance_insights, post_to_slack],
        prompt=ChatPromptTemplate.from_messages([
            ("system",
             "You are an expert SRE. When called, you MUST: "
             "1) Gather evidence from CloudWatch, pod logs, and RDS. "
             "2) Diagnose the root cause. "
             "3) Take the minimum safe remediation action (prefer scaling over restarts). "
             "4) Post a clear summary to #incidents on Slack with: "
             "   - Root cause, actions taken, current status, recommended follow-up. "
             "Never take destructive actions (delete, force-kill) without explicit human instruction."),
            ("human", "{input}"),
            ("placeholder", "{agent_scratchpad}"),
        ]),
    ),
    tools=[get_cloudwatch_alarms, get_pod_logs, scale_deployment,
           get_rds_performance_insights, post_to_slack],
    max_iterations=10,
    verbose=True,
)

# Triggered by CloudWatch EventBridge → Lambda → this agent
incident_agent.invoke({"input": "High latency alarm fired on the checkout API. Investigate and remediate."})

# agents/security_posture_agent.py
"""
Continuously assesses AWS security posture using AWS Security Hub findings,
generates remediation Terraform PRs, and tracks compliance trends.
"""
import boto3
import json
from langchain_aws import ChatBedrock
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain.tools import tool
from langchain_core.prompts import ChatPromptTemplate

llm = ChatBedrock(model_id="anthropic.claude-3-5-sonnet-20241022-v2:0", region_name="us-east-1")

@tool
def get_security_hub_findings(severity: str = "HIGH") -> str:
    """Fetch open AWS Security Hub findings filtered by severity."""
    sh = boto3.client("securityhub", region_name="us-east-1")
    findings = sh.get_findings(
        Filters={
            "SeverityLabel": [{"Value": severity, "Comparison": "EQUALS"}],
            "WorkflowStatus": [{"Value": "NEW", "Comparison": "EQUALS"}],
        },
        MaxResults=20,
    )
    return json.dumps([
        {
            "title": f["Title"],
            "resource": f["Resources"][0]["Id"],
            "description": f["Description"],
        }
        for f in findings["Findings"]
    ])

@tool
def generate_remediation_pr(finding_title: str, resource_arn: str,
                            remediation_terraform: str) -> str:
    """
    Open a GitHub PR with Terraform code to remediate a Security Hub finding.
    The PR is tagged with 'security-remediation' and auto-assigned to the security team.
    """
    # Implementation: create branch, commit HCL, open PR via GitHub API
    return f"PR opened for remediation of: {finding_title}"

@tool
def suppress_finding(finding_id: str, reason: str) -> str:
    """Suppress a Security Hub finding with a documented reason (for accepted risks)."""
    sh = boto3.client("securityhub", region_name="us-east-1")
    sh.batch_update_findings(
        FindingIdentifiers=[{"Id": finding_id, "ProductArn": "..."}],
        Workflow={"Status": "SUPPRESSED"},
        Note={"Text": reason, "UpdatedBy": "security-posture-agent"},
    )
    return f"Finding suppressed: {reason}"

security_agent = AgentExecutor(
    agent=create_tool_calling_agent(
        llm=llm,
        tools=[get_security_hub_findings, generate_remediation_pr, suppress_finding],
        prompt=ChatPromptTemplate.from_messages([
            ("system",
             "You are an AWS security specialist. For each HIGH severity finding: "
             "1) Understand the risk and the affected resource. "
             "2) Generate Terraform code that remediates the finding while "
             "   preserving existing functionality. "
             "3) Open a GitHub PR with the fix. "
             "4) Only suppress a finding if remediation is architecturally impossible "
             "   and document the accepted risk clearly."),
            ("human", "{input}"),
            ("placeholder", "{agent_scratchpad}"),
        ]),
    ),
    tools=[get_security_hub_findings, generate_remediation_pr, suppress_finding],
)

security_agent.invoke({"input": "Review all HIGH severity Security Hub findings and remediate or suppress with justification."})

# agents/observability_agent.py
"""
AI agent that correlates signals from CloudWatch, X-Ray, and application logs
to perform root cause analysis and predict failures before they impact users.
"""
import boto3
import json
from langchain_aws import ChatBedrock
from langchain.tools import tool
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate

llm = ChatBedrock(model_id="anthropic.claude-3-5-sonnet-20241022-v2:0", region_name="us-east-1")

@tool
def query_cloudwatch_logs(log_group: str, query: str, minutes: int = 15) -> str:
    """Run a CloudWatch Logs Insights query and return results as JSON."""
    import datetime, time
    cw = boto3.client("logs", region_name="us-east-1")
    end = datetime.datetime.utcnow()
    start = end - datetime.timedelta(minutes=minutes)
    resp = cw.start_query(logGroupName=log_group, startTime=int(start.timestamp()),
                          endTime=int(end.timestamp()), queryString=query)
    query_id = resp["queryId"]
    while True:
        result = cw.get_query_results(queryId=query_id)
        if result["status"] in ("Complete", "Failed", "Cancelled"):
            return json.dumps(result["results"])
        time.sleep(1)

@tool
def get_xray_service_map(minutes: int = 30) -> str:
    """Retrieve the AWS X-Ray service map to identify latency hotspots."""
    import datetime
    xray = boto3.client("xray", region_name="us-east-1")
    end = datetime.datetime.utcnow()
    start = end - datetime.timedelta(minutes=minutes)
    response = xray.get_service_graph(StartTime=start, EndTime=end)
    services = [
        {"name": s["Name"], "avg_latency_ms": s.get("SummaryStatistics", {}).get("OkCount", 0)}
        for s in response.get("Services", [])
    ]
    return json.dumps(services)

@tool
def get_cloudwatch_metric(namespace: str, metric_name: str, dimensions: dict,
                          minutes: int = 30) -> str:
    """Retrieve CloudWatch metric statistics for anomaly correlation."""
    import datetime
    cw = boto3.client("cloudwatch", region_name="us-east-1")
    end = datetime.datetime.utcnow()
    start = end - datetime.timedelta(minutes=minutes)
    response = cw.get_metric_statistics(
        Namespace=namespace, MetricName=metric_name,
        Dimensions=[{"Name": k, "Value": v} for k, v in dimensions.items()],
        StartTime=start, EndTime=end, Period=60, Statistics=["Average", "Maximum"],
    )
    return json.dumps(response["Datapoints"], default=str)

observability_agent = AgentExecutor(
    agent=create_tool_calling_agent(
        llm=llm,
        tools=[query_cloudwatch_logs, get_xray_service_map, get_cloudwatch_metric],
        prompt=ChatPromptTemplate.from_messages([
            ("system",
             "You are a senior SRE with expertise in distributed systems observability. "
             "When investigating an incident: "
             "1) Correlate signals across metrics, traces, and logs. "
             "2) Identify the root cause service and failure mode. "
             "3) Distinguish between symptoms and root cause. "
             "4) Provide a structured RCA with: timeline, root cause, blast radius, "
             "   immediate mitigation, and long-term fix recommendation."),
            ("human", "{input}"),
            ("placeholder", "{agent_scratchpad}"),
        ]),
    ),
    tools=[query_cloudwatch_logs, get_xray_service_map, get_cloudwatch_metric],
)