# Configuration Management: Runtime Config Updates

# Configuration Management: Runtime Config Updates & Feature Toggles

## Problem

Applications need to:
- **Update configurations without redeployment** (downtime-free changes)
- **Control feature availability dynamically** (gradual rollouts, A/B testing)
- **Manage environment-specific settings** (dev/staging/prod)
- **Handle configuration drift** (consistency across instances)
- **Enable quick rollbacks** (revert bad configs instantly)

Traditional hardcoded configs require rebuilds and redeployment, causing delays and risk.

---

## Solution Architecture

### Core Components

1. **Configuration Store** - Centralized source of truth (database, file, service)
2. **Feature Toggle Engine** - Evaluates rules and conditions
3. **Cache Layer** - Local caching with TTL for performance
4. **Change Propagation** - Push/pull mechanisms for updates
5. **Audit Trail** - Track all configuration changes
6. **Client SDK** - Application integration layer

### Key Patterns

- **Remote Config Service**: Centralized API for config retrieval
- **Local Caching**: Reduce latency and dependency on remote service
- **Gradual Rollout**: Percentage-based or user-segment targeting
- **Kill Switches**: Emergency disable for problematic features
- **Versioning**: Track config versions for rollback capability

---

## Code Implementation

### 1. Feature Toggle Engine (Core)

```python
from enum import Enum
from typing import Any, Dict, List, Optional
from dataclasses import dataclass
from datetime import datetime
import json

class ToggleType(Enum):
    BOOLEAN = "boolean"
    PERCENTAGE = "percentage"
    USER_SEGMENT = "user_segment"
    TIME_WINDOW = "time_window"
    CUSTOM = "custom"

@dataclass
class ToggleRule:
    """Represents a single toggle rule"""
    id: str
    name: str
    type: ToggleType
    enabled: bool
    value: Any
    conditions: Dict[str, Any]
    created_at: datetime
    updated_at: datetime
    version: int

class FeatureToggleEngine:
    """Evaluates feature toggles with various conditions"""
    
    def __init__(self):
        self.toggles: Dict[str, ToggleRule] = {}
        self.audit_log: List[Dict] = []
    
    def register_toggle(self, rule: ToggleRule) -> None:
        """Register a new feature toggle"""
        self.toggles[rule.id] = rule
        self._log_audit("REGISTER", rule.id, rule)
    
    def is_enabled(
        self,
        toggle_id: str,
        user_id: Optional[str] = None,
        context: Optional[Dict[str, Any]] = None
    ) -> bool:
        """Evaluate if a feature is enabled for given context"""
        if toggle_id not in self.toggles:
            return False
        
        rule = self.toggles[toggle_id]
        
        if not rule.enabled:
            return False
        
        # Evaluate based on toggle type
        if rule.type == ToggleType.BOOLEAN:
            return rule.value
        
        elif rule.type == ToggleType.PERCENTAGE:
            return self._evaluate_percentage(toggle_id, user_id, rule)
        
        elif rule.type == ToggleType.USER_SEGMENT:
            return self._evaluate_user_segment(user_id, rule)
        
        elif rule.type == ToggleType.TIME_WINDOW:
            return self._evaluate_time_window(rule)
        
        elif rule.type == ToggleType.CUSTOM:
            return self._evaluate_custom(context, rule)
        
        return False
    
    def _evaluate_percentage(
        self,
        toggle_id: str,
        user_id: Optional[str],
        rule: ToggleRule
    ) -> bool:
        """Percentage-based rollout (consistent per user)"""
        if not user_id:
            return False
        
        percentage = rule.value
        # Hash user_id + toggle_id for consistent bucketing
        hash_input = f"{user_id}:{toggle_id}"
        hash_value = hash(hash_input) % 100
        return hash_value < percentage
    
    def _evaluate_user_segment(
        self,
        user_id: Optional[str],
        rule: ToggleRule
    ) -> bool:
        """User segment targeting"""
        if not user_id:
            return False
        
        allowed_segments = rule.conditions.get("segments", [])
        user_segment = rule.conditions.get("user_segment_map", {}).get(user_id)
        return user_segment in allowed_segments
    
    def _evaluate_time_window(self, rule: ToggleRule) -> bool:
        """Time-based activation"""
        now = datetime.utcnow()
        start = rule.conditions.get("start_time")
        end = rule.conditions.get("end_time")
        
        if start and now < start:
            return False
        if end and now > end:
            return False
        return True
    
    def _evaluate_custom(
        self,
        context: Optional[Dict[str, Any]],
        rule: ToggleRule
    ) -> bool:
        """Custom evaluation logic"""
        if not context:
            return False
        
        evaluator = rule.conditions.get("evaluator")
        return evaluator(context) if evaluator else False
    
    def update_toggle(self, toggle_id: str, updates: Dict[str, Any]) -> None:
        """Update an existing toggle"""
        if toggle_id not in self.toggles:
            raise ValueError(f"Toggle {toggle_id} not found")
        
        rule = self.toggles[toggle_id]
        rule.updated_at = datetime.utcnow()
        rule.version += 1
        
        for key, value in updates.items():
            if hasattr(rule, key):
                setattr(rule, key, value)
        
        self._log_audit("UPDATE", toggle_id, updates)
    
    def _log_audit(self, action: str, toggle_id: str, data: Any) -> None:
        """Log configuration changes"""
        self.audit_log.append({
            "timestamp": datetime.utcnow().isoformat(),
            "action": action,
            "toggle_id": toggle_id,
            "data": str(data)
        })
    
    def get_audit_log(self) -> List[Dict]:
        """Retrieve audit trail"""
        return self.audit_log.copy()
```

