aishastr.live

Agentic Architecture Pattern: Rolling Update

Written by

admin

in


    """
Pattern 273: Rolling Update

Purpose: Sequential instance updates with health checking and rollback capability
"""

from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from collections import defaultdict


class UpdateStatus(Enum):
    """Status of instance update."""
    PENDING = "pending"
    UPDATING = "updating"
    UPDATED = "updated"
    FAILED = "failed"
    ROLLED_BACK = "rolled_back"


class DeploymentPhase(Enum):
    """Phase of rolling deployment."""
    NOT_STARTED = "not_started"
    IN_PROGRESS = "in_progress"
    PAUSED = "paused"
    COMPLETED = "completed"
    ROLLING_BACK = "rolling_back"
    FAILED = "failed"


@dataclass
class Instance:
    """Represents a single instance in the deployment."""
    id: str
    version: str
    status: UpdateStatus
    health_score: float = 1.0
    updated_at: Optional[datetime] = None
    error_message: Optional[str] = None


@dataclass
class UpdateBatch:
    """Represents a batch of instances being updated."""
    batch_number: int
    instance_ids: List[str]
    started_at: Optional[datetime] = None
    completed_at: Optional[datetime] = None
    success: bool = False


@dataclass
class DeploymentRecord:
    """Record of deployment operation."""
    operation: str
    batch_number: Optional[int]
    instances_affected: int
    success: bool
    message: str
    timestamp: datetime


