# Infrastructure Agents

Agents for managing, provisioning, monitoring, and optimizing infrastructure: cloud, Kubernetes, networking, and security.

---

## Infrastructure Agent Capabilities

| Capability | Description | Tools |
|---|---|---|
| **Provisioning** | Create, update, destroy resources | Terraform, Pulumi, CloudFormation, CDK |
| **Configuration** | Manage config, secrets, policies | Ansible, Chef, Puppet, Salt |
| **Orchestration** | Deploy, scale, rollback | Kubernetes, Nomad, ECS, Docker Swarm |
| **Monitoring** | Metrics, logs, traces, alerts | Prometheus, Grafana, Datadog, Honeycomb |
| **Networking** | DNS, load balancing, mesh, policies | Istio, Linkerd, Cilium, Envoy |
| **Security** | Compliance, scanning, hardening | Trivy, Falco, OPA, Kyverno |
| **Cost Optimization** | Rightsizing, scheduling, spot | Kubecost, CloudHealth, Finout |
| **Disaster Recovery** | Backup, restore, failover | Velero, Restic, Crossplane |
| **GitOps** | Declarative, automated sync | ArgoCD, Flux, Fleet |
| **Policy Enforcement** | Admission, validation, mutation | OPA, Kyverno, Gatekeeper |

---

## Infrastructure Agent Architectures

### 1. Infrastructure Provisioning Agent
```python
class InfraProvisioningAgent:
    def __init__(self):
        self.terraform = TerraformExecutor()
        self.pulumi = PulumiExecutor()
        self.state = StateManager()
    
    async def provision(self, spec: ProvisionSpec) -> ProvisionResult:
        # 1. Generate IaC
        if spec.tool == "terraform":
            code = await self.generate_terraform(spec)
            result = await self.terraform.apply(code)
        elif spec.tool == "pulumi":
            code = await self.generate_pulumi(spec)
            result = await self.pulumi.up(code)
        
        # 2. Validate
        validation = await self.validate(result, spec)
        
        # 3. Post-provision
        await self.post_provision(result, spec)
        
        return ProvisionResult(
            resources=result.resources,
            outputs=result.outputs,
            state=result.state
        )
    
    async def generate_terraform(self, spec: ProvisionSpec) -> str:
        # Use templates + variables
        template = self.load_template(spec.template)
        return template.render(
            provider=spec.provider,
            region=spec.region,
            resources=spec.resources,
            variables=spec.variables
        )
```

### 2. Kubernetes Management Agent
```python
class K8sManagementAgent:
    def __init__(self, kubeconfig: str = None):
        self.k8s = KubernetesClient(kubeconfig)
        self.helm = HelmClient()
        self.kustomize = KustomizeClient()
    
    async def deploy(self, spec: DeploymentSpec) -> DeploymentResult:
        # 1. Build manifests
        if spec.method == "helm":
            manifests = await self.helm.template(spec.chart, spec.values)
        elif spec.method == "kustomize":
            manifests = await self.kustomize.build(spec.overlay)
        else:
            manifests = spec.manifests
        
        # 2. Apply with Server-Side Apply
        result = await self.k8s.apply(manifests, server_side=True)
        
        # 3. Wait for rollout
        await self.wait_for_rollout(spec.namespace, spec.selector)
        
        return DeploymentResult(
            resources=result.resources,
            status=await self.get_status(spec.namespace, spec.selector)
        )
    
    async def scale(self, deployment: str, namespace: str, replicas: int) -> ScaleResult:
        await self.k8s.patch_deployment(deployment, namespace, {"spec": {"replicas": replicas}})
        return await self.wait_for_replicas(deployment, namespace, replicas)
    
    async def rollback(self, deployment: str, namespace: str, revision: int = None) -> RollbackResult:
        if revision:
            await self.k8s.rollback(deployment, namespace, revision)
        else:
            await self.k8s.rollback(deployment, namespace)
        return await self.wait_for_rollback(deployment, namespace)
```

### 3. GitOps Agent
```python
class GitOpsAgent:
    def __init__(self):
        self.argocd = ArgoCDClient()
        self.flux = FluxClient()
    
    async def sync_application(self, app_name: str) -> SyncResult:
        return await self.argocd.sync(app_name)
    
    async def create_application(self, spec: ApplicationSpec) -> Application:
        return await self.argocd.create_app(Application(
            metadata={"name": spec.name, "namespace": "argocd"},
            spec=ApplicationSpec(
                project=spec.project,
                source=ApplicationSource(
                    repo_url=spec.repo_url,
                    path=spec.path,
                    target_revision=spec.revision
                ),
                destination=ApplicationDestination(
                    server=spec.cluster_url,
                    namespace=spec.namespace
                ),
                sync_policy=SyncPolicy(
                    automated=SyncPolicyAutomated(
                        prune=spec.auto_prune,
                        self_heal=spec.self_heal
                    ),
                    sync_options=spec.sync_options
                )
            )
    
    async def get_app_status(self, app_name: str) -> ApplicationStatus:
        return await self.argocd.get_app(app_name)
```

