CI/CD & Infrastructure Automation with AI

# .github/workflows/deploy-with-ai-gate.yml
name: AI-Enhanced Deployment
on:
  push:
tags:
      - 'v*'

jobs:
  ai-gate:
    runs-on: ubuntu-latest
    outputs:
      approved: ${{ steps.gate.outputs.approved }}
      reason: ${{ steps.gate.outputs.reason }}
    steps:
      - uses: actions/checkout@v4

      - name: Gather deployment context
        id: context
        run: |
          echo "version=$(git describe --tags)" >> $GITHUB_OUTPUT
          echo "commits=$(git log --oneline $(git describe --tags --abbrev=0)..HEAD | wc -l)" >> $GITHUB_OUTPUT
          echo "changed_files=$(git diff --name-only HEAD~10..HEAD | tr '\n' ' ')" >> $GITHUB_OUTPUT

      - name: AI Deployment Gate
        id: gate
        run: |
          python scripts/ai_deployment_gate.py \
            --version ${{ steps.context.outputs.version }} \
            --commits ${{ steps.context.outputs.commits }} \
            --changes "${{ steps.context.outputs.changed_files }}" \
            --test-results "$(cat test-results.json)"

import os
import json
from openai import OpenAI

client = OpenAI()

def evaluate_deployment(version, commits_count, changed_files, test_results):
    """Use AI to evaluate whether a deployment is safe."""

    test_summary = json.loads(test_results)
    pass_rate = test_summary.get("pass_rate", 0)
    coverage = test_summary.get("coverage", 0)

    prompt = f"""
Evaluate whether this deployment is safe to proceed:

Version: {version}
Commits since last tag: {commits_count}
Changed files: {changed_files}
Test pass rate: {pass_rate}%
Code coverage: {coverage}%

Respond in JSON format:
{{
  "approved": true/false,
  "confidence": 0.0-1.0,
  "reason": "brief explanation",
  "risks": ["list of identified risks"],
  "suggestions": ["precautions to take"]
}}

Consider:
- High commit count with low coverage = higher risk
- Core file changes without tests = flag
- Configuration changes = needs review
- Dependency updates = potential breaking changes
"""

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        response_format={"type": "json_object"},
        temperature=0.2
    )

    decision = json.loads(response.choices[0].message.content)
    return decision

if __name__ == "__main__":
    decision = evaluate_deployment(
        version=os.environ["VERSION"],
        commits_count=os.environ["COMMITS"],
        changed_files=os.environ.get("CHANGES", ""),
        test_results=os.environ.get("TEST_RESULTS", "{}")
    )

    # Set GitHub Actions outputs
    with open(os.environ["GITHUB_OUTPUT"], "a") as f:
        f.write(f"approved={decision['approved']}\n")
        f.write(f"reason={decision['reason']}\n")

    print(json.dumps(decision, indent=2))

{
  "approved": true,
  "confidence": 0.85,
  "reason": "All tests pass with 92% coverage. Changes are scoped to non-critical modules.",
  "risks": ["Updated dependency cryptography to 42.0.0 - verify compatibility"],
  "suggestions": ["Monitor error rates for 30 minutes post-deployment"]
}

def ai_incident_responder(alert_data):
    """Analyze an alert and execute automated response actions."""
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{
            "role": "system",
            "content": ""]
You are an on-call SRE. Analyze the alert data and generate a response plan.