class RollingUpdate:
    """
    Implements rolling update deployment strategy.
    
    Features:
    - Sequential instance-by-instance or batch updates
    - Configurable batch size
    - Health checking between batches
    - Pause/resume capability
    - Automatic rollback on failure
    - Update progress tracking
    """
    
    def __init__(
        self,
        total_instances: int = 10,
        batch_size: int = 2,
        min_health_score: float = 0.85,
        wait_between_batches: int = 5
    ):
        self.total_instances = total_instances
        self.batch_size = batch_size
        self.min_health_score = min_health_score
        self.wait_between_batches = wait_between_batches
        
        # Initialize instances with current version
        self.instances: Dict[str, Instance] = {
            f"instance-{i}": Instance(
                id=f"instance-{i}",
                version="1.0.0",
                status=UpdateStatus.PENDING,
                health_score=1.0
            )
            for i in range(total_instances)
        }
        
        self.target_version: Optional[str] = None
        self.phase = DeploymentPhase.NOT_STARTED
        self.current_batch: Optional[UpdateBatch] = None
        self.batches: List[UpdateBatch] = []
        self.deployment_history: List[DeploymentRecord] = []
        self.paused = False
        
        # Statistics
        self.statistics = {
            "total_deployments": 0,
            "successful_deployments": 0,
            "failed_deployments": 0,
            "rollbacks": 0,
            "total_instances_updated": 0,
            "current_version": "1.0.0"
        }
    
    def start_update(self, new_version: str) -> Dict[str, Any]:
        """
        Start rolling update to new version.
        
        Args:
            new_version: Target version to deploy
            
        Returns:
            Status of update initiation
        """
        if self.phase in [DeploymentPhase.IN_PROGRESS, DeploymentPhase.ROLLING_BACK]:
            return {
                "success": False,
                "message": f"Cannot start update: deployment is {self.phase.value}"
            }
        
        self.target_version = new_version
        self.phase = DeploymentPhase.IN_PROGRESS
        self.paused = False
        self.batches = []
        self.current_batch = None
        
        # Reset all instances to pending
        for instance in self.instances.values():
            instance.status = UpdateStatus.PENDING
        
        # Create batches
        instance_ids = list(self.instances.keys())
        for i in range(0, len(instance_ids), self.batch_size):
            batch = UpdateBatch(
                batch_number=len(self.batches) + 1,
                instance_ids=instance_ids[i:i + self.batch_size]
            )
            self.batches.append(batch)
        
        self.statistics["total_deployments"] += 1
        
        record = DeploymentRecord(
            operation="start_update",
            batch_number=None,
            instances_affected=0,
            success=True,
            message=f"Started rolling update to version {new_version} with {len(self.batches)} batches",
            timestamp=datetime.now()
        )
        self.deployment_history.append(record)
        
        return {
            "success": True,
            "target_version": new_version,
            "total_batches": len(self.batches),
            "batch_size": self.batch_size,
            "message": f"Rolling update initiated to version {new_version}"
        }
    
    def update_next_batch(self) -> Dict[str, Any]:
        """
        Update the next batch of instances.
        
        Returns:
            Status of batch update
        """
        if self.phase != DeploymentPhase.IN_PROGRESS:
            return {
                "success": False,
                "message": f"Cannot update: deployment is {self.phase.value}"
            }
        
        if self.paused:
            return {
                "success": False,
                "message": "Deployment is paused"
            }
        
        # Find next pending batch
        next_batch = None
        for batch in self.batches:
            if batch.started_at is None:
                next_batch = batch
                break
        
        if not next_batch:
            # All batches completed
            self.phase = DeploymentPhase.COMPLETED
            self.statistics["successful_deployments"] += 1
            self.statistics["current_version"] = self.target_version
            
            record = DeploymentRecord(
                operation="complete_update",
                batch_number=None,
                instances_affected=self.total_instances,
                success=True,
                message=f"Rolling update to version {self.target_version} completed successfully",
                timestamp=datetime.now()
            )
            self.deployment_history.append(record)
            
            return {
                "success": True,
                "message": "All batches completed successfully",
                "final_version": self.target_version
            }
        
        # Update batch
        self.current_batch = next_batch
        next_batch.started_at = datetime.now()
        
        # Update instances in batch
        success_count = 0
        for instance_id in next_batch.instance_ids:
            instance = self.instances[instance_id]
            instance.status = UpdateStatus.UPDATING
            
            # Simulate update (90% success rate per instance)
            import random
            if random.random()  Dict[str, Any]:
        """Check health of updated instances in batch."""
        unhealthy_instances = []
        failed_instances = []
        
        for instance_id in batch.instance_ids:
            instance = self.instances[instance_id]
            if instance.status == UpdateStatus.FAILED:
                failed_instances.append(instance_id)
            elif instance.health_score < self.min_health_score:
                unhealthy_instances.append(instance_id)
        
        healthy = len(failed_instances) == 0 and len(unhealthy_instances) == 0
        
        message = "All instances healthy"
        if failed_instances:
            message = f"Failed instances: {', '.join(failed_instances)}"
        elif unhealthy_instances:
            message = f"Unhealthy instances (health  Dict[str, Any]:
        """Pause the rolling update."""
        if self.phase != DeploymentPhase.IN_PROGRESS:
            return {
                "success": False,
                "message": f"Cannot pause: deployment is {self.phase.value}"
            }
        
        self.paused = True
        self.phase = DeploymentPhase.PAUSED
        
        record = DeploymentRecord(
            operation="pause_update",
            batch_number=self.current_batch.batch_number if self.current_batch else None,
            instances_affected=0,
            success=True,
            message="Rolling update paused",
            timestamp=datetime.now()
        )
        self.deployment_history.append(record)
        
        return {
            "success": True,
            "message": "Rolling update paused",
            "current_batch": self.current_batch.batch_number if self.current_batch else None
        }
    
    def resume_update(self) -> Dict[str, Any]:
        """Resume the rolling update."""
        if self.phase != DeploymentPhase.PAUSED:
            return {
                "success": False,
                "message": f"Cannot resume: deployment is {self.phase.value}"
            }
        
        self.paused = False
        self.phase = DeploymentPhase.IN_PROGRESS
        
        record = DeploymentRecord(
            operation="resume_update",
            batch_number=self.current_batch.batch_number if self.current_batch else None,
            instances_affected=0,
            success=True,
            message="Rolling update resumed",
            timestamp=datetime.now()
        )
        self.deployment_history.append(record)
        
        return {
            "success": True,
            "message": "Rolling update resumed"
        }
    
    def rollback(self) -> Dict[str, Any]:
        """
        Rollback all updated instances to previous version.
        
        Returns:
            Status of rollback
        """
        self.phase = DeploymentPhase.ROLLING_BACK
        
        rollback_count = 0
        previous_version = self.statistics["current_version"]
        
        for instance in self.instances.values():
            if instance.status in [UpdateStatus.UPDATED, UpdateStatus.FAILED]:
                instance.version = previous_version
                instance.status = UpdateStatus.ROLLED_BACK
                instance.health_score = 1.0
                instance.updated_at = datetime.now()
                instance.error_message = None
                rollback_count += 1
        
        self.phase = DeploymentPhase.NOT_STARTED
        self.statistics["rollbacks"] += 1
        self.statistics["failed_deployments"] += 1
        
        record = DeploymentRecord(
            operation="rollback",
            batch_number=None,
            instances_affected=rollback_count,
            success=True,
            message=f"Rolled back {rollback_count} instances to version {previous_version}",
            timestamp=datetime.now()
        )
        self.deployment_history.append(record)
        
        return {
            "success": True,
            "instances_rolled_back": rollback_count,
            "version": previous_version,
            "message": f"Successfully rolled back to version {previous_version}"
        }
    
    def get_status(self) -> Dict[str, Any]:
        """Get current deployment status."""
        updated_count = sum(1 for i in self.instances.values() if i.status == UpdateStatus.UPDATED)
        failed_count = sum(1 for i in self.instances.values() if i.status == UpdateStatus.FAILED)
        pending_count = sum(1 for i in self.instances.values() if i.status == UpdateStatus.PENDING)
        
        completed_batches = sum(1 for b in self.batches if b.completed_at is not None)
        
        return {
            "phase": self.phase.value,
            "paused": self.paused,
            "target_version": self.target_version,
            "current_version": self.statistics["current_version"],
            "progress": {
                "total_instances": self.total_instances,
                "updated": updated_count,
                "failed": failed_count,
                "pending": pending_count,
                "percentage": (updated_count / self.total_instances) * 100 if self.total_instances > 0 else 0
            },
            "batches": {
                "total": len(self.batches),
                "completed": completed_batches,
                "remaining": len(self.batches) - completed_batches
            },
            "current_batch": self.current_batch.batch_number if self.current_batch else None
        }
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get deployment statistics."""
        return {
            "total_deployments": self.statistics["total_deployments"],
            "successful_deployments": self.statistics["successful_deployments"],
            "failed_deployments": self.statistics["failed_deployments"],
            "rollbacks": self.statistics["rollbacks"],
            "total_instances_updated": self.statistics["total_instances_updated"],
            "current_version": self.statistics["current_version"],
            "total_instances": self.total_instances,
            "batch_size": self.batch_size
        }


class RollingUpdatePattern:
    """
    Coordinator for rolling update deployments.
    
    Manages rolling update lifecycle with health checking and rollback.
    """
    
    def __init__(
        self,
        total_instances: int = 10,
        batch_size: int = 2,
        min_health_score: float = 0.85
    ):
        self.deployment = RollingUpdate(
            total_instances=total_instances,
            batch_size=batch_size,
            min_health_score=min_health_score
        )
    
    def deploy(self, new_version: str, auto_progress: bool = False) -> Dict[str, Any]:
        """
        Execute rolling update deployment.
        
        Args:
            new_version: Target version
            auto_progress: If True, automatically update all batches
            
        Returns:
            Deployment results
        """
        result = self.deployment.start_update(new_version)
        if not result["success"]:
            return result
        
        if auto_progress:
            results = [result]
            while True:
                batch_result = self.deployment.update_next_batch()
                results.append(batch_result)
                
                if not batch_result["success"]:
                    # Health check failed, initiate rollback
                    rollback_result = self.deployment.rollback()
                    results.append(rollback_result)
                    break
                
                if self.deployment.phase == DeploymentPhase.COMPLETED:
                    break
            
            return {
                "success": self.deployment.phase == DeploymentPhase.COMPLETED,
                "final_status": self.deployment.get_status(),
                "batch_results": results
            }
        
        return result
    
    def get_status(self) -> Dict[str, Any]:
        """Get deployment status."""
        return self.deployment.get_status()
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get deployment statistics."""
        return self.deployment.get_statistics()


