Canary Releasing AI Model Versions in Production Without Downtime

import random
import hashlib
from typing import Optional, Dict, Any
from openai import OpenAI
import anthropic
import time
import logging


class CanaryRouter:
    """Route requests between model versions with canary deployment logic."""
    
    def __init__(self,
                 stable_model: str,
                 canary_model: str,
                 canary_percentage: float = 5.0,
                 sticky_users: bool = True):
        """
        Args:
            stable_model: Current production model (e.g., "gpt-4o")
            canary_model: New model to test (e.g., "gpt-5")
            canary_percentage: % of traffic to route to canary (0-100)
            sticky_users: If True, users consistently get same version
        """
        self.stable_model = stable_model
        self.canary_model = canary_model
        self.canary_percentage = canary_percentage
        self.sticky_users = sticky_users
        
        self.openai_client = OpenAI()
        self.logger = logging.getLogger(__name__)
    
    def _should_use_canary(self, user_id: Optional[str] = None) -> bool:
        """Determine if request should use canary version."""
        if self.sticky_users and user_id:
            # Consistent routing per user (hash-based)
            user_hash = int(hashlib.md5(user_id.encode()).hexdigest(), 16)
            return (user_hash % 100) < self.canary_percentage
        else:
            # Random routing
            return random.random() * 100 < self.canary_percentage
    
    def chat_completion(self,
                       messages: list,
                       user_id: Optional[str] = None,
                       **kwargs) -> Dict[str, Any]:
        """Route request to stable or canary model."""
        use_canary = self._should_use_canary(user_id)
        model = self.canary_model if use_canary else self.stable_model
        
        start_time = time.time()
        
        try:
            response = self.openai_client.chat.completions.create(
                model=model,
                messages=messages,
                **kwargs
            )
            
            latency = time.time() - start_time
            
            # Log for monitoring
            self.logger.info(
                "Model request completed",
                extra={
                    "model": model,
                    "model_version": "canary" if use_canary else "stable",
                    "user_id": user_id,
                    "latency_seconds": latency,
                    "input_tokens": response.usage.prompt_tokens,
                    "output_tokens": response.usage.completion_tokens,
                    "status": "success"
                }
            )
            
            return {
                "response": response.choices[0].message.content,
                "model": model,
                "version": "canary" if use_canary else "stable",
                "latency": latency,
                "usage": {
                    "prompt_tokens": response.usage.prompt_tokens,
                    "completion_tokens": response.usage.completion_tokens
                }
            }
        
        except Exception as e:
            # Log error
            self.logger.error(
                f"Model request failed: {e}",
                extra={
                    "model": model,
                    "model_version": "canary" if use_canary else "stable",
                    "user_id": user_id,
                    "error": str(e)
                },
                exc_info=True
            )
            raise


# Usage example
if __name__ == "__main__":
    import os
    
    # Configure canary deployment
    router = CanaryRouter(
        stable_model="gpt-4o",
        canary_model="gpt-5",
        canary_percentage=10.0,  # 10% canary traffic
        sticky_users=True  # Consistent experience per user
    )
    
    # Simulate requests from different users
    for i in range(20):
        result = router.chat_completion(
            messages=[{"role": "user", "content": "Hello!"}],
            user_id=f"user_{i % 5}",  # 5 unique users
            max_tokens=50
        )
        
        print(f"User {i % 5}: {result['version']} ({result['model']})")

# Kubernetes VirtualService for canary deployment
apiVersion: networking.istio.io/v1beta1
kind: VirtualService
metadata:
  name: llm-inference-canary
  namespace: ai-prod
spec:
  hosts:
    - llm-inference.ai-prod.svc.cluster.local
  http:
    - match:
        - headers:
            x-canary-user:
              exact: "true"  # Force canary for specific users
      route:
        - destination:
            host: llm-inference.ai-prod.svc.cluster.local
            subset: canary
          weight: 100
    
    - route:
        # Default traffic split
        - destination:
            host: llm-inference.ai-prod.svc.cluster.local
            subset: stable
          weight: 90  # 90% to stable
        - destination:
            host: llm-inference.ai-prod.svc.cluster.local
            subset: canary
          weight: 10  # 10% to canary