### 2. Configuration Manager with Caching

```python
from abc import ABC, abstractmethod
from threading import Lock, Thread
import time
from typing import Any, Dict, Optional

class ConfigStore(ABC):
    """Abstract configuration store"""
    
    @abstractmethod
    def get(self, key: str) -> Optional[Any]:
        pass
    
    @abstractmethod
    def set(self, key: str, value: Any) -> None:
        pass
    
    @abstractmethod
    def get_all(self) -> Dict[str, Any]:
        pass

class InMemoryConfigStore(ConfigStore):
    """Simple in-memory store"""
    
    def __init__(self):
        self.data: Dict[str, Any] = {}
        self.lock = Lock()
    
    def get(self, key: str) -> Optional[Any]:
        with self.lock:
            return self.data.get(key)
    
    def set(self, key: str, value: Any) -> None:
        with self.lock:
            self.data[key] = value
    
    def get_all(self) -> Dict[str, Any]:
        with self.lock:
            return self.data.copy()

class CachedConfigManager:
    """Configuration manager with local caching and TTL"""
    
    def __init__(
        self,
        store: ConfigStore,
        cache_ttl_seconds: int = 300
    ):
        self.store = store
        self.cache_ttl = cache_ttl_seconds
        self.cache: Dict[str, tuple[Any, float]] = {}
        self.lock = Lock()
        self.refresh_thread: Optional[Thread] = None
        self.running = False
    
    def get(self, key: str, default: Any = None) -> Any:
        """Get config value with caching"""
        # Check cache
        with self.lock:
            if key in self.cache:
                value, timestamp = self.cache[key]
                if time.time() - timestamp < self.cache_ttl:
                    return value
        
        # Cache miss or expired - fetch from store
        value = self.store.get(key)
        
        # Update cache
        with self.lock:
            self.cache[key] = (value, time.time())
        
        return value if value is not None else default
    
    def set(self, key: str, value: Any) -> None:
        """Set config value and invalidate cache"""
        self.store.set(key, value)
        
        with self.lock:
            self.cache[key] = (value, time.time())
    
    def invalidate_cache(self, key: Optional[str] = None) -> None:
        """Invalidate cache entry or entire cache"""
        with self.lock:
            if key:
                self.cache.pop(key, None)
            else:
                self.cache.clear()
    
    def start_background_refresh(self, interval: int = 60) -> None:
        """Start background refresh thread"""
        self.running = True
        self.refresh_thread = Thread(
            target=self._refresh_loop,
            args=(interval,),
            daemon=True
        )
        self.refresh_thread.start()
    
    def _refresh_loop(self, interval: int) -> None:
        """Background refresh loop"""
        while self.running:
            time.sleep(interval)
            self.invalidate_cache()
    
    def stop_background_refresh(self) -> None:
        """Stop background refresh"""
        self.running = False
        if self.refresh_thread:
            self.refresh_thread.join(timeout=5)
```

