aishastr.live

Agentic Architecture Pattern: Canary Deployment

Written by

admin

in


    """
Pattern 272: Canary Deployment

Purpose: Gradual rollout with traffic routing and automated monitoring.
Tests new versions with small user percentage before full deployment.
"""

from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI


class DeploymentStage(Enum):
    """Canary deployment stages."""
    INITIAL = "initial"  # 5% traffic
    EXPANDED = "expanded"  # 25% traffic
    MAJORITY = "majority"  # 50% traffic
    COMPLETE = "complete"  # 100% traffic
    FAILED = "failed"  # Deployment failed


class HealthStatus(Enum):
    """Health status."""
    HEALTHY = "healthy"
    DEGRADED = "degraded"
    UNHEALTHY = "unhealthy"


@dataclass
class DeploymentVersion:
    """Version deployment info."""
    version: str
    stage: DeploymentStage
    traffic_percentage: float
    error_rate: float = 0.0
    latency_ms: float = 0.0
    health_status: HealthStatus = HealthStatus.HEALTHY
    deployment_time: datetime = field(default_factory=datetime.now)


class CanaryDeployment:
    """
    Canary Deployment - Gradual traffic shift strategy.
    
    Features:
    - Gradual traffic increase
    - Automated health monitoring
    - Error rate tracking
    - Latency monitoring
    - Automatic rollback on failure
    - Stage progression control
    - Metrics comparison
    - Risk mitigation
    """
    
    def __init__(self, name: str):
        self.name = name
        self.llm = ChatOpenAI(model="gpt-4", temperature=0.3)
        
        # Versions
        self.stable_version: DeploymentVersion = DeploymentVersion(
            version="1.0.0",
            stage=DeploymentStage.COMPLETE,
            traffic_percentage=100.0
        )
        self.canary_version: Optional[DeploymentVersion] = None
        
        # Thresholds
        self.max_error_rate = 0.05  # 5%
        self.max_latency_increase = 1.5  # 50% increase
        
        # History
        self.deployment_history: List[Dict[str, Any]] = []
        
        # Statistics
        self.statistics: Dict[str, Any] = {
            "total_deployments": 0,
            "successful_deployments": 0,
            "failed_deployments": 0,
            "rollbacks": 0,
            "current_version": "1.0.0"
        }
    
    def start_canary(self, new_version: str) -> Dict[str, Any]:
        """Start canary deployment with initial traffic."""
        
        if self.canary_version:
            return {
                "success": False,
                "message": "Canary deployment already in progress"
            }
        
        # Create canary with 5% traffic
        self.canary_version = DeploymentVersion(
            version=new_version,
            stage=DeploymentStage.INITIAL,
            traffic_percentage=5.0
        )
        
        # Adjust stable version traffic
        self.stable_version.traffic_percentage = 95.0
        
        self.statistics["total_deployments"] += 1
        
        record = {
            "operation": "start_canary",
            "version": new_version,
            "stage": DeploymentStage.INITIAL.value,
            "traffic": 5.0,
            "timestamp": datetime.now()
        }
        self.deployment_history.append(record)
        
        return {
            "success": True,
            "version": new_version,
            "stage": DeploymentStage.INITIAL.value,
            "traffic_percentage": 5.0,
            "message": f"Started canary deployment v{new_version} at 5% traffic"
        }
    
    def promote_canary(self) -> Dict[str, Any]:
        """Promote canary to next stage if healthy."""
        
        if not self.canary_version:
            return {
                "success": False,
                "message": "No active canary deployment"
            }
        
        # Check health
        health_check = self._check_canary_health()
        if not health_check["healthy"]:
            return {
                "success": False,
                "message": f"Canary unhealthy: {health_check['reason']}",
                "should_rollback": True
            }
        
        # Promote to next stage
        current_stage = self.canary_version.stage
        
        if current_stage == DeploymentStage.INITIAL:
            self.canary_version.stage = DeploymentStage.EXPANDED
            self.canary_version.traffic_percentage = 25.0
            self.stable_version.traffic_percentage = 75.0
            next_stage = "expanded (25%)"
        
        elif current_stage == DeploymentStage.EXPANDED:
            self.canary_version.stage = DeploymentStage.MAJORITY
            self.canary_version.traffic_percentage = 50.0
            self.stable_version.traffic_percentage = 50.0
            next_stage = "majority (50%)"
        
        elif current_stage == DeploymentStage.MAJORITY:
            self.canary_version.stage = DeploymentStage.COMPLETE
            self.canary_version.traffic_percentage = 100.0
            self.stable_version.traffic_percentage = 0.0
            next_stage = "complete (100%)"
            
            # Complete deployment
            self.stable_version = self.canary_version
            self.canary_version = None
            self.statistics["successful_deployments"] += 1
            self.statistics["current_version"] = self.stable_version.version
        
        else:
            return {
                "success": False,
                "message": "Canary already at final stage"
            }
        
        record = {
            "operation": "promote_canary",
            "version": self.canary_version.version if self.canary_version else self.stable_version.version,
            "stage": next_stage,
            "timestamp": datetime.now()
        }
        self.deployment_history.append(record)
        
        return {
            "success": True,
            "stage": next_stage,
            "message": f"Promoted canary to {next_stage}"
        }
    
    def rollback_canary(self) -> Dict[str, Any]:
        """Rollback canary deployment."""
        
        if not self.canary_version:
            return {
                "success": False,
                "message": "No active canary to rollback"
            }
        
        canary_version = self.canary_version.version
        
        # Remove canary
        self.canary_version = None
        self.stable_version.traffic_percentage = 100.0
        
        self.statistics["rollbacks"] += 1
        self.statistics["failed_deployments"] += 1
        
        record = {
            "operation": "rollback_canary",
            "version": canary_version,
            "reason": "Health check failed",
            "timestamp": datetime.now()
        }
        self.deployment_history.append(record)
        
        return {
            "success": True,
            "rolled_back_version": canary_version,
            "current_version": self.stable_version.version,
            "message": f"Rolled back canary v{canary_version}"
        }
    
    def _check_canary_health(self) -> Dict[str, Any]:
        """Check canary health metrics."""
        
        if not self.canary_version:
            return {"healthy": False, "reason": "No canary deployed"}
        
        # Check error rate
        if self.canary_version.error_rate > self.max_error_rate:
            return {
                "healthy": False,
                "reason": f"Error rate too high: {self.canary_version.error_rate:.2%}"
            }
        
        # Check latency
        if self.canary_version.latency_ms > self.stable_version.latency_ms * self.max_latency_increase:
            return {
                "healthy": False,
                "reason": f"Latency increased by {(self.canary_version.latency_ms / self.stable_version.latency_ms - 1):.1%}"
            }
        
        # Check overall health
        if self.canary_version.health_status != HealthStatus.HEALTHY:
            return {
                "healthy": False,
                "reason": f"Health status: {self.canary_version.health_status.value}"
            }
        
        return {"healthy": True}
    
    def update_metrics(
        self,
        version: str,
        error_rate: float,
        latency_ms: float,
        health_status: HealthStatus = HealthStatus.HEALTHY
    ):
        """Update metrics for version."""
        
        if self.canary_version and self.canary_version.version == version:
            self.canary_version.error_rate = error_rate
            self.canary_version.latency_ms = latency_ms
            self.canary_version.health_status = health_status
        elif self.stable_version.version == version:
            self.stable_version.error_rate = error_rate
            self.stable_version.latency_ms = latency_ms
            self.stable_version.health_status = health_status
    
    def get_status(self) -> Dict[str, Any]:
        """Get deployment status."""
        
        status = {
            "stable_version": {
                "version": self.stable_version.version,
                "traffic_percentage": self.stable_version.traffic_percentage,
                "error_rate": self.stable_version.error_rate,
                "latency_ms": self.stable_version.latency_ms,
                "health": self.stable_version.health_status.value
            }
        }
        
        if self.canary_version:
            status["canary_version"] = {
                "version": self.canary_version.version,
                "stage": self.canary_version.stage.value,
                "traffic_percentage": self.canary_version.traffic_percentage,
                "error_rate": self.canary_version.error_rate,
                "latency_ms": self.canary_version.latency_ms,
                "health": self.canary_version.health_status.value
            }
        
        return status
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get deployment statistics."""
        return {
            "name": self.name,
            **self.statistics
        }


class CanaryPattern:
    """Canary Deployment Pattern - Manage canary deployments."""
    
    def __init__(self):
        self.deployments: Dict[str, CanaryDeployment] = {}
    
    def create_deployment(self, name: str) -> CanaryDeployment:
        """Create canary deployment manager."""
        deployment = CanaryDeployment(name)
        self.deployments[name] = deployment
        print(f"   Created canary deployment: {name}")
        return deployment


# Example usage
if __name__ == "__main__":
    print("=== Canary Deployment Pattern Demo ===\n")
    
    pattern = CanaryPattern()
    deployment = pattern.create_deployment("ProductionCanary")
    
    print("\n1. Initial Status:")
    status = deployment.get_status()
    print(f"Stable: v{status['stable_version']['version']} ({status['stable_version']['traffic_percentage']}% traffic)")
    
    print("\n" + "="*60)
    print("\n2. Starting Canary Deployment (v2.0.0)...")
    result = deployment.start_canary("2.0.0")
    print(f"Success: {result['success']}")
    print(f"Message: {result['message']}")
    
    print("\n3. Status After Canary Start:")
    status = deployment.get_status()
    print(f"Stable: v{status['stable_version']['version']} ({status['stable_version']['traffic_percentage']}%)")
    print(f"Canary: v{status['canary_version']['version']} ({status['canary_version']['traffic_percentage']}%) - {status['canary_version']['stage']}")
    
    # Simulate healthy metrics
    deployment.update_metrics("2.0.0", error_rate=0.01, latency_ms=50)
    
    print("\n" + "="*60)
    print("\n4. Promoting Canary to 25%...")
    result = deployment.promote_canary()
    print(f"Success: {result['success']}")
    print(f"Stage: {result['stage']}")
    
    status = deployment.get_status()
    print(f"Canary Traffic: {status['canary_version']['traffic_percentage']}%")
    
    print("\n" + "="*60)
    print("\n5. Promoting to 50%...")
    result = deployment.promote_canary()
    print(f"Stage: {result['stage']}")
    
    print("\n" + "="*60)
    print("\n6. Completing Deployment (100%)...")
    result = deployment.promote_canary()
    print(f"Success: {result['success']}")
    print(f"Message: {result['message']}")
    
    print("\n7. Final Status:")
    status = deployment.get_status()
    print(f"Stable: v{status['stable_version']['version']} ({status['stable_version']['traffic_percentage']}%)")
    print(f"Canary Active: {'canary_version' in status}")
    
    print("\n8. Statistics:")
    stats = deployment.get_statistics()
    for key, value in stats.items():
        print(f"   {key}: {value}")
    
    print("\nāœ“ Comprehensive canary deployment with gradual rollout complete!")

Agentic Architecture Pattern: Canary Deployment

Pattern 272: 272 Canary Deployment – Line-by-Line Explanation

Overview

Pattern 272: Canary Deployment

Purpose: Gradual rollout with traffic routing and automated monitoring. Tests new versions with small user percentage before full deployment.

Code Structure Analysis

File Information

  • Pattern Number: 272
  • Pattern Name: 272 Canary Deployment
  • File Path: patterns\deployment\272_canary_deployment.py
  • Total Lines: 373

Detailed Code Explanation

Module Docstring

"""
Pattern 272: Canary Deployment