---
# DestinationRule defining stable and canary subsets
apiVersion: networking.istio.io/v1beta1
kind: DestinationRule
metadata:
  name: llm-inference-subsets
  namespace: ai-prod
spec:
  host: llm-inference.ai-prod.svc.cluster.local
  subsets:
    - name: stable
      labels:
        version: v1.0  # Stable model version
    - name: canary
      labels:
        version: v2.0  # Canary model version

---
# Deployments for each version
apiVersion: apps/v1
kind: Deployment
metadata:
  name: llm-inference-stable
  namespace: ai-prod
spec:
  replicas: 5
  selector:
    matchLabels:
      app: llm-inference
      version: v1.0
  template:
    metadata:
      labels:
        app: llm-inference
        version: v1.0
    spec:
      containers:
        - name: inference
          image: myregistry/llama-4-8b:stable
          resources:
            requests:
              memory: "16Gi"
              nvidia.com/gpu: 1
            limits:
              memory: "32Gi"
              nvidia.com/gpu: 1
          ports:
            - containerPort: 8000

---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: llm-inference-canary
  namespace: ai-prod
spec:
  replicas: 1  # Start with 1 replica for canary
  selector:
    matchLabels:
      app: llm-inference
      version: v2.0
  template:
    metadata:
      labels:
        app: llm-inference
        version: v2.0
    spec:
      containers:
        - name: inference
          image: myregistry/llama-4-8b:canary
          resources:
            requests:
              memory: "16Gi"
              nvidia.com/gpu: 1
            limits:
              memory: "32Gi"
              nvidia.com/gpu: 1
          ports:
            - containerPort: 8000

import dataclasses
from typing import List, Dict, Any
from datetime import datetime, timedelta
import psycopg2
from scipy import stats


@dataclasses.dataclass
class ModelMetrics:
    """Aggregated metrics for a model version."""
    version: str
    request_count: int
    error_count: int
    error_rate: float
    avg_latency_ms: float
    p95_latency_ms: float
    p99_latency_ms: float
    avg_input_tokens: float
    avg_output_tokens: float
    avg_cost_usd: float
    refusal_count: int
    refusal_rate: float