### 3. Remote Configuration Service

```python
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Dict, Any
import asyncio

app = FastAPI()

class ConfigUpdate(BaseModel):
    key: str
    value: Any
    version: int

class ToggleUpdate(BaseModel):
    toggle_id: str
    enabled: bool
    value: Any
    conditions: Dict[str, Any]

# Global instances
config_manager = CachedConfigManager(InMemoryConfigStore())
toggle_engine = FeatureToggleEngine()

@app.get("/config/{key}")
async def get_config(key: str):
    """Get single config value"""
    value = config_manager.get(key)
    if value is None:
        raise HTTPException(status_code=404, detail="Config not found")
    return {"key": key, "value": value}

@app.get("/config")
async def get_all_config():
    """Get all configurations"""
    return config_manager.store.get_all()

@app.post("/config")
async def update_config(update: ConfigUpdate):
    """Update configuration"""
    config_manager.set(update.key, update.value)
    return {"status": "updated", "key": update.key}

@app.get("/toggle/{toggle_id}")
async def check_toggle(
    toggle_id: str,
    user_id: str = None,
    context: Dict[str, Any] = None
):
    """Check if feature is enabled"""
    is_enabled = toggle_engine.is_enabled(toggle_id, user_id, context)
    return {"toggle_id": toggle_id, "enabled": is_enabled}

@app.post("/toggle")
async def update_toggle(update: ToggleUpdate):
    """Update feature toggle"""
    toggle_engine.update_toggle(
        update.toggle_id,
        {
            "enabled": update.enabled,
            "value": update.value,
            "conditions": update.conditions
        }
    )
    return {"status": "updated", "toggle_id": update.toggle_id}

@app.get("/audit")
async def get_audit_log():
    """Get audit trail"""
    return toggle_engine.get_audit_log()
```

### 4. Client SDK

```python
from typing import Optional, Dict, Any
import requests
from functools import lru_cache

class ConfigClient:
    """Client SDK for applications"""
    
    def __init__(self, base_url: str, cache_size: int = 128):
        self.base_url = base_url
        self.cache_size = cache_size
    
    @lru_cache(maxsize=128)
    def get_config(self, key: str) -> Optional[Any]:
        """Get configuration value"""
        try:
            response = requests.get(f"{self.base_url}/config/{key}")
            if response.status_code == 200:
                return response.json()["value"]
        except Exception as e:
            print(f"Error fetching config: {e}")
        return None
    
    def is_feature_enabled(
        self,
        feature_id: str,
        user_id: Optional[str] = None,
        context: Optional[Dict[str, Any]] = None
    ) -> bool:
        """Check if feature is enabled"""
        try:
            params = {}
            if user_id:
                params["user_id"] = user_id
            
            response = requests.get(
                f"{self.base_url}/toggle/{feature_id}",
                params=params,
                json=context
            )
            if response.status_code == 200:
                return response.json()["enabled"]
        except Exception as e:
            print(f"Error checking toggle: {e}")
        return False
    
    def clear_cache(self) -> None:
        """Clear local cache"""
        self.get_config.cache_clear()

# Usage in application
config_client = ConfigClient("http://config-service:8000")

def process_payment(user_id: str, amount: float):
    """Example: Use feature toggle for new payment processor"""
    
    if config_client.is_feature_enabled("new_payment_processor", user_id):
        return process_with_new_processor(amount)
    else:
        return process_with_legacy_processor(amount)

def get_api_timeout() -> int:
    """Example: Dynamic configuration"""
    return config_client.get_config("api_timeout_ms") or 5000
```

