Self-Service Infrastructure

Self-service infrastructure empowers developers to provision and manage resources without depending on operations teams. This accelerates development cycles while maintaining security, compliance, and cost controls through automation and guardrails.

The Self-Service Model

Self-service infrastructure provides developers with controlled access to provision resources on demand.

Traditional Infrastructure Provisioning

Manual processes create bottlenecks:

sequenceDiagram
    participant D as Developer
    participant T as Ticket System
    participant O as Ops Team
    participant C as Cloud Provider
    
    D->>T: Create infrastructure request ticket
    Note over T: Wait in queue (2-3 days)
    O->>T: Review ticket
    O->>O: Manual validation
    O->>C: Provision resources
    C-->>O: Resources created
    O->>T: Update ticket with details
    T-->>D: Resources ready
    
    Note over D,C: Total time: 3-7 days

Self-Service Approach

Automated provisioning with guardrails:

sequenceDiagram
    participant D as Developer
    participant P as Platform API
    participant V as Validation
    participant I as IaC Engine
    participant C as Cloud Provider
    
    D->>P: Request infrastructure via portal/CLI
    P->>V: Validate request against policies
    V-->>P: Approved
    P->>I: Generate and apply IaC
    I->>C: Provision resources
    C-->>I: Resources created
    I-->>P: Deployment complete
    P-->>D: Resources ready with endpoints
    
    Note over D,C: Total time: 2-5 minutes

Self-Service Patterns

Several patterns enable safe, scalable self-service infrastructure.

Template-Based Provisioning

Predefined templates ensure consistency:

# infrastructure-template.yaml
apiVersion: platform.example.com/v1
kind: InfrastructureTemplate
metadata:
  name: web-service-standard
  description: Standard web service with database and cache
spec:
  parameters:
    - name: serviceName
      type: string
      required: true
      pattern: "^[a-z][a-z0-9-]*$"
    
    - name: environment
      type: string
      enum: [dev, staging, production]
      required: true
    
    - name: instanceType
      type: string
      default: medium
      enum: [small, medium, large]
      
    - name: databaseType
      type: string
      default: postgres
      enum: [postgres, mysql, mongodb]
  
  resources:
    - type: kubernetes.namespace
      name: "${serviceName}-${environment}"
      
    - type: database.instance
      name: "${serviceName}-db"
      engine: "${databaseType}"
      size: "${instanceType}"
      backup:
        enabled: true
        retention: 7
      
    - type: cache.redis
      name: "${serviceName}-cache"
      size: small
      
    - type: storage.s3bucket
      name: "${serviceName}-assets"
      encryption: true
      lifecycle:
        deleteAfter: 90
  
  networking:
    ingress:
      enabled: true
      host: "${serviceName}.${environment}.example.com"
      tls: true
    
    egress:
      allowList:
        - "*.example.com"
        - "api.thirdparty.com"
  
  monitoring:
    - type: dashboard
      metrics: [cpu, memory, requests, errors]
    - type: alerts
      critical: [high_error_rate, high_latency]

Using the template:

# CLI usage
platform deploy \
  --template web-service-standard \
  --param serviceName=payment-api \
  --param environment=production \
  --param instanceType=large \
  --param databaseType=postgres

Policy-Based Guardrails

Enforce organizational requirements automatically:

# Policy enforcement engine
from enum import Enum
from typing import List, Dict

class PolicyLevel(Enum):
    ERROR = "error"      # Blocks deployment
    WARNING = "warning"  # Allows with warning
    INFO = "info"        # Informational only

class Policy:
    def __init__(self, name: str, level: PolicyLevel):
        self.name = name
        self.level = level
    
    def evaluate(self, resource: Dict) -> bool:
        raise NotImplementedError

