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)

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

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

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

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

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

# 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

# Track toggle usage
toggle_metrics = {
    "feature_id": "new_checkout",
    "enabled_count": 1250,
    "disabled_count": 8750,
    "error_rate": 0.02
}

6. Testing

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.