aishastr.live

Agentic Architecture Pattern: Blue Green Deployment

Written by

admin

in


    """
Pattern 271: Blue-Green Deployment

Purpose: Zero-downtime deployment with environment switching and instant rollback.
Maintains two identical production environments for seamless version transitions.
"""

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


class EnvironmentColor(Enum):
    """Environment designation colors."""
    BLUE = "blue"
    GREEN = "green"


class EnvironmentStatus(Enum):
    """Environment status."""
    ACTIVE = "active"  # Serving production traffic
    STANDBY = "standby"  # Ready but not serving
    DEPLOYING = "deploying"  # Deployment in progress
    TESTING = "testing"  # Under validation
    FAILED = "failed"  # Deployment failed


@dataclass
class Environment:
    """Production environment."""
    color: EnvironmentColor
    version: str
    status: EnvironmentStatus
    agent_count: int = 0
    health_score: float = 1.0
    traffic_percentage: float = 0.0
    deployment_time: datetime = field(default_factory=datetime.now)


@dataclass
class DeploymentRecord:
    """Record of deployment operation."""
    operation: str  # deploy, switch, rollback
    from_env: Optional[str] = None
    to_env: Optional[str] = None
    version: str = ""
    success: bool = True
    message: str = ""
    timestamp: datetime = field(default_factory=datetime.now)


