COST06-BP03 - Select resource type, size, and number automatically based on metrics

import boto3
import json
import numpy as np
from datetime import datetime, timedelta
from dataclasses import dataclass
from typing import Dict, List, Optional, Tuple
from enum import Enum
import logging
from concurrent.futures import ThreadPoolExecutor
import time

class AutomationAction(Enum):
    SCALE_UP = "scale_up"
    SCALE_DOWN = "scale_down"
    RIGHTSIZE_UP = "rightsize_up"
    RIGHTSIZE_DOWN = "rightsize_down"
    CHANGE_INSTANCE_FAMILY = "change_instance_family"
    NO_ACTION = "no_action"

@dataclass
class AutomationDecision:
    resource_id: str
    current_config: str
    recommended_action: AutomationAction
    target_config: str
    confidence_score: float
    expected_cost_impact: float
    expected_performance_impact: str
    rationale: str
    safety_checks_passed: bool
    execution_timestamp: Optional[datetime] = None

@dataclass
class MetricThreshold:
    metric_name: str
    threshold_value: float
    comparison_operator: str  # >, <, >=, <=, ==
    evaluation_periods: int
    datapoints_to_alarm: int

class IntelligentResourceManager:
    def __init__(self):
        self.cloudwatch = boto3.client('cloudwatch')
        self.ec2 = boto3.client('ec2')
        self.autoscaling = boto3.client('autoscaling')
        self.lambda_client = boto3.client('lambda')
        self.events = boto3.client('events')
        
        # Configuration
        self.automation_config = {
            'max_changes_per_hour': 5,
            'min_confidence_threshold': 0.8,
            'safety_check_enabled': True,
            'dry_run_mode': False,
            'notification_topic_arn': None
        }
        
        # Metrics and thresholds
        self.metric_thresholds = {
            'cpu_high': MetricThreshold('CPUUtilization', 80, '>', 3, 2),
            'cpu_low': MetricThreshold('CPUUtilization', 20, '<', 6, 4),
            'memory_high': MetricThreshold('mem_used_percent', 85, '>', 3, 2),
            'memory_low': MetricThreshold('mem_used_percent', 30, '<', 6, 4),
            'cost_target_exceeded': MetricThreshold('EstimatedCharges', 1000, '>', 1, 1)
        }
        
        # Decision history for learning
        self.decision_history = []
        
        # Setup logging
        logging.basicConfig(level=logging.INFO)
        self.logger = logging.getLogger(__name__)
    
    def monitor_and_optimize_resources(self, resource_ids: List[str]) -> List[AutomationDecision]:
        """Main function to monitor resources and make optimization decisions"""
        
        decisions = []
        
        # Check rate limiting
        if not self.check_rate_limits():
            self.logger.warning("Rate limit exceeded, skipping optimization cycle")
            return decisions
        
        # Process resources in parallel
        with ThreadPoolExecutor(max_workers=5) as executor:
            futures = [
                executor.submit(self.analyze_and_decide, resource_id) 
                for resource_id in resource_ids
            ]
            
            for future in futures:
                try:
                    decision = future.result(timeout=30)
                    if decision and decision.recommended_action != AutomationAction.NO_ACTION:
                        decisions.append(decision)
                except Exception as e:
                    self.logger.error(f"Error processing resource: {e}")
        
        # Execute decisions if not in dry-run mode
        if not self.automation_config['dry_run_mode']:
            executed_decisions = self.execute_decisions(decisions)
            return executed_decisions
        else:
            self.logger.info(f"Dry-run mode: Would execute {len(decisions)} decisions")
            return decisions
    
    def analyze_and_decide(self, resource_id: str) -> Optional[AutomationDecision]:
        """Analyze a single resource and make optimization decision"""
        
        try:
            # Collect current metrics
            current_metrics = self.collect_current_metrics(resource_id)
            
            # Get resource configuration
            resource_config = self.get_resource_configuration(resource_id)
            
            # Analyze metrics against thresholds
            metric_analysis = self.analyze_metrics(current_metrics)
            
            # Make optimization decision
            decision = self.make_optimization_decision(
                resource_id, resource_config, current_metrics, metric_analysis
            )
            
            # Perform safety checks
            if decision and self.automation_config['safety_check_enabled']:
                decision.safety_checks_passed = self.perform_safety_checks(decision, current_metrics)
            else:
                decision.safety_checks_passed = True
            
            return decision
            
        except Exception as e:
            self.logger.error(f"Error analyzing resource {resource_id}: {e}")
            return None
    
    def collect_current_metrics(self, resource_id: str) -> Dict:
        """Collect current metrics for a resource"""
        
        end_time = datetime.now()
        start_time = end_time - timedelta(hours=1)
        
        metrics = {}
        
        # CPU Utilization
        cpu_data = self.get_metric_data(
            resource_id, 'AWS/EC2', 'CPUUtilization', start_time, end_time
        )
        if cpu_data:
            metrics['cpu_utilization'] = {
                'current': cpu_data[-1]['Average'] if cpu_data else 0,
                'average': np.mean([dp['Average'] for dp in cpu_data]),
                'maximum': max([dp['Maximum'] for dp in cpu_data]) if cpu_data else 0,
                'trend': self.calculate_trend([dp['Average'] for dp in cpu_data])
            }
        
        # Memory Utilization (if available)
        memory_data = self.get_metric_data(
            resource_id, 'CWAgent', 'mem_used_percent', start_time, end_time
        )
        if memory_data:
            metrics['memory_utilization'] = {
                'current': memory_data[-1]['Average'] if memory_data else 0,
                'average': np.mean([dp['Average'] for dp in memory_data]),
                'maximum': max([dp['Maximum'] for dp in memory_data]) if memory_data else 0
            }
        
        # Network Utilization
        network_data = self.get_metric_data(
            resource_id, 'AWS/EC2', 'NetworkIn', start_time, end_time
        )
        if network_data:
            metrics['network_utilization'] = {
                'current': network_data[-1]['Average'] if network_data else 0,
                'average': np.mean([dp['Average'] for dp in network_data])
            }
        
        # Cost metrics (estimated)
        metrics['estimated_hourly_cost'] = self.estimate_current_hourly_cost(resource_id)
        
        return metrics
    
    def analyze_metrics(self, metrics: Dict) -> Dict:
        """Analyze metrics against defined thresholds"""
        
        analysis = {
            'threshold_violations': [],
            'optimization_signals': [],
            'performance_indicators': {}
        }
        
        # Check CPU thresholds
        if 'cpu_utilization' in metrics:
            cpu_current = metrics['cpu_utilization']['current']
            cpu_average = metrics['cpu_utilization']['average']
            
            if cpu_average > self.metric_thresholds['cpu_high'].threshold_value:
                analysis['threshold_violations'].append('cpu_high')
                analysis['optimization_signals'].append('scale_up_or_rightsize_up')
            elif cpu_average < self.metric_thresholds['cpu_low'].threshold_value:
                analysis['threshold_violations'].append('cpu_low')
                analysis['optimization_signals'].append('scale_down_or_rightsize_down')
        
        # Check memory thresholds
        if 'memory_utilization' in metrics:
            memory_average = metrics['memory_utilization']['average']
            
            if memory_average > self.metric_thresholds['memory_high'].threshold_value:
                analysis['threshold_violations'].append('memory_high')
                analysis['optimization_signals'].append('memory_constrained')
            elif memory_average < self.metric_thresholds['memory_low'].threshold_value:
                analysis['threshold_violations'].append('memory_low')
                analysis['optimization_signals'].append('memory_over_provisioned')
        
        # Performance indicators
        analysis['performance_indicators'] = {
            'cpu_efficiency': self.calculate_efficiency_score(metrics.get('cpu_utilization', {})),
            'memory_efficiency': self.calculate_efficiency_score(metrics.get('memory_utilization', {})),
            'overall_health': self.calculate_overall_health_score(metrics)
        }
        
        return analysis
    
    def make_optimization_decision(self, resource_id: str, resource_config: Dict, 
                                 metrics: Dict, analysis: Dict) -> Optional[AutomationDecision]:
        """Make optimization decision based on analysis"""
        
        current_instance_type = resource_config.get('instance_type', 'unknown')
        optimization_signals = analysis['optimization_signals']
        
        # Decision logic based on signals
        if 'scale_up_or_rightsize_up' in optimization_signals:
            # High utilization - need more capacity
            if self.is_in_auto_scaling_group(resource_id):
                # Prefer scaling over rightsizing for ASG resources
                decision = AutomationDecision(
                    resource_id=resource_id,
                    current_config=current_instance_type,
                    recommended_action=AutomationAction.SCALE_UP,
                    target_config=f"Scale ASG capacity +1",
                    confidence_score=0.9,
                    expected_cost_impact=self.estimate_scaling_cost_impact(resource_id, 1),
                    expected_performance_impact="Positive - Reduced load per instance",
                    rationale=f"High CPU utilization ({metrics['cpu_utilization']['average']:.1f}%) detected",
                    safety_checks_passed=False
                )
            else:
                # Rightsize individual instance
                larger_instance = self.get_larger_instance_type(current_instance_type)
                decision = AutomationDecision(
                    resource_id=resource_id,
                    current_config=current_instance_type,
                    recommended_action=AutomationAction.RIGHTSIZE_UP,
                    target_config=larger_instance,
                    confidence_score=0.85,
                    expected_cost_impact=self.estimate_rightsizing_cost_impact(
                        current_instance_type, larger_instance
                    ),
                    expected_performance_impact="Positive - Increased capacity",
                    rationale=f"High utilization requires larger instance type",
                    safety_checks_passed=False
                )
        
        elif 'scale_down_or_rightsize_down' in optimization_signals:
            # Low utilization - can reduce capacity
            if self.is_in_auto_scaling_group(resource_id):
                decision = AutomationDecision(
                    resource_id=resource_id,
                    current_config=current_instance_type,
                    recommended_action=AutomationAction.SCALE_DOWN,
                    target_config=f"Scale ASG capacity -1",
                    confidence_score=0.8,
                    expected_cost_impact=self.estimate_scaling_cost_impact(resource_id, -1),
                    expected_performance_impact="Minimal - Low utilization indicates excess capacity",
                    rationale=f"Low CPU utilization ({metrics['cpu_utilization']['average']:.1f}%) detected",
                    safety_checks_passed=False
                )
            else:
                smaller_instance = self.get_smaller_instance_type(current_instance_type)
                if smaller_instance != current_instance_type:
                    decision = AutomationDecision(
                        resource_id=resource_id,
                        current_config=current_instance_type,
                        recommended_action=AutomationAction.RIGHTSIZE_DOWN,
                        target_config=smaller_instance,
                        confidence_score=0.8,
                        expected_cost_impact=self.estimate_rightsizing_cost_impact(
                            current_instance_type, smaller_instance
                        ),
                        expected_performance_impact="Low risk - Current utilization well below capacity",
                        rationale=f"Low utilization indicates over-provisioning",
                        safety_checks_passed=False
                    )
                else:
                    decision = None
        
        # Check for instance family optimization opportunities
        elif analysis['performance_indicators']['overall_health'] > 0.7:
            alternative_instance = self.suggest_alternative_instance_family(
                current_instance_type, metrics
            )
            if alternative_instance and alternative_instance != current_instance_type:
                cost_impact = self.estimate_rightsizing_cost_impact(
                    current_instance_type, alternative_instance
                )
                if cost_impact < 0:  # Cost savings
                    decision = AutomationDecision(
                        resource_id=resource_id,
                        current_config=current_instance_type,
                        recommended_action=AutomationAction.CHANGE_INSTANCE_FAMILY,
                        target_config=alternative_instance,
                        confidence_score=0.7,
                        expected_cost_impact=cost_impact,
                        expected_performance_impact="Neutral to positive - Better price-performance ratio",
                        rationale=f"Alternative instance family provides better value",
                        safety_checks_passed=False
                    )
                else:
                    decision = None
            else:
                decision = None
        else:
            decision = None
        
        # Apply confidence threshold
        if decision and decision.confidence_score < self.automation_config['min_confidence_threshold']:
            self.logger.info(f"Decision confidence {decision.confidence_score} below threshold, skipping")
            return None
        
        return decision
    
    def perform_safety_checks(self, decision: AutomationDecision, metrics: Dict) -> bool:
        """Perform safety checks before executing decision"""
        
        safety_checks = []
        
        # Check 1: Ensure resource is not already under stress
        if 'cpu_utilization' in metrics:
            current_cpu = metrics['cpu_utilization']['current']
            if decision.recommended_action in [AutomationAction.RIGHTSIZE_DOWN, AutomationAction.SCALE_DOWN]:
                if current_cpu > 60:
                    safety_checks.append(False)
                    self.logger.warning(f"Safety check failed: Current CPU {current_cpu}% too high for downsizing")
                else:
                    safety_checks.append(True)
            else:
                safety_checks.append(True)
        
        # Check 2: Verify resource is not in a critical state
        resource_health = self.check_resource_health(decision.resource_id)
        if resource_health == 'unhealthy':
            safety_checks.append(False)
            self.logger.warning(f"Safety check failed: Resource {decision.resource_id} is unhealthy")
        else:
            safety_checks.append(True)
        
        # Check 3: Ensure change window compliance
        if not self.is_in_change_window():
            safety_checks.append(False)
            self.logger.warning("Safety check failed: Outside of approved change window")
        else:
            safety_checks.append(True)
        
        # Check 4: Verify no recent changes
        if self.has_recent_changes(decision.resource_id, hours=2):
            safety_checks.append(False)
            self.logger.warning(f"Safety check failed: Recent changes detected for {decision.resource_id}")
        else:
            safety_checks.append(True)
        
        return all(safety_checks)
    
    def execute_decisions(self, decisions: List[AutomationDecision]) -> List[AutomationDecision]:
        """Execute approved automation decisions"""
        
        executed_decisions = []
        
        for decision in decisions:
            if not decision.safety_checks_passed:
                self.logger.warning(f"Skipping decision for {decision.resource_id}: Safety checks failed")
                continue
            
            try:
                success = self.execute_single_decision(decision)
                if success:
                    decision.execution_timestamp = datetime.now()
                    executed_decisions.append(decision)
                    self.decision_history.append(decision)
                    
                    # Send notification
                    self.send_notification(decision)
                    
                    self.logger.info(f"Successfully executed decision for {decision.resource_id}")
                else:
                    self.logger.error(f"Failed to execute decision for {decision.resource_id}")
                    
            except Exception as e:
                self.logger.error(f"Error executing decision for {decision.resource_id}: {e}")
        
        return executed_decisions
    
    def execute_single_decision(self, decision: AutomationDecision) -> bool:
        """Execute a single automation decision"""
        
        try:
            if decision.recommended_action == AutomationAction.SCALE_UP:
                return self.scale_auto_scaling_group(decision.resource_id, 1)
            
            elif decision.recommended_action == AutomationAction.SCALE_DOWN:
                return self.scale_auto_scaling_group(decision.resource_id, -1)
            
            elif decision.recommended_action in [
                AutomationAction.RIGHTSIZE_UP, 
                AutomationAction.RIGHTSIZE_DOWN,
                AutomationAction.CHANGE_INSTANCE_FAMILY
            ]:
                return self.rightsize_instance(decision.resource_id, decision.target_config)
            
            return False
            
        except Exception as e:
            self.logger.error(f"Error in execute_single_decision: {e}")
            return False
    
    def scale_auto_scaling_group(self, resource_id: str, scale_direction: int) -> bool:
        """Scale an Auto Scaling Group"""
        
        try:
            # Get ASG name for the instance
            asg_name = self.get_asg_name_for_instance(resource_id)
            if not asg_name:
                return False
            
            # Get current capacity
            response = self.autoscaling.describe_auto_scaling_groups(
                AutoScalingGroupNames=[asg_name]
            )
            
            if not response['AutoScalingGroups']:
                return False
            
            asg = response['AutoScalingGroups'][0]
            current_capacity = asg['DesiredCapacity']
            new_capacity = max(asg['MinSize'], min(asg['MaxSize'], current_capacity + scale_direction))
            
            if new_capacity == current_capacity:
                self.logger.info(f"No scaling needed for ASG {asg_name}")
                return True
            
            # Update desired capacity
            self.autoscaling.set_desired_capacity(
                AutoScalingGroupName=asg_name,
                DesiredCapacity=new_capacity,
                HonorCooldown=True
            )
            
            self.logger.info(f"Scaled ASG {asg_name} from {current_capacity} to {new_capacity}")
            return True
            
        except Exception as e:
            self.logger.error(f"Error scaling ASG: {e}")
            return False
    
    def rightsize_instance(self, resource_id: str, target_instance_type: str) -> bool:
        """Rightsize an EC2 instance"""
        
        try:
            # Stop the instance
            self.ec2.stop_instances(InstanceIds=[resource_id])
            
            # Wait for instance to stop
            waiter = self.ec2.get_waiter('instance_stopped')
            waiter.wait(InstanceIds=[resource_id], WaiterConfig={'Delay': 15, 'MaxAttempts': 40})
            
            # Modify instance type
            self.ec2.modify_instance_attribute(
                InstanceId=resource_id,
                InstanceType={'Value': target_instance_type}
            )
            
            # Start the instance
            self.ec2.start_instances(InstanceIds=[resource_id])
            
            self.logger.info(f"Rightsized instance {resource_id} to {target_instance_type}")
            return True
            
        except Exception as e:
            self.logger.error(f"Error rightsizing instance: {e}")
            return False
    
    def create_automation_policies(self) -> Dict:
        """Create comprehensive automation policies"""
        
        policies = {
            'scaling_policies': {
                'cpu_scale_up': {
                    'metric': 'CPUUtilization',
                    'threshold': 80,
                    'comparison': 'GreaterThanThreshold',
                    'evaluation_periods': 2,
                    'scaling_adjustment': 1,
                    'cooldown': 300
                },
                'cpu_scale_down': {
                    'metric': 'CPUUtilization',
                    'threshold': 20,
                    'comparison': 'LessThanThreshold',
                    'evaluation_periods': 5,
                    'scaling_adjustment': -1,
                    'cooldown': 300
                }
            },
            'rightsizing_policies': {
                'low_utilization_threshold': 20,
                'high_utilization_threshold': 80,
                'evaluation_period_hours': 24,
                'confidence_threshold': 0.8,
                'max_changes_per_day': 3
            },
            'safety_policies': {
                'change_window': {
                    'start_hour': 2,
                    'end_hour': 6,
                    'timezone': 'UTC',
                    'excluded_days': ['saturday', 'sunday']
                },
                'minimum_uptime_hours': 24,
                'maximum_cpu_for_downsizing': 60,
                'require_approval_for_production': True
            }
        }
        
        return policies
    
    def setup_automation_infrastructure(self) -> Dict:
        """Set up the infrastructure for automated resource management"""
        
        infrastructure = {
            'lambda_functions': self.create_automation_lambda_functions(),
            'cloudwatch_alarms': self.create_automation_alarms(),
            'eventbridge_rules': self.create_automation_event_rules(),
            'iam_roles': self.create_automation_iam_roles(),
            'step_functions': self.create_automation_workflows()
        }
        
        return infrastructure
    
    def create_automation_lambda_functions(self) -> List[Dict]:
        """Create Lambda functions for automation"""
        
        functions = [
            {
                'function_name': 'resource-optimizer',
                'description': 'Main function for resource optimization decisions',
                'runtime': 'python3.9',
                'timeout': 300,
                'memory_size': 512,
                'environment_variables': {
                    'CONFIDENCE_THRESHOLD': '0.8',
                    'DRY_RUN_MODE': 'false'
                }
            },
            {
                'function_name': 'safety-checker',
                'description': 'Performs safety checks before automation actions',
                'runtime': 'python3.9',
                'timeout': 60,
                'memory_size': 256
            },
            {
                'function_name': 'cost-calculator',
                'description': 'Calculates cost impacts of optimization decisions',
                'runtime': 'python3.9',
                'timeout': 120,
                'memory_size': 256
            }
        ]
        
        return functions
    
    def monitor_automation_performance(self) -> Dict:
        """Monitor the performance of automation systems"""
        
        performance_metrics = {
            'decisions_made': len(self.decision_history),
            'successful_executions': len([d for d in self.decision_history if d.execution_timestamp]),
            'total_cost_savings': sum(d.expected_cost_impact for d in self.decision_history if d.expected_cost_impact < 0),
            'average_confidence_score': np.mean([d.confidence_score for d in self.decision_history]) if self.decision_history else 0,
            'safety_check_pass_rate': len([d for d in self.decision_history if d.safety_checks_passed]) / len(self.decision_history) if self.decision_history else 0
        }
        
        return performance_metrics