# Example usage
if __name__ == "__main__":
    print("=== Rolling Update Deployment Demo ===\n")
    
    # Create rolling update deployment with 10 instances, batch size 2
    pattern = RollingUpdatePattern(total_instances=10, batch_size=2, min_health_score=0.85)
    
    print("Initial status:")
    status = pattern.get_status()
    print(f"  Phase: {status['phase']}")
    print(f"  Current version: {status['current_version']}")
    print(f"  Total instances: {status['progress']['total_instances']}")
    
    # Start rolling update
    print("\n1. Starting rolling update to v2.0.0...")
    result = pattern.deployment.start_update("2.0.0")
    print(f"  Status: {result['message']}")
    print(f"  Total batches: {result['total_batches']}")
    print(f"  Batch size: {result['batch_size']}")
    
    # Update batches one by one
    print("\n2. Updating batches sequentially...")
    batch_num = 1
    while pattern.deployment.phase == DeploymentPhase.IN_PROGRESS:
        print(f"\n  Batch {batch_num}:")
        batch_result = pattern.deployment.update_next_batch()
        
        if not batch_result["success"]:
            print(f"    ❌ Failed: {batch_result['message']}")
            print(f"    Recommendation: {batch_result.get('recommendation', 'N/A')}")
            
            # Rollback
            print(f"\n3. Rolling back deployment...")
            rollback_result = pattern.deployment.rollback()
            print(f"  {rollback_result['message']}")
            print(f"  Instances rolled back: {rollback_result['instances_rolled_back']}")
            break
        
        print(f"    ✓ Success: Updated {batch_result['instances_updated']}/{batch_result['total_instances']} instances")
        print(f"    Health check: {batch_result['health_check']['message']}")
        print(f"    Remaining batches: {batch_result['remaining_batches']}")
        
        batch_num += 1
    
    # Final status
    print("\n4. Final deployment status:")
    status = pattern.get_status()
    print(f"  Phase: {status['phase']}")
    print(f"  Current version: {status['current_version']}")
    print(f"  Progress: {status['progress']['updated']}/{status['progress']['total_instances']} instances ({status['progress']['percentage']:.1f}%)")
    print(f"  Batches completed: {status['batches']['completed']}/{status['batches']['total']}")
    
    # Statistics
    print("\n5. Deployment statistics:")
    stats = pattern.get_statistics()
    print(f"  Total deployments: {stats['total_deployments']}")
    print(f"  Successful: {stats['successful_deployments']}")
    print(f"  Failed: {stats['failed_deployments']}")
    print(f"  Rollbacks: {stats['rollbacks']}")
    print(f"  Total instances updated: {stats['total_instances_updated']}")
    
    print("\n✓ Comprehensive rolling update with batch processing complete!")