class CostLimitPolicy(Policy):
    def __init__(self, max_monthly_cost: float):
        super().__init__("cost-limit", PolicyLevel.ERROR)
        self.max_monthly_cost = max_monthly_cost
    
    def evaluate(self, resource: Dict) -> bool:
        estimated_cost = self.calculate_cost(resource)
        return estimated_cost <= self.max_monthly_cost
    
    def calculate_cost(self, resource: Dict) -> float:
        # Cost calculation logic
        costs = {
            'small': 50,
            'medium': 150,
            'large': 500
        }
        return costs.get(resource.get('instanceType'), 0)

class EncryptionPolicy(Policy):
    def __init__(self):
        super().__init__("encryption-required", PolicyLevel.ERROR)
    
    def evaluate(self, resource: Dict) -> bool:
        if resource.get('type') == 'database':
            return resource.get('encryption', False) is True
        if resource.get('type') == 'storage':
            return resource.get('encryption', False) is True
        return True

class TaggingPolicy(Policy):
    def __init__(self, required_tags: List[str]):
        super().__init__("required-tags", PolicyLevel.WARNING)
        self.required_tags = required_tags
    
    def evaluate(self, resource: Dict) -> bool:
        tags = resource.get('tags', {})
        return all(tag in tags for tag in self.required_tags)

class PolicyEngine:
    def __init__(self):
        self.policies = [
            CostLimitPolicy(max_monthly_cost=1000),
            EncryptionPolicy(),
            TaggingPolicy(required_tags=['owner', 'cost-center', 'environment'])
        ]
    
    def validate(self, resource: Dict) -> Dict:
        results = {
            'allowed': True,
            'errors': [],
            'warnings': []
        }
        
        for policy in self.policies:
            if not policy.evaluate(resource):
                if policy.level == PolicyLevel.ERROR:
                    results['allowed'] = False
                    results['errors'].append(f"Policy violation: {policy.name}")
                elif policy.level == PolicyLevel.WARNING:
                    results['warnings'].append(f"Policy warning: {policy.name}")
        
        return results

Policy enforcement in action:

graph TD
    A[Infrastructure Request] --> B{Policy Validation}
    
    B -->|Cost Check| C{Within Budget?}
    C -->|No| D[REJECT: Cost Limit Exceeded]
    C -->|Yes| E{Encryption Check}
    
    E -->|Missing| F[REJECT: Encryption Required]
    E -->|Present| G{Tagging Check}
    
    G -->|Missing| H[WARN: Tags Required]
    G -->|Complete| I[APPROVE]
    H --> I
    
    I --> J[Provision Infrastructure]
    
    style D fill:#ffcccc
    style F fill:#ffcccc
    style H fill:#fff4cc
    style I fill:#ccffcc

Resource Lifecycle Management

Automated cleanup prevents waste:

// Automatic resource lifecycle management
package platform

import (
    "time"
)

type ResourceLifecycle struct {
    CreatedAt      time.Time
    ExpiresAt      *time.Time
    LastAccessed   time.Time
    Owner          string
    Environment    string
}

type LifecyclePolicy struct {
    Environment     string
    MaxAge          time.Duration
    IdleTimeout     time.Duration
    NotifyBefore    time.Duration
}

var defaultPolicies = []LifecyclePolicy{
    {
        Environment:  "dev",
        MaxAge:       30 * 24 * time.Hour,  // 30 days
        IdleTimeout:  7 * 24 * time.Hour,   // 7 days
        NotifyBefore: 3 * 24 * time.Hour,   // 3 days
    },
    {
        Environment:  "staging",
        MaxAge:       90 * 24 * time.Hour,  // 90 days
        IdleTimeout:  14 * 24 * time.Hour,  // 14 days
        NotifyBefore: 7 * 24 * time.Hour,   // 7 days
    },
    {
        Environment:  "production",
        MaxAge:       0,                     // No automatic deletion
        IdleTimeout:  0,
        NotifyBefore: 0,
    },
}

