Platform Engineering: Building Internal Developer Platforms (IDP) for Scale

# What developers see (simple)
apiVersion: developer.company/v1
kind: Application
metadata:
  name: my-app
spec:
  language: python
  framework: fastapi
  replicas: 3
  domain: myapp.example.com
---
# What the platform creates (complex)
apiVersion: apps/v1
kind: Deployment
metadata:
  name: my-app
  namespace: production
  labels:
    app: my-app
    managed-by: platform-team
spec:
  replicas: 3
  # ... 100+ lines of configuration

# Platform service catalog
from fastapi import FastAPI
from typing import Dict, List

app = FastAPI()

class ServiceTemplate:
    """Define service templates for common patterns"""
    
    def __init__(self, name: str, description: str):
        self.name = name
        self.description = description
        self.components = []
    
    def add_component(self, component_type: str, config: Dict):
        """Add standardized components"""
        self.components.append({
            'type': component_type,
            'config': config
        })
    
    def generate_manifests(self):
        """Generate K8s manifests from template"""
        manifests = []
        for component in self.components:
            manifest = self._render_component(component)
            manifests.append(manifest)
        return manifests
    
    def _render_component(self, component):
        """Render individual component"""
        if component['type'] == 'deployment':
            return self._render_deployment(component['config'])
        elif component['type'] == 'service':
            return self._render_service(component['config'])
        # ... more component types

# Create reusable templates
python_web_template = ServiceTemplate(
    "Python Web Application",
    "Flask/FastAPI web service with standard patterns"
)
python_web_template.add_component('deployment', {
    'image': 'python:3.11',
    'replicas': 3,
    'memory': '256Mi',
    'cpu': '250m'
})
python_web_template.add_component('service', {
    'port': 80,
    'type': 'LoadBalancer'
})
python_web_template.add_component('configmap', {
    'environment': 'production'
})

# IDP Portal Configuration
apiVersion: portal.platform.io/v1
kind: DeveloperPortal
metadata:
  name: company-developer-portal
spec:
  features:
    - name: service-catalog
      enabled: true
      description: "Browse and deploy pre-configured services"
    
    - name: deployment-dashboard
      enabled: true
      description: "Monitor deployed applications"
    
    - name: self-service-secrets
      enabled: true
      description: "Manage application secrets securely"
    
    - name: cost-tracking
      enabled: true
      description: "View resource costs per team"
  
  integrations:
    - name: github
      type: git
      config:
        org: "my-company"
    
    - name: slack
      type: notification
      config:
        webhook: "https://hooks.slack.com/..."
    
    - name: datadog
      type: monitoring
      config:
        api_key: "${SECRET_DATADOG_API_KEY}"

# Policy enforcement in platform
from dataclasses import dataclass
from typing import Callable, List

@dataclass
class Policy:
    name: str
    description: str
    validate: Callable
    on_violation: str  # "warn", "block", "auto-fix"

class PolicyEngine:
    def __init__(self):
        self.policies: List[Policy] = []
    
    def register_policy(self, policy: Policy):
        self.policies.append(policy)
    
    def validate_application(self, app_config: Dict) -> Dict:
        """Validate app against all policies"""
        violations = []
        warnings = []
        auto_fixes = []
        
        for policy in self.policies:
            result = policy.validate(app_config)
            
            if not result['valid']:
                violation = {
                    'policy': policy.name,
                    'message': result['message'],
                    'severity': policy.on_violation
                }
                
                if policy.on_violation == 'warn':
                    warnings.append(violation)
                elif policy.on_violation == 'block':
                    violations.append(violation)
                elif policy.on_violation == 'auto-fix':
                    auto_fixes.append({
                        'policy': policy.name,
                        'fix': result['fix']
                    })
        
        return {
            'valid': len(violations) == 0,
            'violations': violations,
            'warnings': warnings,
            'auto_fixes': auto_fixes
        }

# Define security policies
engine = PolicyEngine()

# Policy 1: Resource limits required
def validate_resource_limits(config):
    if 'resources' not in config:
        return {
            'valid': False,
            'message': 'Resource limits must be specified',
            'fix': {
                'resources': {
                    'requests': {'memory': '256Mi', 'cpu': '250m'},
                    'limits': {'memory': '512Mi', 'cpu': '500m'}
                }
            }
        }
    return {'valid': True}

engine.register_policy(Policy(
    name="enforce-resource-limits",
    description="Enforce CPU and memory limits",
    validate=validate_resource_limits,
    on_violation="auto-fix"
))

# Policy 2: Security scanning required
def validate_security_scanning(config):
    if 'security' not in config or 'scan_enabled' not in config['security']:
        return {
            'valid': False,
            'message': 'Security scanning must be enabled',
            'fix': {
                'security': {
                    'scan_enabled': True,
                    'scan_frequency': 'daily'
                }
            }
        }
    return {'valid': True}

engine.register_policy(Policy(
    name="enforce-security-scanning",
    description="Enforce container image scanning",
    validate=validate_security_scanning,
    on_violation="block"
))

# Policy 3: High availability required for production
def validate_high_availability(config):
    if config.get('environment') == 'production':
        if config.get('replicas', 1) < 3:
            return {
                'valid': False,
                'message': 'Production apps must have at least 3 replicas'
            }
    return {'valid': True}

engine.register_policy(Policy(
    name="enforce-ha-production",
    description="High availability for production",
    validate=validate_high_availability,
    on_violation="warn"
))