### 4. Network & Service Mesh Agent
```python
class NetworkMeshAgent:
    def __init__(self):
        self.istio = IstioClient()
        self.cilium = CiliumClient()
        self.gateway = GatewayAPIClient()
    
    async def configure_mtls(self, namespace: str, mode: MTLSMode) -> MTLSResult:
        if self.istio:
            return await self.istio.configure_mtls(namespace, mode)
        elif self.cilium:
            return await self.cilium.configure_mtls(namespace, mode)
    
    async def configure_traffic_split(self, spec: TrafficSplitSpec) -> TrafficSplitResult:
        return await self.istio.configure_virtual_service(spec)
    
    async def configure_authorization(self, spec: AuthzSpec) -> AuthzResult:
        return await self.istio.create_authorization_policy(spec)
    
    async def configure_gateway(self, spec: GatewaySpec) -> GatewayResult:
        return await self.gateway.create_gateway(spec)
```

---

## Cloud Provider Agents

### AWS Agent
```python
class AWSAgent:
    def __init__(self, region: str = "us-east-1"):
        self.session = boto3.Session(region_name=region)
        self.cloudformation = self.session.client("cloudformation")
        self.ec2 = self.session.client("ec2")
        self.ecs = self.session.client("ecs")
        self.eks = self.session.client("eks")
        self.lambda_client = self.session.client("lambda")
        self.s3 = self.session.client("s3")
        self.rds = self.session.client("rds")
        self.iam = self.session.client("iam")
    
    async def deploy_stack(self, template: str, params: dict) -> StackResult:
        response = await self.cloudformation.create_stack(
            StackName=f"stack-{uuid4()}",
            TemplateBody=template,
            Parameters=[{"ParameterKey": k, "ParameterValue": v} for k, v in params.items()],
            Capabilities=["CAPABILITY_IAM", "CAPABILITY_NAMED_IAM"]
        )
        return await self.wait_for_stack(response["StackId"])
    
    async def create_eks_cluster(self, spec: EKSSpec) -> ClusterResult:
        # Use eksctl or CloudFormation
        cluster = await self.eks.create_cluster(
            name=spec.name,
            version=spec.version,
            roleArn=spec.role_arn,
            resourcesVpcConfig=spec.vpc_config
        )
        return await self.wait_for_cluster(cluster.name)
```

### GCP Agent
```python
class GCPAgent:
    def __init__(self, project_id: str):
        self.project_id = project_id
        self.compute = compute_v1.InstancesClient()
        self.container = container_v1.ClusterManagerClient()
        self.storage = storage.Client(project=project_id)
        self.run = run_v2.ServicesClient()
    
    async def deploy_cloud_run(self, spec: CloudRunSpec) -> ServiceResult:
        service = await self.run.create_service(
            parent=f"projects/{self.project_id}/locations/{spec.region}",
            service=Service(
                template=RevisionTemplate(
                    containers=[Container(
                        image=spec.image,
                        resources=ResourceRequirements(limits=spec.resources),
                        env=[EnvVar(name=k, value=v) for k, v in spec.env.items()]
                    )],
                    traffic=[TrafficTarget(percent=100, latest_revision=True)]
                ),
            service_id=spec.name
        )
        return await self.wait_for_service(service.name)
```

### Azure Agent
```python
class AzureAgent:
    def __init__(self, subscription_id: str):
        self.subscription_id = subscription_id
        self.credential = DefaultAzureCredential()
        self.compute = ComputeManagementClient(self.credential, subscription_id)
        self.network = NetworkManagementClient(self.credential, subscription_id)
        self.storage = StorageManagementClient(self.credential, subscription_id)
        self.containers = ContainerRegistryManagementClient(self.credential, subscription_id)
    
    async def deploy_aks(self, spec: AKSSpec) -> ClusterResult:
        cluster = await self.containers.managed_clusters.begin_create_or_update(
            resource_group_name=spec.resource_group,
            resource_name=spec.name,
            parameters=ManagedCluster(
                location=spec.location,
                kubernetes_version=spec.version,
                agent_pool_profiles=[AgentPoolProfile(
                    name="default",
                    count=spec.node_count,
                    vm_size=spec.vm_size,
                    mode="System"
                )],
                identity=ManagedClusterIdentity(type="SystemAssigned")
            )
        )
        return await cluster.result()
```

---

## Monitoring & Observability Agents

