---
name: cloud-native
description: Generic Cloud-Native Deployment and Infrastructure as Code patterns for 2025. Provides comprehensive implementation strategies for multi-cloud deployments, GitOps workflows, progressive delivery, and platform engineering. Framework-agnostic approach supporting any cloud provider, deployment tool, and orchestration platform.
license: MIT
---

# 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

```python
# 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,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE
        )

        stdout, stderr = await process.communicate()

        return {
            "returncode": process.returncode,
            "stdout": stdout.decode(),
            "stderr": stderr.decode()
        }

    def _get_required_providers(self, provider: CloudProvider) -> Dict[str, Any]:
        """Get required providers configuration"""
        providers = {}

        if provider == CloudProvider.AWS:
            providers["aws"] = {"source": "hashicorp/aws", "version": "~> 5.0"}
        elif provider == CloudProvider.AZURE:
            providers["azurerm"] = {"source": "hashicorp/azurerm", "version": "~> 3.0"}
        elif provider == CloudProvider.GCP:
            providers["google"] = {"source": "hashicorp/google", "version": "~> 5.0"}

        return providers

    def _get_backend_config(self) -> Dict[str, Any]:
        """Get Terraform backend configuration"""
        if self.state_backend:
            return {"s3": self.state_backend}
        return {}

    def _get_provider_config(self, config: InfrastructureConfig) -> Dict[str, Any]:
        """Get provider-specific configuration"""
        base_config = {"region": config.region}

        # Add provider-specific settings
        if config.provider == CloudProvider.AWS:
            base_config["default_tags"] = {
                "tags": {
                    "Environment": config.environment,
                    "ManagedBy": "terraform",
                    "Project": config.name
                }
            }

        return base_config

    def _convert_resources(self, resources: List[Resource]) -> Dict[str, Any]:
        """Convert generic resources to Terraform format"""
        terraform_resources = {}

        for resource in resources:
            resource_key = f"{resource.type}_{resource.name}"

            resource_config = {
                "provider": resource.provider.value if resource.provider else None,
                **resource.properties
            }

            if resource.tags:
                resource_config["tags"] = resource.tags

            terraform_resources[resource_key] = resource_config

        return terraform_resources

    def _infer_type(self, value: Any) -> str:
        """Infer Terraform variable type from value"""
        if isinstance(value, bool):
            return "bool"
        elif isinstance(value, int):
            return "number"
        elif isinstance(value, list):
            return "list(string)"
        elif isinstance(value, dict):
            return "map(string)"
        else:
            return "string"

class PulumiBackend(IaCBackend):
    """Pulumi backend implementation"""

    def __init__(self, working_dir: Path, language: str = "python"):
        self.working_dir = working_dir
        self.language = language

    async def initialize(self, config: InfrastructureConfig) -> None:
        """Initialize Pulumi"""
        # Generate Pulumi program
        await self._generate_pulumi_program(config)

        # Run pulumi stack init
        cmd = ["pulumi", "stack", "init", f"{config.name}-{config.environment}"]
        result = await self._run_command(cmd, cwd=self.working_dir)

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

    async def plan(self, config: InfrastructureConfig) -> Dict[str, Any]:
        """Preview Pulumi changes"""
        cmd = ["pulumi", "preview", "--json"]
        result = await self._run_command(cmd, cwd=self.working_dir)

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

    async def apply(self, config: InfrastructureConfig) -> Dict[str, Any]:
        """Apply Pulumi changes"""
        cmd = ["pulumi", "up", "--yes", "--json"]
        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 Pulumi resources"""
        cmd = ["pulumi", "destroy", "--yes", "--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 Pulumi stack output"""
        cmd = ["pulumi", "stack", "output", "--json"]
        result = await self._run_command(cmd, cwd=self.working_dir)

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

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

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

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

        return errors

    async def _generate_pulumi_program(self, config: InfrastructureConfig) -> None:
        """Generate Pulumi program based on language"""
        if self.language == "python":
            await self._generate_python_program(config)
        elif self.language == "typescript":
            await self._generate_typescript_program(config)

    async def _generate_python_program(self, config: InfrastructureConfig) -> None:
        """Generate Python Pulumi program"""
        program = f'''"""
Pulumi program for {config.name} - {config.environment}
"""

import pulumi
import pulumi_{config.provider.value} as provider

# Configuration
config = pulumi.Config()

# Resources
'''

        # Generate resources
        for resource in config.resources:
            provider_module = f"provider.{resource.type.replace('_', '.')}"
            program += f'''
{resource.name}_resource = {provider_module}.{resource.title()}(
    "{resource.name}",
'''

            # Add properties
            for key, value in resource.properties.items():
                if isinstance(value, str):
                    program += f'    {key}="{value}",\n'
                else:
                    program += f'    {key}={value},\n'

            # Add tags if present
            if resource.tags:
                program += f'    tags={resource.tags},\n'

            program += ')\n\n'

        # Generate outputs
        for key, value in config.outputs.items():
            program += f'pulumi.export("{key}", {value})\n'

        # Write program file
        with open(self.working_dir / "__main__.py", "w") as f:
            f.write(program)

        # Generate requirements.txt
        with open(self.working_dir / "requirements.txt", "w") as f:
            f.write(f"pulumi>={config.provider.value}\n")

    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,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE
        )

        stdout, stderr = await process.communicate()

        return {
            "returncode": process.returncode,
            "stdout": stdout.decode(),
            "stderr": stderr.decode()
        }

# Factory for creating IaC backends
class IaCFactory:
    """Factory for creating IaC backends"""

    @staticmethod
    def create(
        provider: IaCProvider,
        working_dir: Path,
        **kwargs
    ) -> IaCBackend:
        """Create IaC backend instance"""

        working_dir.mkdir(parents=True, exist_ok=True)

        if provider == IaCProvider.TERRAFORM:
            return TerraformBackend(working_dir)
        elif provider == IaCProvider.PULUMI:
            return PulumiBackend(
                working_dir,
                language=kwargs.get("language", "python")
            )
        else:
            raise ValueError(f"Unsupported IaC provider: {provider}")
```