### 5. Advanced: Webhook-Based Push Updates

```python
from typing import Callable, List
import json

class ConfigChangeListener:
    """Listen for configuration changes"""
    
    def __init__(self):
        self.listeners: Dict[str, List[Callable]] = {}
    
    def subscribe(self, key: str, callback: Callable) -> None:
        """Subscribe to config changes"""
        if key not in self.listeners:
            self.listeners[key] = []
        self.listeners[key].append(callback)
    
    def notify(self, key: str, old_value: Any, new_value: Any) -> None:
        """Notify all listeners of change"""
        if key in self.listeners:
            for callback in self.listeners[key]:
                try:
                    callback(key, old_value, new_value)
                except Exception as e:
                    print(f"Error in callback: {e}")

# Global listener
change_listener = ConfigChangeListener()

@app.post("/webhook/config-changed")
async def config_changed_webhook(update: ConfigUpdate):
    """Webhook endpoint for config changes"""
    old_value = config_manager.get(update.key)
    config_manager.set(update.key, update.value)
    change_listener.notify(update.key, old_value, update.value)
    return {"status": "processed"}

# Application usage
def on_timeout_changed(key: str, old: Any, new: Any):
    print(f"Timeout changed from {old}ms to {new}ms")

change_listener.subscribe("api_timeout_ms", on_timeout_changed)
```

---

## Tips & Best Practices

### 1. **Caching Strategy**
- Use **local caching with TTL** to reduce latency and load
- Implement **cache invalidation** for critical configs
- Balance between freshness and performance

### 2. **Gradual Rollouts**
```python
# Percentage-based rollout
toggle_engine.register_toggle(ToggleRule(
    id="new_feature",
    type=ToggleType.PERCENTAGE,
    value=10,  # 10% of users
    enabled=True
))
# Increase gradually: 10% → 25% → 50% → 100%
```

### 3. **Kill Switches**
- Always have **emergency disable** for critical features
- Test kill switch regularly
- Document rollback procedures

### 4. **Audit & Compliance**
- Log all configuration changes with **timestamps and user info**
- Maintain **version history** for rollback
- Implement **approval workflows** for production changes

### 5. **Monitoring**
```python
# Track toggle usage
toggle_metrics = {
    "feature_id": "new_checkout",
    "enabled_count": 1250,
    "disabled_count": 8750,
    "error_rate": 0.02
}
```

### 6. **Testing**
```python
def test_feature_toggle():
    engine = FeatureToggleEngine()
    rule = ToggleRule(
        id="test_feature",
        type=ToggleType.PERCENTAGE,
        value=50,
        enabled=True
    )
    engine.register_toggle(rule)
    
    # Test consistency
    results = [engine.is_enabled("test_feature", f"user_{i}") 
               for i in range(100)]
    assert sum(results) == 50  # Approximately 50%
```

### 7. **Fallback Defaults**
- Always provide **sensible defaults** when config unavailable
- Never fail open for security-related toggles
- Log when defaults are used

### 8. **Performance Considerations**
- Use **in-memory caching** for frequently accessed configs
- Implement **batch fetching** for multiple configs
- Consider **CDN distribution** for global applications

---

## Summary

Runtime configuration management enables:
- ✅ **Zero-downtime deployments** via feature toggles
- ✅ **A/B testing and gradual rollouts** with user targeting
- ✅ **Emergency kill switches** for quick incident response
- ✅ **Audit trails** for compliance and debugging
- ✅ **Reduced deployment risk** through controlled releases

Combine with monitoring and alerting for production-grade configuration management.