func (lm *LifecycleManager) EvaluateResource(resource Resource) Action {
    lifecycle := resource.Lifecycle
    policy := lm.getPolicyForEnvironment(lifecycle.Environment)
    
    now := time.Now()
    
    // Check max age
    if policy.MaxAge > 0 {
        age := now.Sub(lifecycle.CreatedAt)
        if age > policy.MaxAge {
            return Action{
                Type:   ActionDelete,
                Reason: "Resource exceeded maximum age",
                When:   now,
            }
        }
        
        // Notify before deletion
        timeUntilExpiry := policy.MaxAge - age
        if timeUntilExpiry < policy.NotifyBefore {
            return Action{
                Type:   ActionNotify,
                Reason: "Resource will expire soon",
                When:   now,
            }
        }
    }
    
    // Check idle timeout
    if policy.IdleTimeout > 0 {
        idleTime := now.Sub(lifecycle.LastAccessed)
        if idleTime > policy.IdleTimeout {
            return Action{
                Type:   ActionDelete,
                Reason: "Resource has been idle too long",
                When:   now,
            }
        }
    }
    
    return Action{Type: ActionNone}
}

Self-Service Infrastructure Components

Implementing self-service requires several integrated systems.

API Layer

RESTful API for infrastructure requests:

// Infrastructure API
import express from 'express';
import { validateRequest, enforceQuota, auditLog } from './middleware';

const app = express();

interface InfrastructureRequest {
  template: string;
  parameters: Record<string, any>;
  requester: string;
  team: string;
}

app.post('/api/v1/infrastructure', 
  validateRequest,
  enforceQuota,
  auditLog,
  async (req, res) => {
    const request: InfrastructureRequest = req.body;
    
    try {
      // 1. Validate against policies
      const validation = await policyEngine.validate(request);
      if (!validation.allowed) {
        return res.status(403).json({
          error: 'Policy validation failed',
          details: validation.errors
        });
      }
      
      // 2. Check quota
      const quota = await quotaService.checkQuota(request.team);
      if (!quota.available) {
        return res.status(429).json({
          error: 'Quota exceeded',
          details: quota
        });
      }
      
      // 3. Generate infrastructure code
      const infraCode = await templateEngine.render(
        request.template,
        request.parameters
      );
      
      // 4. Submit for provisioning
      const deployment = await provisioningEngine.deploy({
        code: infraCode,
        requester: request.requester,
        team: request.team
      });
      
      // 5. Register in catalog
      await catalog.register({
        id: deployment.id,
        owner: request.team,
        resources: deployment.resources
      });
      
      res.status(202).json({
        deploymentId: deployment.id,
        status: 'pending',
        estimatedTime: '2-5 minutes',
        trackingUrl: `/api/v1/deployments/${deployment.id}`
      });
      
    } catch (error) {
      res.status(500).json({
        error: 'Infrastructure provisioning failed',
        message: error.message
      });
    }
  }
);

app.get('/api/v1/deployments/:id', async (req, res) => {
  const deployment = await provisioningEngine.getStatus(req.params.id);
  res.json(deployment);
});

Quota Management

Prevent runaway resource consumption:

# Quota system
from dataclasses import dataclass
from typing import Dict

@dataclass
class Quota:
    team: str
    max_cpu_cores: int
    max_memory_gb: int
    max_storage_gb: int
    max_databases: int
    max_cost_monthly: float

@dataclass
class Usage:
    cpu_cores: int
    memory_gb: int
    storage_gb: int
    databases: int
    cost_monthly: float