Auto_Scaling_Configuration:
  auto_scaling_group: "web-servers-asg"
  
  scaling_policies:
    scale_up_policy:
      policy_name: "cpu-scale-up"
      policy_type: "TargetTrackingScaling"
      target_tracking_configuration:
        target_value: 70.0
        predefined_metric_specification:
          predefined_metric_type: "ASGAverageCPUUtilization"
        scale_out_cooldown: 300
        scale_in_cooldown: 300
        
    predictive_scaling:
      policy_name: "predictive-scaling"
      policy_type: "PredictiveScaling"
      predictive_scaling_configuration:
        metric_specifications:
          - target_value: 70.0
            predefined_metric_specification:
              predefined_metric_type: "ASGAverageCPUUtilization"
        mode: "ForecastAndScale"
        scheduling_buffer_time: 300
        
  cost_optimization:
    mixed_instances_policy:
      instances_distribution:
        on_demand_base_capacity: 2
        on_demand_percentage_above_base_capacity: 25
        spot_allocation_strategy: "diversified"
      launch_template:
        launch_template_specification:
          launch_template_name: "cost-optimized-template"
          version: "$Latest"
        overrides:
          - instance_type: "m5.large"
            weighted_capacity: 1
          - instance_type: "m5a.large"
            weighted_capacity: 1
          - instance_type: "m4.large"
            weighted_capacity: 1

def create_rightsizing_automation():
    """Create automated rightsizing configuration"""
    
    config = {
        'rightsizing_schedule': {
            'frequency': 'daily',
            'time': '02:00',
            'timezone': 'UTC'
        },
        'analysis_parameters': {
            'lookback_period_days': 14,
            'minimum_data_points': 336,  # 14 days * 24 hours
            'confidence_threshold': 0.8,
            'utilization_thresholds': {
                'cpu_low': 20,
                'cpu_high': 80,
                'memory_low': 30,
                'memory_high': 85
            }
        },
        'execution_parameters': {
            'max_changes_per_run': 5,
            'change_window': {
                'start': '02:00',
                'end': '06:00',
                'days': ['monday', 'tuesday', 'wednesday', 'thursday', 'friday']
            },
            'safety_checks': {
                'require_health_check': True,
                'minimum_uptime_hours': 24,
                'exclude_production_without_approval': True
            }
        },
        'notification_settings': {
            'sns_topic_arn': 'arn:aws:sns:us-east-1:123456789012:rightsizing-notifications',
            'notify_on_decisions': True,
            'notify_on_executions': True,
            'notify_on_failures': True
        }
    }
    
    return config