Agentic Architecture Pattern: Rolling Update

Pattern 273: 273 Rolling Update – Line-by-Line Explanation

Overview

Pattern 273: Rolling Update

Purpose: Sequential instance updates with health checking and rollback capability

Code Structure Analysis

File Information

  • Pattern Number: 273
  • Pattern Name: 273 Rolling Update
  • File Path: patterns\deployment\273_rolling_update.py
  • Total Lines: 585

Detailed Code Explanation

Module Docstring

"""
Pattern 273: Rolling Update

Purpose: Sequential instance updates with health checking and rollback capability
"""

Explanation: This module docstring describes the purpose and key features of the 273 Rolling Update 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
            import random  # 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

Enumerations

UpdateStatus (Line 14)

class UpdateStatus(Enum):
    PENDING
    UPDATING
    UPDATED
    FAILED
    ROLLED_BACK

Purpose: Status of instance update.

Values: – PENDING: Initial state, not yet processed – UPDATING: Represents updating – UPDATED: Represents updated – FAILED: Execution failed – ROLLED_BACK: Represents rolled back

DeploymentPhase (Line 23)

class DeploymentPhase(Enum):
    NOT_STARTED
    IN_PROGRESS
    PAUSED
    COMPLETED
    ROLLING_BACK
    FAILED