### Prometheus/Grafana Agent
```python
class PrometheusAgent:
    def __init__(self, prometheus_url: str):
        self.prom = PrometheusClient(prometheus_url)
        self.alertmanager = AlertmanagerClient()
    
    async def query(self, query: str, time_range: TimeRange = None) -> QueryResult:
        if time_range:
            return await self.prom.query_range(query, time_range.start, time_range.end)
        return await self.prom.query(query)
    
    async def create_alert(self, rule: AlertRule) -> AlertResult:
        return await self.alertmanager.create_alert(rule)
    
    async def get_alerts(self) -> list[Alert]:
        return await self.alertmanager.get_alerts()
```

### Logging Agent (Loki/ELK)
```python
class LogAnalysisAgent:
    def __init__(self, loki_url: str = None, elasticsearch_url: str = None):
        self.loki = LokiClient(loki_url) if loki_url else None
        self.es = ElasticsearchClient(elasticsearch_url) if elasticsearch_url else None
    
    async def query_logs(self, query: LogQuery) -> LogResult:
        if self.loki:
            return await self.loki.query_range(query.query, query.start, query.end)
        return await self.es.search(query)
    
    async def detect_anomalies(self, query: LogQuery) -> AnomalyReport:
        logs = await self.query_logs(query)
        return await self.detect_anomalies_in_logs(logs)
```

### Distributed Tracing Agent
```python
class TracingAgent:
    def __init__(self, jaeger_url: str = None, tempo_url: str = None):
        self.jaeger = JaegerClient(jaeger_url) if jaeger_url else None
        self.tempo = TempoClient(tempo_url) if tempo_url else None
    
    async def get_trace(self, trace_id: str) -> Trace:
        if self.jaeger:
            return await self.jaeger.get_trace(trace_id)
        return await self.tempo.get_trace(trace_id)
    
    async def find_traces(self, query: TraceQuery) -> list[Trace]:
        if self.jaeger:
            return await self.jaeger.find_traces(query)
        return await self.tempo.find_traces(query)
    
    async def analyze_latency(self, service: str, window: TimeWindow) -> LatencyAnalysis:
        traces = await self.find_traces(TraceQuery(service=service, start=window.start, end=window.end))
        return self.analyze_latency(traces)
```

---

## Security & Compliance Agents

### Vulnerability Scanning Agent
```python
class VulnerabilityScanner:
    def __init__(self):
        self.trivy = TrivyScanner()
        self.grype = GrypeScanner()
        self.snyk = SnykScanner()
    
    async def scan_image(self, image: str) -> ScanResult:
        results = await asyncio.gather(
            self.trivy.scan(image),
            self.grype.scan(image),
            self.snyk.scan(image)
        )
        return self.merge_results(results)
    
    async def scan_k8s(self, cluster: str) -> K8sScanResult:
        return await self.trivy.scan_k8s(cluster)
    
    async def scan_terraform(self, plan: str) -> TerraformScanResult:
        return await self.trivy.scan_terraform(plan)
```

### Policy Enforcement Agent (OPA/Kyverno)
```python
class PolicyAgent:
    def __init__(self):
        self.opa = OPAClient()
        self.kyverno = KyvernoClient()
    
    async def validate(self, resource: dict, policies: list[Policy]) -> ValidationResult:
        # OPA evaluation
        input_data = {"resource": resource}
        result = await self.opa.evaluate("data.policy.allow", input_data)
        
        return ValidationResult(
            allowed=result.get("allow", False),
            violations=result.get("violations", [])
        )
    
    async def mutate(self, resource: dict, policies: list[Policy]) -> dict:
        return await self.kyverno.mutate(resource, policies)
```

### Compliance Agent
```python
class ComplianceAgent:
    def __init__(self):
        self.frameworks = {
            "SOC2": SOC2Framework(),
            "PCI-DSS": PCIDSSFramework(),
            "HIPAA": HIPAAFramework(),
            "GDPR": GDPRFramework(),
            "CIS": CISBenchmark()
        }
    
    async def assess(self, resources: list[Resource], framework: str) -> ComplianceReport:
        framework = self.frameworks[framework]
        results = await asyncio.gather(*[
            framework.check(resource) for resource in resources
        ])
        
        return ComplianceReport(
            framework=framework,
            passed=sum(1 for r in results if r.passed),
            failed=sum(1 for r in results if not r.passed),
            findings=[r for r in results if not r.passed],
            score=sum(r.score for r in results) / len(results)
        )
```

---