class BlueGreenDeployment:
    """
    Blue-Green Deployment - Zero-downtime deployment strategy.
    
    Features:
    - Dual environment management
    - Instant traffic switching
    - Zero-downtime deployments
    - Quick rollback capability
    - Health monitoring
    - Traffic routing control
    - Version tracking
    - Deployment validation
    """
    
    def __init__(self, name: str):
        self.name = name
        self.llm = ChatOpenAI(model="gpt-4", temperature=0.3)
        
        # Environments
        self.environments: Dict[EnvironmentColor, Environment] = {
            EnvironmentColor.BLUE: Environment(
                color=EnvironmentColor.BLUE,
                version="1.0.0",
                status=EnvironmentStatus.ACTIVE,
                agent_count=5,
                traffic_percentage=100.0
            ),
            EnvironmentColor.GREEN: Environment(
                color=EnvironmentColor.GREEN,
                version="1.0.0",
                status=EnvironmentStatus.STANDBY,
                agent_count=5
            )
        }
        
        # Current active environment
        self.active_env = EnvironmentColor.BLUE
        
        # Deployment history
        self.deployment_history: List[DeploymentRecord] = []
        
        # Statistics
        self.statistics: Dict[str, Any] = {
            "total_deployments": 0,
            "successful_deployments": 0,
            "failed_deployments": 0,
            "rollbacks": 0,
            "switches": 0,
            "current_version": "1.0.0"
        }
    
    def deploy_to_standby(
        self,
        new_version: str
    ) -> Dict[str, Any]:
        """Deploy new version to standby environment."""
        
        # Identify standby environment
        standby_color = (
            EnvironmentColor.GREEN if self.active_env == EnvironmentColor.BLUE
            else EnvironmentColor.BLUE
        )
        standby_env = self.environments[standby_color]
        
        # Update standby environment
        standby_env.status = EnvironmentStatus.DEPLOYING
        standby_env.version = new_version
        standby_env.deployment_time = datetime.now()
        
        # Simulate deployment
        success = True  # In real scenario, this would involve actual deployment
        
        if success:
            standby_env.status = EnvironmentStatus.TESTING
            standby_env.health_score = 1.0
            
            self.statistics["total_deployments"] += 1
            self.statistics["successful_deployments"] += 1
            
            record = DeploymentRecord(
                operation="deploy",
                to_env=standby_color.value,
                version=new_version,
                success=True,
                message=f"Successfully deployed v{new_version} to {standby_color.value}"
            )
        else:
            standby_env.status = EnvironmentStatus.FAILED
            
            self.statistics["total_deployments"] += 1
            self.statistics["failed_deployments"] += 1
            
            record = DeploymentRecord(
                operation="deploy",
                to_env=standby_color.value,
                version=new_version,
                success=False,
                message=f"Failed to deploy v{new_version}"
            )
        
        self.deployment_history.append(record)
        
        return {
            "success": success,
            "standby_env": standby_color.value,
            "version": new_version,
            "status": standby_env.status.value,
            "message": record.message
        }
    
    def switch_traffic(self) -> Dict[str, Any]:
        """Switch traffic from active to standby environment."""
        
        # Get environments
        old_active = self.active_env
        new_active = (
            EnvironmentColor.GREEN if old_active == EnvironmentColor.BLUE
            else EnvironmentColor.BLUE
        )
        
        old_env = self.environments[old_active]
        new_env = self.environments[new_active]
        
        # Validate standby is ready
        if new_env.status not in [EnvironmentStatus.TESTING, EnvironmentStatus.STANDBY]:
            return {
                "success": False,
                "message": f"Standby environment ({new_active.value}) not ready for switch"
            }
        
        if new_env.health_score  Dict[str, Any]:
        """Instant rollback to previous environment."""
        
        current_active = self.active_env
        previous_env = (
            EnvironmentColor.GREEN if current_active == EnvironmentColor.BLUE
            else EnvironmentColor.BLUE
        )
        
        current = self.environments[current_active]
        previous = self.environments[previous_env]
        
        # Switch back
        current.traffic_percentage = 0.0
        current.status = EnvironmentStatus.STANDBY
        
        previous.traffic_percentage = 100.0
        previous.status = EnvironmentStatus.ACTIVE
        
        self.active_env = previous_env
        self.statistics["rollbacks"] += 1
        self.statistics["current_version"] = previous.version
        
        record = DeploymentRecord(
            operation="rollback",
            from_env=current_active.value,
            to_env=previous_env.value,
            version=previous.version,
            success=True,
            message=f"Rolled back from {current_active.value} to {previous_env.value}"
        )
        
        self.deployment_history.append(record)
        
        return {
            "success": True,
            "from_env": current_active.value,
            "to_env": previous_env.value,
            "version": previous.version,
            "message": record.message
        }
    
    def get_environment_status(self) -> Dict[str, Any]:
        """Get status of both environments."""
        
        return {
            "active_env": self.active_env.value,
            "environments": {
                color.value: {
                    "version": env.version,
                    "status": env.status.value,
                    "agent_count": env.agent_count,
                    "health_score": env.health_score,
                    "traffic_percentage": env.traffic_percentage,
                    "is_active": color == self.active_env
                }
                for color, env in self.environments.items()
            }
        }
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get deployment statistics."""
        return {
            "name": self.name,
            **self.statistics,
            "active_environment": self.active_env.value
        }


class BlueGreenPattern:
    """
    Blue-Green Deployment Pattern - Manage blue-green deployments.
    
    Coordinates zero-downtime deployment operations.
    """
    
    def __init__(self):
        self.deployments: Dict[str, BlueGreenDeployment] = {}
    
    def create_deployment(self, name: str) -> BlueGreenDeployment:
        """Create blue-green deployment manager."""
        deployment = BlueGreenDeployment(name)
        self.deployments[name] = deployment
        print(f"   Created blue-green deployment: {name}")
        return deployment
    
    def deploy(
        self,
        deployment_name: str,
        version: str
    ) -> Dict[str, Any]:
        """Deploy new version."""
        
        deployment = self.deployments.get(deployment_name)
        if not deployment:
            return {"error": f"Deployment {deployment_name} not found"}
        
        result = deployment.deploy_to_standby(version)
        
        print(f"\n=== Deploy v{version} to Standby ===")
        print(f"Success: {result['success']}")
        print(f"Standby Environment: {result['standby_env']}")
        print(f"Status: {result['status']}")
        print(f"Message: {result['message']}")
        
        return result
    
    def switch(self, deployment_name: str) -> Dict[str, Any]:
        """Switch traffic to standby."""
        
        deployment = self.deployments.get(deployment_name)
        if not deployment:
            return {"error": f"Deployment {deployment_name} not found"}
        
        result = deployment.switch_traffic()
        
        print(f"\n=== Switch Traffic ===")
        print(f"Success: {result['success']}")
        if result['success']:
            print(f"From: {result['from_env']} ā†’ To: {result['to_env']}")
            print(f"Version: {result['version']}")
        print(f"Message: {result['message']}")
        
        return result


# Example usage
if __name__ == "__main__":
    print("=== Blue-Green Deployment Pattern Demo ===\n")
    
    # Create pattern
    pattern = BlueGreenPattern()
    
    # Create deployment
    print("1. Setting up blue-green deployment...")
    deployment = pattern.create_deployment("ProductionDeployment")
    
    # Show initial status
    print("\n2. Initial environment status:")
    status = deployment.get_environment_status()
    print(f"Active Environment: {status['active_env']}")
    for env_name, env_status in status['environments'].items():
        print(f"\n{env_name.upper()}:")
        print(f"  Version: {env_status['version']}")
        print(f"  Status: {env_status['status']}")
        print(f"  Traffic: {env_status['traffic_percentage']}%")
        print(f"  Health: {env_status['health_score']}")
    
    # Deploy new version
    print("\n" + "="*60)
    print("\n3. Deploying new version...")
    result1 = pattern.deploy("ProductionDeployment", "2.0.0")
    
    # Show updated status
    print("\n" + "="*60)
    print("\n4. Status after deployment:")
    status = deployment.get_environment_status()
    for env_name, env_status in status['environments'].items():
        if not env_status['is_active']:
            print(f"\n{env_name.upper()} (Standby):")
            print(f"  Version: {env_status['version']}")
            print(f"  Status: {env_status['status']}")
    
    # Switch traffic
    print("\n" + "="*60)
    print("\n5. Switching traffic to new version...")
    result2 = pattern.switch("ProductionDeployment")
    
    # Final status
    print("\n" + "="*60)
    print("\n6. Final environment status:")
    status = deployment.get_environment_status()
    print(f"Active Environment: {status['active_env']}")
    for env_name, env_status in status['environments'].items():
        print(f"\n{env_name.upper()}:")
        print(f"  Version: {env_status['version']}")
        print(f"  Status: {env_status['status']}")
        print(f"  Traffic: {env_status['traffic_percentage']}%")
        print(f"  Active: {env_status['is_active']}")
    
    # Statistics
    print("\n" + "="*60)
    print("\n7. Deployment Statistics:")
    stats = deployment.get_statistics()
    for key, value in stats.items():
        print(f"   {key}: {value}")
    
    print("\nāœ“ Comprehensive blue-green deployment with zero-downtime switching complete!")

Agentic Architecture Pattern: Blue Green Deployment

Pattern 271: 271 Blue Green Deployment – Line-by-Line Explanation

Overview

Pattern 271: Blue-Green Deployment

Purpose: Zero-downtime deployment with environment switching and instant rollback. Maintains two identical production environments for seamless version transitions.

Code Structure Analysis

File Information

  • Pattern Number: 271
  • Pattern Name: 271 Blue Green Deployment
  • File Path: patterns\deployment\271_blue_green_deployment.py
  • Total Lines: 412

Detailed Code Explanation

Module Docstring

"""
Pattern 271: Blue-Green Deployment

