A/B Rollout

A/B Rollout im Überblick

A/B Rollout rollt neue Features schrittweise aus: erst an eine kleine Nutzergruppe, dann an größere Segmente, schließlich an alle. So erkennst du Probleme früh, bevor alle betroffen sind.

Der Prozess:

Phase 1: Kleine Testgruppe → Monitoring, Fehler prüfen
Phase 2: Größeres Segment → Performance, Feedback
Phase 3: Breiter Rollout → Skalierung testen
Phase 4: Alle Nutzer → Vollständiger Launch

Bei Problemen: Sofort auf 0% zurück!

Vergleich Deployment-Strategien:

Technisch betrachtet

Feature Flag Implementation

from hashlib import md5

class FeatureFlag:
    def __init__(self, name, rollout_percentage=0):
        self.name = name
        self.rollout_percentage = rollout_percentage
    
    def is_enabled(self, user_id):
        # Sticky Assignment: User bekommt immer gleiches Ergebnis
        hash_value = int(md5(f"{self.name}:{user_id}".encode()).hexdigest(), 16)
        bucket = hash_value % 100
        
        return bucket < self.rollout_percentage

# Nutzung
new_checkout = FeatureFlag("new_checkout", rollout_percentage=10)

if new_checkout.is_enabled(user.id):
    return render_new_checkout()
else:
    return render_old_checkout()

Rollout-Plan

# rollout-plan.yaml
feature: new_recommendation_engine
stages:
  - name: canary
    percentage: 1
    duration: 2h
    success_criteria:
      error_rate: "< 0.1%"
      latency_p99: "< 200ms"
  
  - name: early_adopters
    percentage: 10
    duration: 24h
    success_criteria:
      error_rate: "< 0.1%"
      conversion_rate: "> baseline"
  
  - name: half
    percentage: 50
    duration: 24h
    success_criteria:
      error_rate: "< 0.1%"
      user_satisfaction: "> baseline"
  
  - name: full
    percentage: 100
    
rollback_trigger:
  - error_rate: "> 1%"
  - latency_p99: "> 500ms"

Metriken während Rollout

def compare_rollout_metrics(feature_name):
    control = get_metrics(feature_name, enabled=False)
    treatment = get_metrics(feature_name, enabled=True)
    
    comparison = {
        "error_rate": {
            "control": control["error_rate"],
            "treatment": treatment["error_rate"],
            "diff": treatment["error_rate"] - control["error_rate"]
        },
        "latency_p99": {
            "control": control["latency_p99"],
            "treatment": treatment["latency_p99"],
            "diff": treatment["latency_p99"] - control["latency_p99"]
        },
        "conversion": {
            "control": control["conversion_rate"],
            "treatment": treatment["conversion_rate"],
            "lift": (treatment["conversion_rate"] - control["conversion_rate"]) 
                    / control["conversion_rate"] * 100
        }
    }
    
    return comparison

Automatisches Rollback

def auto_rollback_check(feature, metrics):
    thresholds = feature.rollback_thresholds
    
    if metrics["error_rate"] > thresholds["max_error_rate"]:
        rollback(feature, reason="Error rate exceeded")
        alert("Automatic rollback triggered")
        return True
    
    if metrics["latency_p99"] > thresholds["max_latency"]:
        rollback(feature, reason="Latency exceeded")
        alert("Automatic rollback triggered")
        return True
    
    return False