class CanaryMonitor:
    """Monitor canary deployment metrics and detect degradation."""
    
    def __init__(self, db_connection_string: str):
        self.conn = psycopg2.connect(db_connection_string)
    
    def get_metrics(self,
                   version: str,
                   start_time: datetime,
                   end_time: datetime) -> ModelMetrics:
        """Get aggregated metrics for a model version."""
        with self.conn.cursor() as cur:
            cur.execute("""
                WITH metrics AS (
                    SELECT
                        COUNT(*) as request_count,
                        SUM(CASE WHEN status = 'error' THEN 1 ELSE 0 END) as error_count,
                        AVG(latency_ms) as avg_latency_ms,
                        PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY latency_ms) as p95_latency,
                        PERCENTILE_CONT(0.99) WITHIN GROUP (ORDER BY latency_ms) as p99_latency,
                        AVG(input_tokens) as avg_input_tokens,
                        AVG(output_tokens) as avg_output_tokens,
                        AVG(cost_usd) as avg_cost_usd,
                        SUM(CASE WHEN is_refusal THEN 1 ELSE 0 END) as refusal_count
                    FROM llm_requests
                    WHERE model_version = %s
                      AND timestamp BETWEEN %s AND %s
                )
                SELECT
                    request_count,
                    error_count,
                    CAST(error_count AS FLOAT) / NULLIF(request_count, 0) as error_rate,
                    avg_latency_ms,
                    p95_latency,
                    p99_latency,
                    avg_input_tokens,
                    avg_output_tokens,
                    avg_cost_usd,
                    refusal_count,
                    CAST(refusal_count AS FLOAT) / NULLIF(request_count, 0) as refusal_rate
                FROM metrics
            """, (version, start_time, end_time))
            
            row = cur.fetchone()
            if not row:
                return None
            
            return ModelMetrics(
                version=version,
                request_count=row[0] or 0,
                error_count=row[1] or 0,
                error_rate=row[2] or 0.0,
                avg_latency_ms=row[3] or 0.0,
                p95_latency_ms=row[4] or 0.0,
                p99_latency_ms=row[5] or 0.0,
                avg_input_tokens=row[6] or 0.0,
                avg_output_tokens=row[7] or 0.0,
                avg_cost_usd=row[8] or 0.0,
                refusal_count=row[9] or 0,
                refusal_rate=row[10] or 0.0
            )
    
    def compare_versions(self,
                        stable_version: str,
                        canary_version: str,
                        window_hours: int = 24) -> Dict[str, Any]:
        """Compare stable vs canary metrics."""
        end_time = datetime.now()
        start_time = end_time - timedelta(hours=window_hours)
        
        stable_metrics = self.get_metrics(stable_version, start_time, end_time)
        canary_metrics = self.get_metrics(canary_version, start_time, end_time)
        
        if not stable_metrics or not canary_metrics:
            return {"error": "Insufficient data for comparison"}
        
        # Calculate relative differences
        comparison = {
            "window_hours": window_hours,
            "stable": dataclasses.asdict(stable_metrics),
            "canary": dataclasses.asdict(canary_metrics),
            "differences": {
                "error_rate_change": (
                    (canary_metrics.error_rate - stable_metrics.error_rate) / 
                    max(stable_metrics.error_rate, 0.001) * 100
                ),
                "latency_p95_change_pct": (
                    (canary_metrics.p95_latency_ms - stable_metrics.p95_latency_ms) / 
                    stable_metrics.p95_latency_ms * 100
                ),
                "cost_change_pct": (
                    (canary_metrics.avg_cost_usd - stable_metrics.avg_cost_usd) / 
                    stable_metrics.avg_cost_usd * 100
                ),
                "refusal_rate_change": (
                    (canary_metrics.refusal_rate - stable_metrics.refusal_rate) / 
                    max(stable_metrics.refusal_rate, 0.001) * 100
                )
            }
        }
        
        return comparison
    
    def should_rollback(self,
                       comparison: Dict[str, Any],
                       thresholds: Dict[str, float]) -> tuple[bool, List[str]]:
        """Determine if canary should be rolled back based on thresholds."""
        reasons = []
        differences = comparison["differences"]
        
        # Check error rate
        if differences["error_rate_change"] > thresholds.get("max_error_rate_increase_pct", 50):
            reasons.append(
                f"Error rate increased by {differences['error_rate_change']:.1f}% "
                f"(threshold: {thresholds.get('max_error_rate_increase_pct')}%)"
            )
        
        # Check latency
        if differences["latency_p95_change_pct"] > thresholds.get("max_latency_increase_pct", 30):
            reasons.append(
                f"P95 latency increased by {differences['latency_p95_change_pct']:.1f}% "
                f"(threshold: {thresholds.get('max_latency_increase_pct')}%)"
            )
        
        # Check cost
        if differences["cost_change_pct"] > thresholds.get("max_cost_increase_pct", 50):
            reasons.append(
                f"Cost increased by {differences['cost_change_pct']:.1f}% "
                f"(threshold: {thresholds.get('max_cost_increase_pct')}%)"
            )
        
        # Check refusal rate
        if differences["refusal_rate_change"] > thresholds.get("max_refusal_increase_pct", 100):
            reasons.append(
                f"Refusal rate increased by {differences['refusal_rate_change']:.1f}% "
                f"(threshold: {thresholds.get('max_refusal_increase_pct')}%)"
            )
        
        should_rollback = len(reasons) > 0
        return should_rollback, reasons