Purpose: Zero-downtime deployment with environment switching and instant rollback.
Maintains two identical production environments for seamless version transitions.
"""

Explanation: This module docstring describes the purpose and key features of the 271 Blue Green 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 collections import defaultdict
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

EnvironmentColor (Line 17)

class EnvironmentColor(Enum):
    BLUE
    GREEN

Purpose: Environment designation colors.

Values: – BLUE: Represents blue – GREEN: Represents green

EnvironmentStatus (Line 23)

class EnvironmentStatus(Enum):
    ACTIVE
    STANDBY
    DEPLOYING
    TESTING
    FAILED

Purpose: Environment status.

Values: – ACTIVE: Currently operational – STANDBY: Represents standby – DEPLOYING: Represents deploying – TESTING: Represents testing – FAILED: Execution failed

Classes

EnvironmentColor (Line 17)

Description: Environment designation colors.

EnvironmentStatus (Line 23)

Description: Environment status.

Environment (Line 33)

Description: Production environment.

DeploymentRecord (Line 45)

Description: Record of deployment operation.

BlueGreenDeployment (Line 56)

Description: Blue-Green Deployment – Zero-downtime deployment strategy.

Features: – Dual environment management – Instant traffic switching – Zero-downtime deployments – Quick rollback capability – Health monitoring – Traffic routing control – Version tracking – Deployment validation

Methods:

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

Purpose: No method docstring.

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

deploy_to_standby (Line 108)
def deploy_to_standby(self, new_version):
    """Deploy new version to standby environment."""

Purpose: Deploy new version to standby environment.

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

switch_traffic (Line 167)
def switch_traffic(self):
    """Switch traffic from active to standby environment."""

Purpose: Switch traffic from active to standby environment.

Parameters: – self: Instance reference

rollback (Line 224)
def rollback(self):
    """Instant rollback to previous environment."""

Purpose: Instant rollback to previous environment.

Parameters: – self: Instance reference

get_environment_status (Line 266)
def get_environment_status(self):
    """Get status of both environments."""

Purpose: Get status of both environments.

Parameters: – self: Instance reference

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

Purpose: Get deployment statistics.

Parameters: – self: Instance reference

BlueGreenPattern (Line 293)

Description: Blue-Green Deployment Pattern – Manage blue-green deployments.

Coordinates zero-downtime deployment operations.

Methods:

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

Purpose: No method docstring.

Parameters: – self: Instance reference

create_deployment (Line 303)
def create_deployment(self, name):
    """Create blue-green deployment manager."""

Purpose: Create blue-green deployment manager.

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

deploy (Line 310)
def deploy(self, deployment_name, version):
    """Deploy new version."""

Purpose: Deploy new version.

Parameters: – self: Instance reference – deployment_name: Parameter for deployment_name – version: Parameter for version

switch (Line 331)
def switch(self, deployment_name):
    """Switch traffic to standby."""

Purpose: Switch traffic to standby.

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

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 = EnvironmentColor(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