Cloud-Native Deployment & Infrastructure Patterns

This skill provides comprehensive patterns for implementing cloud-native deployments using modern Infrastructure as Code, GitOps workflows, and progressive delivery strategies. The patterns are designed to be framework-agnostic and applicable across any cloud provider or orchestration platform.

When to Use This Skill

Use this skill when you need to:

• Implement Infrastructure as Code with any provider (Terraform, Pulumi, CDK)
• Build GitOps workflows with Argo CD, Flux, or similar tools
• Create multi-cloud or hybrid deployment strategies
• Implement progressive delivery (canary, blue-green)
• Build internal developer platforms
• Set up policy as code and compliance automation
• Implement zero-trust security architectures
• Create disaster recovery and failover strategies
• Optimize costs across multiple providers
• Build observability into deployments

1. Infrastructure as Code Patterns

Generic IaC Abstraction Layer

# iac/core/abstraction.py
from abc import ABC, abstractmethod
from typing import Dict, List, Any, Optional, Union
from dataclasses import dataclass, field
from enum import Enum
from pathlib import Path
import json
import yaml
import asyncio
from datetime import datetime

class IaCProvider(str, Enum):
    """Supported IaC providers"""
    TERRAFORM = "terraform"
    PULUMI = "pulumi"
    CDK = "cdk"
    CROSSPLANE = "crossplane"

class CloudProvider(str, Enum):
    """Cloud providers"""
    AWS = "aws"
    AZURE = "azure"
    GCP = "gcp"
    DIGITALOCEAN = "digitalocean"
    ON_PREM = "on-prem"

@dataclass
class Resource:
    """Generic resource definition"""
    type: str
    name: str
    properties: Dict[str, Any] = field(default_factory=dict)
    dependencies: List[str] = field(default_factory=list)
    provider: Optional[CloudProvider] = None
    tags: Dict[str, str] = field(default_factory=dict)

@dataclass
class InfrastructureConfig:
    """Infrastructure configuration"""
    name: str
    environment: str
    provider: CloudProvider
    region: str
    resources: List[Resource] = field(default_factory=list)
    variables: Dict[str, Any] = field(default_factory=dict)
    outputs: Dict[str, str] = field(default_factory=dict)

class IaCBackend(ABC):
    """Abstract IaC backend interface"""

    @abstractmethod
    async def initialize(self, config: InfrastructureConfig) -> None:
        """Initialize IaC backend"""
        pass

    @abstractmethod
    async def plan(self, config: InfrastructureConfig) -> Dict[str, Any]:
        """Generate execution plan"""
        pass

    @abstractmethod
    async def apply(self, config: InfrastructureConfig) -> Dict[str, Any]:
        """Apply infrastructure changes"""
        pass

    @abstractmethod
    async def destroy(self, config: InfrastructureConfig) -> Dict[str, Any]:
        """Destroy infrastructure"""
        pass

    @abstractmethod
    async def output(self, config: InfrastructureConfig, key: str) -> Any:
        """Get output value"""
        pass

    @abstractmethod
    async def validate(self, config: InfrastructureConfig) -> List[str]:
        """Validate configuration"""
        pass

class TerraformBackend(IaCBackend):
    """Terraform backend implementation"""

    def __init__(self, working_dir: Path):
        self.working_dir = working_dir
        self.state_backend: Optional[str] = None

    async def initialize(self, config: InfrastructureConfig) -> None:
        """Initialize Terraform"""
        # Generate Terraform files
        await self._generate_terraform_files(config)

        # Run terraform init
        cmd = ["terraform", "init"]
        result = await self._run_command(cmd, cwd=self.working_dir)

        if result["returncode"] != 0:
            raise RuntimeError(f"Terraform init failed: {result['stderr']}")

    async def plan(self, config: InfrastructureConfig) -> Dict[str, Any]:
        """Generate Terraform plan"""
        cmd = ["terraform", "plan", "-out=tfplan", "-json"]
        result = await self._run_command(cmd, cwd=self.working_dir)

        return {
            "success": result["returncode"] == 0,
            "plan_file": self.working_dir / "tfplan",
            "output": result["stdout"],
            "changes": json.loads(result["stdout"]) if result["stdout"] else {}
        }

    async def apply(self, config: InfrastructureConfig) -> Dict[str, Any]:
        """Apply Terraform changes"""
        cmd = ["terraform", "apply", "-auto-approve", "-json", "tfplan"]
        result = await self._run_command(cmd, cwd=self.working_dir)

        return {
            "success": result["returncode"] == 0,
            "output": result["stdout"],
            "applied_at": datetime.utcnow().isoformat()
        }

    async def destroy(self, config: InfrastructureConfig) -> Dict[str, Any]:
        """Destroy Terraform resources"""
        cmd = ["terraform", "destroy", "-auto-approve", "-json"]
        result = await self._run_command(cmd, cwd=self.working_dir)

        return {
            "success": result["returncode"] == 0,
            "output": result["stdout"],
            "destroyed_at": datetime.utcnow().isoformat()
        }

    async def output(self, config: InfrastructureConfig, key: str) -> Any:
        """Get Terraform output"""
        cmd = ["terraform", "output", "-json", key]
        result = await self._run_command(cmd, cwd=self.working_dir)

        if result["returncode"] == 0:
            return json.loads(result["stdout"])
        return None

    async def validate(self, config: InfrastructureConfig) -> List[str]:
        """Validate Terraform configuration"""
        errors = []

        # Run terraform validate
        cmd = ["terraform", "validate"]
        result = await self._run_command(cmd, cwd=self.working_dir)

        if result["returncode"] != 0:
            errors.append(f"Validation failed: {result['stderr']}")

        # Run terraform fmt check
        cmd = ["terraform", "fmt", "-check", "-diff"]
        result = await self._run_command(cmd, cwd=self.working_dir)

        if result["returncode"] != 0:
            errors.append("Formatting check failed - run terraform fmt")

        # Check for required variables
        cmd = ["terraform", "validate", "-json"]
        result = await self._run_command(cmd, cwd=self.working_dir)

        return errors

    async def _generate_terraform_files(self, config: InfrastructureConfig) -> None:
        """Generate Terraform configuration files"""
        # Generate main.tf
        main_tf = {
            "terraform": {
                "required_providers": self._get_required_providers(config.provider),
                "backend": self._get_backend_config()
            },
            "provider": {config.provider.value: self._get_provider_config(config)},
            "resource": self._convert_resources(config.resources)
        }

        with open(self.working_dir / "main.tf", "w") as f:
            yaml.dump(main_tf, f, sort_keys=False)

        # Generate variables.tf
        if config.variables:
            variables_tf = {
                "variable": {
                    k: {"type": self._infer_type(v), "default": v}
                    for k, v in config.variables.items()
                }
            }

            with open(self.working_dir / "variables.tf", "w") as f:
                yaml.dump(variables_tf, f, sort_keys=False)

        # Generate outputs.tf
        if config.outputs:
            outputs_tf = {
                "output": {
                    k: {"value": v}
                    for k, v in config.outputs.items()
                }
            }

            with open(self.working_dir / "outputs.tf", "w") as f:
                yaml.dump(outputs_tf, f, sort_keys=False)

    async def _run_command(self, cmd: List[str], cwd: Path) -> Dict[str, Any]:
        """Run command and return result"""
        process = await asyncio.create_subprocess_exec(
            *cmd,
            cwd=cwd