class QuotaService:
    def __init__(self):
        self.quotas: Dict[str, Quota] = {}
        self.usage: Dict[str, Usage] = {}
    
    def check_quota(self, team: str, requested: Usage) -> Dict:
        """Check if team has quota for requested resources"""
        quota = self.quotas.get(team)
        current = self.usage.get(team, Usage(0, 0, 0, 0, 0.0))
        
        # Calculate new usage
        new_usage = Usage(
            cpu_cores=current.cpu_cores + requested.cpu_cores,
            memory_gb=current.memory_gb + requested.memory_gb,
            storage_gb=current.storage_gb + requested.storage_gb,
            databases=current.databases + requested.databases,
            cost_monthly=current.cost_monthly + requested.cost_monthly
        )
        
        # Check each limit
        violations = []
        if new_usage.cpu_cores > quota.max_cpu_cores:
            violations.append(f"CPU limit: {new_usage.cpu_cores}/{quota.max_cpu_cores}")
        if new_usage.memory_gb > quota.max_memory_gb:
            violations.append(f"Memory limit: {new_usage.memory_gb}/{quota.max_memory_gb}")
        if new_usage.storage_gb > quota.max_storage_gb:
            violations.append(f"Storage limit: {new_usage.storage_gb}/{quota.max_storage_gb}")
        if new_usage.databases > quota.max_databases:
            violations.append(f"Database limit: {new_usage.databases}/{quota.max_databases}")
        if new_usage.cost_monthly > quota.max_cost_monthly:
            violations.append(f"Cost limit: ${new_usage.cost_monthly}/${quota.max_cost_monthly}")
        
        return {
            'available': len(violations) == 0,
            'violations': violations,
            'current_usage': current,
            'requested': requested,
            'quota': quota
        }

Audit Trail

Track all infrastructure changes:

// Audit logging
package audit

import (
    "encoding/json"
    "time"
)

type AuditEvent struct {
    ID          string                 `json:"id"`
    Timestamp   time.Time              `json:"timestamp"`
    Action      string                 `json:"action"`
    Resource    string                 `json:"resource"`
    Requester   string                 `json:"requester"`
    Team        string                 `json:"team"`
    Parameters  map[string]interface{} `json:"parameters"`
    Result      string                 `json:"result"`
    IPAddress   string                 `json:"ip_address"`
    UserAgent   string                 `json:"user_agent"`
}

type AuditLogger interface {
    Log(event AuditEvent) error
    Query(filters map[string]string) ([]AuditEvent, error)
}

type DatabaseAuditLogger struct {
    db Database
}

func (l *DatabaseAuditLogger) Log(event AuditEvent) error {
    query := `
        INSERT INTO audit_log (
            id, timestamp, action, resource, 
            requester, team, parameters, result,
            ip_address, user_agent
        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
    `
    
    parametersJSON, _ := json.Marshal(event.Parameters)
    
    return l.db.Exec(query,
        event.ID,
        event.Timestamp,
        event.Action,
        event.Resource,
        event.Requester,
        event.Team,
        parametersJSON,
        event.Result,
        event.IPAddress,
        event.UserAgent,
    )
}

Self-Service Best Practices

Successful self-service infrastructure follows key principles.

Progressive Permissions

Start restrictive, expand based on trust:

# Permission tiers
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: platform-tier-1  # New teams
rules:
  - apiGroups: ["platform.example.com"]
    resources: ["infrastructuretemplates"]
    verbs: ["get", "list"]
    resourceNames: ["web-service-small", "worker-service-small"]
  
  - apiGroups: ["platform.example.com"]
    resources: ["infrastructuredeployments"]
    verbs: ["create", "get", "list"]

---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: platform-tier-2  # Established teams
rules:
  - apiGroups: ["platform.example.com"]
    resources: ["infrastructuretemplates"]
    verbs: ["get", "list"]
    # Access to more templates
  
  - apiGroups: ["platform.example.com"]
    resources: ["infrastructuredeployments"]
    verbs: ["create", "get", "list", "update", "delete"]

---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: platform-tier-3  # Trusted teams
rules:
  - apiGroups: ["platform.example.com"]
    resources: ["*"]
    verbs: ["*"]

Blast Radius Limitation

Isolate failures to prevent cascading issues:

Isolation Strategies:
├── Network Isolation
│   ├── Separate VPCs per team
│   ├── Restricted security groups
│   └── Network policies
├── Resource Isolation
│   ├── Separate namespaces
│   ├── Resource quotas
│   └── Pod security policies
├── Data Isolation
│   ├── Separate database instances
│   ├── Encrypted at rest
│   └── Access controls
└── Blast Radius Controls
    ├── Rate limiting
    ├── Circuit breakers
    └── Automatic rollback

Key Takeaways