### Multi-Cloud Resource Manager

```python
# iac/multicloud/manager.py
from typing import Dict, List, Any, Optional
from dataclasses import dataclass
import asyncio
from ..core.abstraction import (
    InfrastructureConfig, Resource, CloudProvider, IaCBackend
)

@dataclass
class CloudDeployment:
    """Cloud deployment configuration"""
    provider: CloudProvider
    region: str
    environment: str
    resources: List[Resource]
    priority: int = 0  # Higher priority = primary deployment
    enabled: bool = True

class MultiCloudManager:
    """Manages deployments across multiple cloud providers"""

    def __init__(self):
        self.deployments: Dict[str, CloudDeployment] = {}
        self.backends: Dict[str, IaCBackend] = {}
        self.global_resources: List[Resource] = []
        self.shared_state: Dict[str, Any] = {}

    def add_deployment(self, name: str, deployment: CloudDeployment) -> None:
        """Add a cloud deployment"""
        self.deployments[name] = deployment

    def add_global_resource(self, resource: Resource) -> None:
        """Add a global resource (DNS, CDN, etc.)"""
        self.global_resources.append(resource)

    async def deploy_all(self) -> Dict[str, Any]:
        """Deploy to all configured clouds"""
        results = {}

        # Sort by priority (primary deployments first)
        sorted_deployments = sorted(
            self.deployments.items(),
            key=lambda x: x[1].priority,
            reverse=True
        )

        for name, deployment in sorted_deployments:
            if not deployment.enabled:
                continue

            print(f"Deploying to {name} ({deployment.provider.value})...")

            try:
                result = await self._deploy_single(name, deployment)
                results[name] = result

                # Store shared outputs
                if result["success"]:
                    self.shared_state[name] = result.get("outputs", {})

            except Exception as e:
                results[name] = {
                    "success": False,
                    "error": str(e)
                }
                print(f"Failed to deploy to {name}: {e}")

        return results

    async def _deploy_single(
        self,
        name: str,
        deployment: CloudDeployment
    ) -> Dict[str, Any]:
        """Deploy to a single cloud provider"""

        # Create IaC config
        config = InfrastructureConfig(
            name=name,
            environment=deployment.environment,
            provider=deployment.provider,
            region=deployment.region,
            resources=self._merge_resources(deployment),
            variables=self._get_variables(deployment)
        )

        # Get or create backend
        if name not in self.backends:
            from ..core.abstraction import IaCFactory, IaCProvider
            backend = IaCFactory.create(
                IaCProvider.TERRAFORM,
                Path(f"iac/{name}")
            )
            self.backends[name] = backend

        backend = self.backends[name]

        # Initialize and deploy
        await backend.initialize(config)

        # Plan
        plan = await backend.plan(config)