Output JSON:
{
  "severity": "critical|high|medium|low",
  "root_cause": "likely root cause",
  "immediate_action": "what to do right now",
  "runbook": "path to relevant runbook",
  "notify": ["team members to ping"],
  "auto_remediate": true/false
}
"""
        }, {
            "role": "user",
            "content": f"Alert: {json.dumps(alert_data)}"
        }]
    )

    plan = json.loads(response.choices[0].message.content)
    return plan

# Example alert from monitoring system
alert = {
    "type": "latency_spike",
    "service": "api-gateway",
    "current_p99": 5200,
    "baseline_p99": 200,
    "threshold": 1000,
    "affected_users": 14500,
    "timestamp": "2026-06-22T14:30:00Z"
}

plan = ai_incident_responder(alert)
print(json.dumps(plan, indent=2))

# Execute auto-remediation if appropriate
if plan.get("auto_remediate"):
    print(f"Auto-remediating: {plan['immediate_action']}")
    # Execute the remediation step
    subprocess.run(["kubectl", "rollout", "restart", "deployment/api-gateway"])

{
  "severity": "critical",
  "root_cause": "Likely a recent deployment or traffic surge overwhelming the API gateway pods",
  "immediate_action": "Restart the api-gateway deployment and scale from 3 to 5 replicas",
  "runbook": "runbooks/api-gateway-latency.md",
  "notify": ["@sre-team", "@backend-owners"],
  "auto_remediate": true
}

import boto3

def analyze_cloud_costs(aws_profile="production"):
    """Analyze AWS cost data and suggest optimizations using AI."""
    session = boto3.Session(profile_name=aws_profile)
    ce = session.client("ce")

    # Get cost data
    response = ce.get_cost_and_usage(
        TimePeriod={
            "Start": "2026-05-01",
            "End": "2026-06-01"
        },
        Granularity="DAILY",
        Metrics=["BlendedCost"],
        GroupBy=[{"Type": "DIMENSION", "Key": "SERVICE"}]
    )

    cost_data = response["ResultsByTime"]
    service_costs = {}
    for day in cost_data:
        for group in day["Groups"]:
            service = group["Keys"][0]
            cost = float(group["Metrics"]["BlendedCost"]["Amount"])
            service_costs[service] = service_costs.get(service, 0) + cost

    # Ask AI for optimization suggestions
    cost_summary = "\n".join([
        f"{svc}: ${cost:.2f}]
        for svc, cost in sorted(service_costs.items(), key=lambda x: x[1], reverse=True)[:10]
    ])

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{
            "role": "system",
            "content": ""]
You are a cloud cost optimization expert. Analyze the spending data and suggest:
1. Top 3 cost-saving opportunities
2. Estimated monthly savings
3. Implementation complexity (easy/medium/hard)
4. Specific AWS services or configurations to change

Output as a markdown table.
"""
        }, {
            "role": "user",
            "content": f"Monthly cloud costs by service:\n\n{cost_summary}"
        }]
    )

    print(response.choices[0].message.content)

analyze_cloud_costs()

def self_healing_loop(check_interval=60):
    """Monitor and auto-heal infrastructure issues using AI decisions."""
    import time

    while True:
        # Collect system metrics
        metrics = {
            "cpu": get_average_cpu_usage(),
            "memory": get_memory_usage(),
            "disk": get_disk_usage(),
            "error_rate": get_error_rate(),
            "pod_status": get_pod_status()
        }

        # Ask AI if action is needed
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[{
                "role": "system",
                "content": ""]
Analyze these infrastructure metrics. If there is an issue, determine:
- What is wrong
- How critical it is
- What action to take (scale, restart, cleanup, ignore)

Respond with JSON: {"issue_found": bool, "severity": "low|medium|high",
"action": "action_to_take", "reason": "explanation"}
"""
            }, {
                "role": "user",
                "content": f"Current metrics: {json.dumps(metrics)}"
            }]
        )

        decision = json.loads(response.choices[0].message.content)

        if decision.get("issue_found"):
            print(f"[{time.ctime()}] Issue detected: {decision['reason']}")
            print(f"Taking action: {decision['action']}")
            execute_remediation(decision["action"])
        else:
            print(f"[{time.ctime()}] System healthy")

        time.sleep(check_interval)

def generate_pipeline_config(project_type, language, frameworks):
    """Generate a CI/CD pipeline configuration using AI."""
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{
            "role": "user",
            "content": f""]
Generate a complete GitHub Actions workflow for:
- Project type: {project_type}
- Language: {language}
- Frameworks: {frameworks}

Include stages for: lint, test, build, security scan, deploy.
Use current best practices and actions versions.
Output only valid YAML.
"""
        }]
    )

    with open(".github/workflows/ci-cd.yml", "w") as f:
        f.write(response.choices[0].message.content)

    print("Generated CI/CD pipeline configuration")

generate_pipeline_config(
    "web application",
    "TypeScript",
    "Next.js, Jest, <a href="/testing-qa/playwright/">Playwright</a>"
)