Purpose: Phase of rolling deployment.

Values: – NOT_STARTED: Represents not started – IN_PROGRESS: Represents in progress – PAUSED: Represents paused – COMPLETED: Successfully finished – ROLLING_BACK: Represents rolling back – FAILED: Execution failed

Classes

UpdateStatus (Line 14)

Description: Status of instance update.

DeploymentPhase (Line 23)

Description: Phase of rolling deployment.

Instance (Line 34)

Description: Represents a single instance in the deployment.

UpdateBatch (Line 45)

Description: Represents a batch of instances being updated.

DeploymentRecord (Line 55)

Description: Record of deployment operation.

RollingUpdate (Line 65)

Description: Implements rolling update deployment strategy.

Features: – Sequential instance-by-instance or batch updates – Configurable batch size – Health checking between batches – Pause/resume capability – Automatic rollback on failure – Update progress tracking

Methods:

__init__ (Line 78)
def __init__(self, total_instances, batch_size, min_health_score, wait_between_batches):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – total_instances: Parameter for total_instances – batch_size: Parameter for batch_size – min_health_score: Parameter for min_health_score – wait_between_batches: Parameter for wait_between_batches

start_update (Line 118)
def start_update(self, new_version):
    """Start rolling update to new version.

Args:
    new_version: Target version to deploy

Returns:
    Status of update initiation"""

Purpose: Start rolling update to new version.

Args: new_version: Target version to deploy

Returns: Status of update initiation

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

update_next_batch (Line 173)
def update_next_batch(self):
    """Update the next batch of instances.

Returns:
    Status of batch update"""

Purpose: Update the next batch of instances.

Returns: Status of batch update

Parameters: – self: Instance reference

_check_batch_health (Line 293)
def _check_batch_health(self, batch):
    """Check health of updated instances in batch."""

Purpose: Check health of updated instances in batch.

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

pause_update (Line 320)
def pause_update(self):
    """Pause the rolling update."""

Purpose: Pause the rolling update.

Parameters: – self: Instance reference

resume_update (Line 347)
def resume_update(self):
    """Resume the rolling update."""

Purpose: Resume the rolling update.

Parameters: – self: Instance reference

rollback (Line 373)
def rollback(self):
    """Rollback all updated instances to previous version.

Returns:
    Status of rollback"""

Purpose: Rollback all updated instances to previous version.

Returns: Status of rollback

Parameters: – self: Instance reference

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

Purpose: Get current deployment status.

Parameters: – self: Instance reference

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

Purpose: Get deployment statistics.

Parameters: – self: Instance reference

RollingUpdatePattern (Line 457)

Description: Coordinator for rolling update deployments.

Manages rolling update lifecycle with health checking and rollback.

Methods:

__init__ (Line 464)
def __init__(self, total_instances, batch_size, min_health_score):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – total_instances: Parameter for total_instances – batch_size: Parameter for batch_size – min_health_score: Parameter for min_health_score

deploy (Line 476)
def deploy(self, new_version, auto_progress):
    """Execute rolling update deployment.

Args:
    new_version: Target version
    auto_progress: If True, automatically update all batches

Returns:
    Deployment results"""

Purpose: Execute rolling update deployment.

Args: new_version: Target version auto_progress: If True, automatically update all batches

Returns: Deployment results

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

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

Purpose: Get deployment status.

Parameters: – self: Instance reference

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

Purpose: Get deployment statistics.

Parameters: – self: Instance reference

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