## Cost Optimization Agent
```python
class CostOptimizationAgent:
    def __init__(self):
        self.providers = {
            "aws": AWSCostExplorer(),
            "gcp": GCPCostExplorer(),
            "azure": AzureCostManagement(),
            "k8s": KubecostClient()
        }
    
    async def analyze(self, scope: CostScope) -> CostAnalysis:
        provider = self.providers[scope.provider]
        
        # 1. Current spend
        current = await provider.get_spend(scope.time_range)
        
        # 2. Identify waste
        waste = await self.identify_waste(current)
        
        # 3. Rightsizing
        rightsizing = await self.rightsizing_recommendations(current)
        
        # 4. Spot/reserved optimization
        savings = await self.optimization_opportunities(current)
        
        # 5. Forecast
        forecast = await provider.forecast(next_months=3)
        
        return CostAnalysis(
            current_spend=current.total,
            waste=waste,
            rightsizing=rightsizing,
            savings_opportunities=savings,
            forecast=forecast,
            recommendations=self.generate_recommendations(waste, savings)
        )
    
    async def identify_waste(self, spend: SpendData) -> WasteReport:
        waste = []
        for resource in spend.resources:
            if resource.utilization < 0.1:
                waste.append(WasteItem(
                    resource=resource.id,
                    type="idle",
                    monthly_cost=resource.cost,
                    recommendation="Terminate or downsize"
                ))
            elif resource.utilization < 0.3:
                waste.append(WasteItem(
                    resource=resource.id,
                    type="underutilized",
                    monthly_cost=resource.cost * 0.7,
                    recommendation="Downsize or use spot"
                ))
        return WasteReport(items=waste)
```

---

## Disaster Recovery Agent
```python
class DisasterRecoveryAgent:
    def __init__(self):
        self.velero = VeleroClient()
        self.restic = ResticClient()
        self.crossplane = CrossplaneClient()
    
    async def backup(self, spec: BackupSpec) -> BackupResult:
        if spec.target == "k8s":
            return await self.velero.backup(spec)
        elif spec.target == "vm":
            return await self.restic.backup_vm(spec)
        elif spec.target == "database":
            return await self.backup_database(spec)
    
    async def restore(self, spec: RestoreSpec) -> RestoreResult:
        if spec.source == "k8s":
            return await self.velero.restore(spec)
        elif spec.source == "vm":
            return await self.restic.restore_vm(spec)
    
    async def test_dr(self, spec: DRTestSpec) -> DRTestResult:
        # 1. Create isolated test environment
        test_env = await self.create_test_environment()
        
        # 2. Restore
        restore = await self.restore(RestoreSpec(
            backup=spec.backup_id,
            target=test_env
        ))
        
        # 3. Validate
        validation = await self.validate_restore(restore, spec.validation_checks)
        
        # 4. Cleanup
        await self.cleanup_test_environment(test_env)
        
        return DRTestResult(
            restored=restore.success,
            validation=validation,
            rto=restore.duration,
            rpo=spec.rpo
        )
```

---

## Related Notes

- [[DevOps/CI-CD-Best-Practices.md]] — CI/CD pipelines
- [[DevOps/Observability.md]] — Monitoring, logging, tracing
- [[DevOps/Infrastructure-as-Code.md]] — IaC patterns
- [[DevOps/Git-Workflow.md]] — GitOps
- [[ToolUse/Function-Calling.md]] — Infrastructure as tools
- [[ToolUse/MCP-Protocol.md]] — MCP for infrastructure
- [[ToolUse/API-Integration.md]] — Cloud APIs
- [[Security/Supply-Chain-Security.md]] — Supply chain security
- [[Security/Secure-Coding.md]] — Secure infrastructure code
- [[Production/Deployment-Architecture.md]] — Deployment patterns
- [[Production/Scaling-Strategies.md]] — Scaling strategies
- [[Production/Reliability.md]] — Reliability patterns
- [[Production/Security.md]] — Security hardening
- [[Production/Compliance.md]] — Compliance automation
- [[Production/Cost-Optimization.md]] — Cost optimization
- [[Production/Reliability.md]] — DR, backup
- [[Production/Deployment-Architecture.md]] — Deployment architecture
- [[Production/Scaling-Strategies.md]] — Auto-scaling
- [[Security/Supply-Chain-Security.md]] — Supply chain
- [[Security/Secure-Coding.md]] — Secure IaC
- [[Production/Deployment-Architecture.md]] — Deployment patterns
- [[Frontiers/Research-Directions.md]] — Infrastructure research

---

*Sources: Terraform, Pulumi, Crossplane, Kubernetes, Helm, Kustomize, ArgoCD, Flux, Istio, Linkerd, Cilium, Envoy, Gateway API, Prometheus, Grafana, Loki, Tempo, Jaeger, OpenTelemetry, Trivy, Grype, Syft, OPA, Kyverno, Gatekeeper, Falco, Velero, Restic, Crossplane, Kubecost, CloudHealth, Finout, AWS Cost Explorer, GCP Billing, Azure Cost Management, CloudFormation, CDK, eksctl, kops, kubeadm, kind, k3s, k0s, Talos, Bottlerocket, Flatcar, Container Linux, eBPF (Cilium, Tetragon), Service Mesh (Istio, Linkerd, Consul Connect), CNCF Landscape.*