# Example usage
if __name__ == "__main__":
    monitor = CanaryMonitor("postgresql://user:pass@localhost/aiapp")
    
    # Compare stable vs canary
    comparison = monitor.compare_versions(
        stable_version="gpt-4o",
        canary_version="gpt-5",
        window_hours=24
    )
    
    print(f"\nStable (gpt-4o):")
    print(f"  Requests: {comparison['stable']['request_count']}")
    print(f"  Error rate: {comparison['stable']['error_rate']*100:.2f}%")
    print(f"  P95 latency: {comparison['stable']['p95_latency_ms']:.0f}ms")
    
    print(f"\nCanary (gpt-5):")
    print(f"  Requests: {comparison['canary']['request_count']}")
    print(f"  Error rate: {comparison['canary']['error_rate']*100:.2f}%")
    print(f"  P95 latency: {comparison['canary']['p95_latency_ms']:.0f}ms")
    
    print(f"\nDifferences:")
    for metric, change in comparison['differences'].items():
        print(f"  {metric}: {change:+.1f}%")
    
    # Check if rollback needed
    thresholds = {
        "max_error_rate_increase_pct": 50,
        "max_latency_increase_pct": 30,
        "max_cost_increase_pct": 50,
        "max_refusal_increase_pct": 100
    }
    
    should_rollback, reasons = monitor.should_rollback(comparison, thresholds)
    
    if should_rollback:
        print(f"\n⚠️  ROLLBACK RECOMMENDED:")
        for reason in reasons:
            print(f"  - {reason}")
    else:
        print(f"\n✓ Canary performing within acceptable thresholds")

# Automated rollback script
import sys
import subprocess
import time

def rollback_canary_kubernetes(namespace: str = "ai-prod"):
    """Rollback canary by setting traffic to 0%."""
    print("Initiating canary rollback...")
    
    # Update VirtualService to route 100% traffic to stable
    kubectl_patch = f"""
    kubectl patch virtualservice llm-inference-canary -n {namespace} --type=json -p='[
      {{
        "op": "replace",
        "path": "/spec/http/0/route/0/weight",
        "value": 100
      }},
      {{
        "op": "replace",
        "path": "/spec/http/0/route/1/weight",
        "value": 0
      }}
    ]'
    """
    
    result = subprocess.run(kubectl_patch, shell=True, capture_output=True, text=True)
    
    if result.returncode == 0:
        print("✓ Canary traffic set to 0% - rollback complete")
        print("  All traffic now routed to stable version")
        return True
    else:
        print(f"✗ Rollback failed: {result.stderr}")
        return False

def rollback_canary_application(router: 'CanaryRouter'):
    """Rollback canary at application level."""
    print("Rolling back canary deployment...")
    router.canary_percentage = 0.0
    print("✓ Canary traffic set to 0%")

# Monitoring loop with auto-rollback
def monitor_and_auto_rollback(check_interval_minutes: int = 15):
    """Continuously monitor canary and rollback if needed."""
    monitor = CanaryMonitor("postgresql://user:pass@localhost/aiapp")
    
    thresholds = {
        "max_error_rate_increase_pct": 50,
        "max_latency_increase_pct": 30,
        "max_cost_increase_pct": 50,
        "max_refusal_increase_pct": 100
    }
    
    while True:
        try:
            comparison = monitor.compare_versions(
                stable_version="gpt-4o",
                canary_version="gpt-5",
                window_hours=1  # Check last hour
            )
            
            should_rollback, reasons = monitor.should_rollback(comparison, thresholds)
            
            if should_rollback:
                print(f"\n⚠️  AUTO-ROLLBACK TRIGGERED at {datetime.now()}")
                for reason in reasons:
                    print(f"  - {reason}")
                
                # Execute rollback
                success = rollback_canary_kubernetes()
                
                if success:
                    # Send alert
                    print("\n📧 Sending rollback alert to team...")
                    # send_alert_to_slack/pagerduty(reasons)
                    break  # Exit monitoring loop
            else:
                print(f"✓ Canary healthy at {datetime.now()}")
        
        except Exception as e:
            print(f"Error during monitoring: {e}")
        
        # Wait before next check
        time.sleep(check_interval_minutes * 60)

if __name__ == "__main__":
    monitor_and_auto_rollback(check_interval_minutes=15)