Purpose: Gradual rollout with traffic routing and automated monitoring.
Tests new versions with small user percentage before full deployment.
"""

Explanation: This module docstring describes the purpose and key features of the 272 Canary Deployment pattern.

Imports Section

from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI  # First 20 imports shown

Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional: Type hints for better code documentation – from dataclasses import dataclass, field: Data class decorators for clean data structures – from datetime import datetime: Date and time utilities – from enum import Enum: Enumeration support for fixed sets of values – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration

Enumerations

DeploymentStage (Line 16)

class DeploymentStage(Enum):
    INITIAL
    EXPANDED
    MAJORITY
    COMPLETE
    FAILED

Purpose: Canary deployment stages.

Values: – INITIAL: Represents initial – EXPANDED: Represents expanded – MAJORITY: Represents majority – COMPLETE: Represents complete – FAILED: Execution failed

HealthStatus (Line 25)

class HealthStatus(Enum):
    HEALTHY
    DEGRADED
    UNHEALTHY

Purpose: Health status.

Values: – HEALTHY: Represents healthy – DEGRADED: Represents degraded – UNHEALTHY: Represents unhealthy

Classes

DeploymentStage (Line 16)

Description: Canary deployment stages.

HealthStatus (Line 25)

Description: Health status.

DeploymentVersion (Line 33)

Description: Version deployment info.

CanaryDeployment (Line 44)

Description: Canary Deployment – Gradual traffic shift strategy.

Features: – Gradual traffic increase – Automated health monitoring – Error rate tracking – Latency monitoring – Automatic rollback on failure – Stage progression control – Metrics comparison – Risk mitigation

Methods:

__init__ (Line 59)
def __init__(self, name):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label

start_canary (Line 87)
def start_canary(self, new_version):
    """Start canary deployment with initial traffic."""

Purpose: Start canary deployment with initial traffic.

Parameters: – self: Instance reference – new_version: Parameter for new_version

promote_canary (Line 125)
def promote_canary(self):
    """Promote canary to next stage if healthy."""

Purpose: Promote canary to next stage if healthy.

Parameters: – self: Instance reference

rollback_canary (Line 190)
def rollback_canary(self):
    """Rollback canary deployment."""

Purpose: Rollback canary deployment.

Parameters: – self: Instance reference

_check_canary_health (Line 223)
def _check_canary_health(self):
    """Check canary health metrics."""

Purpose: Check canary health metrics.

Parameters: – self: Instance reference

update_metrics (Line 252)
def update_metrics(self, version, error_rate, latency_ms, health_status):
    """Update metrics for version."""

Purpose: Update metrics for version.

Parameters: – self: Instance reference – version: Parameter for version – error_rate: Parameter for error_rate – latency_ms: Parameter for latency_ms – health_status: Parameter for health_status

get_status (Line 270)
def get_status(self):
    """Get deployment status."""

Purpose: Get deployment status.

Parameters: – self: Instance reference

get_statistics (Line 295)
def get_statistics(self):
    """Get deployment statistics."""

Purpose: Get deployment statistics.

Parameters: – self: Instance reference

CanaryPattern (Line 303)

Description: Canary Deployment Pattern – Manage canary deployments.

Methods:

__init__ (Line 306)
def __init__(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

create_deployment (Line 309)
def create_deployment(self, name):
    """Create canary deployment manager."""

Purpose: Create canary deployment manager.

Parameters: – self: Instance reference – name: Identifier or label

Design Patterns Used

  • Strategy Pattern: Different execution strategies for various scenarios
  • State Pattern: Managing different states in the workflow
  • Chain of Responsibility: Sequential processing through workflow nodes
  • Observer Pattern: Tracking and monitoring pattern execution
  • Factory Pattern: Creating instances with proper configuration

Usage Example

# Initialize the pattern
pattern = DeploymentStage(name="example", model_name="gpt-4")

# Configure and execute
result = pattern.execute(data={"key": "value"})

# Check results
print(result)

Key Concepts

  • LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
  • Workflow Orchestration: LangGraph manages execution flow through nodes
  • State Management: TypedDict defines workflow state structure
  • Error Handling: Comprehensive error handling and fallback mechanisms
  • Extensibility: Designed for easy customization and extension

Extension Points

This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities

Comments

Leave a Reply

Your email address will not be published. Required fields are marked *

More posts