"""
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