Cloud Cost and Security Optimization: Balancing Budget an...

requirements.txt:

boto3>=1.34.0
python-dateutil>=2.8.2

Complete Cloud Cost and Security Optimization Manager:

#!/usr/bin/env python3
"""
Cloud Cost & Security Optimization - Cost Analysis Manager
Production-ready cost analysis and optimization with security considerations
"""

import boto3
from botocore.exceptions import ClientError, BotoCoreError
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass, asdict, field
from enum import Enum
from datetime import datetime, timedelta
import logging
import os

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)


class OptimizationError(Exception):
    """Base exception for optimization errors."""
    pass


class CostCategory(Enum):
    """Cost categories."""
    COMPUTE = "compute"
    STORAGE = "storage"
    NETWORK = "network"
    DATABASE = "database"
    SECURITY = "security"
    OTHER = "other"


@dataclass
class CostAnalysis:
    """Cost analysis result."""
    total_cost: float
    start_date: datetime
    end_date: datetime
    costs_by_service: Dict[str, float]
    costs_by_category: Dict[CostCategory, float]
    trends: List[Dict] = field(default_factory=list)
    
    def to_dict(self) -> Dict:
        """Convert to dictionary."""
        result = asdict(self)
        result['start_date'] = self.start_date.isoformat()
        result['end_date'] = self.end_date.isoformat()
        result['costs_by_category'] = {k.value: v for k, v in self.costs_by_category.items()}
        return result


@dataclass
class OptimizationOpportunity:
    """Optimization opportunity."""
    type: str
    description: str
    potential_savings: float
    security_impact: str
    implementation_effort: str
    recommendations: List[str] = field(default_factory=list)
    
    def to_dict(self) -> Dict:
        """Convert to dictionary."""
        return asdict(self)


class CloudCostOptimizer:
    """Manages cloud cost optimization with security considerations."""
    
    def __init__(
        self,
        region_name: str = 'us-east-1',
        aws_access_key_id: Optional[str] = None,
        aws_secret_access_key: Optional[str] = None
    ):
        """Initialize cost optimizer.
        
        Args:
            region_name: AWS region (default: us-east-1)
            aws_access_key_id: AWS access key (defaults to env/credentials)
            aws_secret_access_key: AWS secret key (defaults to env/credentials)
        """
        self.region_name = region_name
        
        try:
            session = boto3.Session(
                aws_access_key_id=aws_access_key_id or os.getenv('AWS_ACCESS_KEY_ID'),
                aws_secret_access_key=aws_secret_access_key or os.getenv('AWS_SECRET_ACCESS_KEY'),
                region_name=region_name
            )
            
            self.ce = session.client('ce')  # Cost Explorer
            self.ec2 = session.client('ec2', region_name=region_name)
            self.s3 = session.client('s3', region_name=region_name)
            self.rds = session.client('rds', region_name=region_name)
            
            logger.info(f"Initialized CloudCostOptimizer for region: {region_name}")
            
        except (ClientError, BotoCoreError) as e:
            error_msg = f"Failed to initialize AWS clients: {e}"
            logger.error(error_msg)
            raise OptimizationError(error_msg) from e
    
    def analyze_costs(
        self,
        start_date: str,
        end_date: str,
        granularity: str = 'DAILY'
    ) -> CostAnalysis:
        """Analyze cloud costs comprehensively.
        
        Args:
            start_date: Start date (YYYY-MM-DD)
            end_date: End date (YYYY-MM-DD)
            granularity: MONTHLY, DAILY, or HOURLY
            
        Returns:
            CostAnalysis object with detailed breakdown
            
        Raises:
            OptimizationError: If cost analysis fails
        """
        try:
            logger.info(f"Analyzing costs from {start_date} to {end_date}")
            
            # Get cost and usage data
            response = self.ce.get_cost_and_usage(
                TimePeriod={
                    'Start': start_date,
                    'End': end_date
                },
                Granularity=granularity,
                Metrics=['BlendedCost', 'UnblendedCost'],
                GroupBy=[
                    {'Type': 'DIMENSION', 'Key': 'SERVICE'},
                    {'Type': 'DIMENSION', 'Key': 'USAGE_TYPE'}
                ]
            )
            
            # Parse results
            total_cost = 0.0
            costs_by_service: Dict[str, float] = {}
            costs_by_category: Dict[CostCategory, float] = {
                cat: 0.0 for cat in CostCategory
            }
            trends = []
            
            for result in response.get('ResultsByTime', []):
                time_period = result['TimePeriod']
                total_period = 0.0
                
                for group in result.get('Groups', []):
                    service = group['Keys'][0] if group['Keys'] else 'Unknown'
                    amount = float(group['Metrics']['BlendedCost']['Amount'])
                    
                    total_cost += amount
                    total_period += amount
                    
                    # Accumulate by service
                    costs_by_service[service] = costs_by_service.get(service, 0) + amount
                    
                    # Categorize
                    category = self._categorize_service(service)
                    costs_by_category[category] = costs_by_category.get(category, 0) + amount
                
                trends.append({
                    'date': time_period['Start'],
                    'cost': total_period
                })
            
            analysis = CostAnalysis(
                total_cost=total_cost,
                start_date=datetime.strptime(start_date, '%Y-%m-%d'),
                end_date=datetime.strptime(end_date, '%Y-%m-%d'),
                costs_by_service=costs_by_service,
                costs_by_category=costs_by_category,
                trends=trends
            )
            
            logger.info(f"Cost analysis complete: Total cost ${total_cost:,.2f}")
            
            return analysis
        
        except ClientError as e:
            error_msg = f"Failed to analyze costs: {e}"
            logger.error(error_msg)
            raise OptimizationError(error_msg) from e
        except Exception as e:
            error_msg = f"Unexpected error analyzing costs: {e}"
            logger.error(error_msg, exc_info=True)
            raise OptimizationError(error_msg) from e
    
    def _categorize_service(self, service: str) -> CostCategory:
        """Categorize AWS service by cost type.
        
        Args:
            service: AWS service name
            
        Returns:
            CostCategory
        """
        service_lower = service.lower()
        
        if 'ec2' in service_lower or 'lambda' in service_lower or 'fargate' in service_lower:
            return CostCategory.COMPUTE
        elif 's3' in service_lower or 'ebs' in service_lower or 'glacier' in service_lower:
            return CostCategory.STORAGE
        elif 'data-transfer' in service_lower or 'cloudfront' in service_lower:
            return CostCategory.NETWORK
        elif 'rds' in service_lower or 'dynamodb' in service_lower or 'redshift' in service_lower:
            return CostCategory.DATABASE
        elif 'guardduty' in service_lower or 'inspector' in service_lower or 'security' in service_lower:
            return CostCategory.SECURITY
        else:
            return CostCategory.OTHER
    
    def find_optimization_opportunities(
        self,
        cost_analysis: Optional[CostAnalysis] = None,
        start_date: Optional[str] = None,
        end_date: Optional[str] = None
    ) -> List[OptimizationOpportunity]:
        """Find cost optimization opportunities with security considerations.
        
        Args:
            cost_analysis: Optional pre-computed cost analysis
            start_date: Start date for analysis (if cost_analysis not provided)
            end_date: End date for analysis (if cost_analysis not provided)
            
        Returns:
            List of OptimizationOpportunity objects
        """
        opportunities = []
        
        # Get cost analysis if not provided
        if cost_analysis is None:
            if not start_date or not end_date:
                # Default to last 30 days
                end = datetime.utcnow()
                start = end - timedelta(days=30)
                start_date = start.strftime('%Y-%m-%d')
                end_date = end.strftime('%Y-%m-%d')
            
            cost_analysis = self.analyze_costs(start_date, end_date)
        
        # Find unused resources
        unused_resources = self._find_unused_resources()
        if unused_resources:
            opportunities.append(OptimizationOpportunity(
                type="unused_resources",
                description=f"Found {len(unused_resources)} unused resources",
                potential_savings=sum(r.get('estimated_savings', 0) for r in unused_resources),
                security_impact="LOW - Removing unused resources improves security posture",
                implementation_effort="LOW",
                recommendations=[
                    "Delete unused resources",
                    "Implement automated cleanup policies",
                    "Monitor resource usage regularly"
                ]
            ))
        
        # Find over-provisioned instances
        over_provisioned = self._find_over_provisioned_instances()
        if over_provisioned:
            total_savings = sum(i.get('potential_savings', 0) for i in over_provisioned)
            opportunities.append(OptimizationOpportunity(
                type="right_sizing",
                description=f"Found {len(over_provisioned)} over-provisioned instances",
                potential_savings=total_savings,
                security_impact="LOW - Right-sizing improves resource efficiency without security impact",
                implementation_effort="MEDIUM",
                recommendations=[
                    "Review instance utilization metrics",
                    "Downsize instances with low utilization",
                    "Use CloudWatch metrics for decision-making"
                ]
            ))
        
        # Find reserved instance opportunities
        ri_opportunities = self._find_reserved_instance_opportunities()
        if ri_opportunities:
            total_savings = sum(ri.get('potential_savings', 0) for ri in ri_opportunities)
            opportunities.append(OptimizationOpportunity(
                type="reserved_instances",
                description=f"Reserved instance opportunities for {len(ri_opportunities)} instances",
                potential_savings=total_savings,
                security_impact="NONE - Reserved instances don't affect security",
                implementation_effort="LOW",
                recommendations=[
                    "Purchase Reserved Instances for predictable workloads",
                    "Consider Savings Plans for flexibility",
                    "Monitor usage patterns before committing"
                ]
            ))
        
        # Find storage optimization opportunities
        storage_opportunities = self._find_storage_optimizations()
        if storage_opportunities:
            total_savings = sum(s.get('potential_savings', 0) for s in storage_opportunities)
            opportunities.append(OptimizationOpportunity(
                type="storage_optimization",
                description="Storage optimization opportunities identified",
                potential_savings=total_savings,
                security_impact="LOW - Storage optimization maintains security controls",
                implementation_effort="MEDIUM",
                recommendations=[
                    "Move infrequently accessed data to cheaper storage classes",
                    "Enable S3 lifecycle policies",
                    "Review and delete unused snapshots"
                ]
            ))
        
        # Security-focused optimizations (don't compromise security)
        security_optimizations = self._find_security_optimizations(cost_analysis)
        if security_optimizations:
            opportunities.extend(security_optimizations)
        
        logger.info(f"Found {len(opportunities)} optimization opportunities")
        
        return opportunities
    
    def _find_unused_resources(self) -> List[Dict]:
        """Find unused resources.
        
        Returns:
            List of unused resources with savings estimates
        """
        unused = []
        
        try:
            # Find stopped EC2 instances
            response = self.ec2.describe_instances(
                Filters=[
                    {'Name': 'instance-state-name', 'Values': ['stopped']}
                ]
            )
            
            for reservation in response.get('Reservations', []):
                for instance in reservation.get('Instances', []):
                    stopped_time = instance.get('LaunchTime', datetime.utcnow())
                    days_stopped = (datetime.utcnow() - stopped_time.replace(tzinfo=None)).days
                    
                    if days_stopped > 7:  # Stopped for more than a week
                        unused.append({
                            'resource_id': instance['InstanceId'],
                            'resource_type': 'ec2_instance',
                            'status': 'stopped',
                            'days_unused': days_stopped,
                            'estimated_savings': 0,  # Already not costing much when stopped
                            'recommendation': 'Terminate if not needed'
                        })
        
        except ClientError as e:
            logger.warning(f"Error finding unused resources: {e}")
        
        return unused
    
    def _find_over_provisioned_instances(self) -> List[Dict]:
        """Find over-provisioned EC2 instances.
        
        Returns:
            List of over-provisioned instances
        """
        # This would require CloudWatch metrics analysis
        # Simplified implementation
        return []
    
    def _find_reserved_instance_opportunities(self) -> List[Dict]:
        """Find Reserved Instance purchase opportunities.
        
        Returns:
            List of RI opportunities
        """
        # This would analyze instance usage patterns
        # Simplified implementation
        return []
    
    def _find_storage_optimizations(self) -> List[Dict]:
        """Find storage optimization opportunities.
        
        Returns:
            List of storage optimizations
        """
        optimizations = []
        
        try:
            # Find old S3 objects that could be moved to cheaper storage
            response = self.s3.list_buckets()
            
            for bucket in response.get('Buckets', []):
                bucket_name = bucket['Name']
                # Check for objects in Standard storage that could be moved to IA or Glacier
                # This is simplified - full implementation would analyze object access patterns
                pass
        
        except ClientError as e:
            logger.warning(f"Error finding storage optimizations: {e}")
        
        return optimizations
    
    def _find_security_optimizations(self, cost_analysis: CostAnalysis) -> List[OptimizationOpportunity]:
        """Find security-focused optimizations that don't compromise security.
        
        Args:
            cost_analysis: Cost analysis results
            
        Returns:
            List of security optimization opportunities
        """
        opportunities = []
        
        # Optimize security tool usage without compromising protection
        security_cost = cost_analysis.costs_by_category.get(CostCategory.SECURITY, 0)
        if security_cost > 100:  # If spending significant amount on security
            opportunities.append(OptimizationOpportunity(
                type="security_tool_optimization",
                description="Optimize security tool usage",
                potential_savings=security_cost * 0.1,  # 10% savings potential
                security_impact="NONE - Optimize without reducing protection",
                implementation_effort="MEDIUM",
                recommendations=[
                    "Review security tool coverage for redundancy",
                    "Consolidate overlapping security services",
                    "Use AWS Security Hub for unified view"
                ]
            ))
        
        return opportunities


# Example usage
if __name__ == "__main__":
    optimizer = CloudCostOptimizer(region_name='us-east-1')
    
    # Analyze costs for last 30 days
    end_date = datetime.utcnow()
    start_date = end_date - timedelta(days=30)
    
    analysis = optimizer.analyze_costs(
        start_date=start_date.strftime('%Y-%m-%d'),
        end_date=end_date.strftime('%Y-%m-%d')
    )
    
    print(f"Total Cost: ${analysis.total_cost:,.2f}")
    print(f"By Service: {dict(sorted(analysis.costs_by_service.items(), key=lambda x: x[1], reverse=True)[:5])}")
    
    # Find optimization opportunities
    opportunities = optimizer.find_optimization_opportunities(cost_analysis=analysis)
    
    print(f"\nFound {len(opportunities)} optimization opportunities:")
    for opp in opportunities:
        print(f"- {opp.type}: ${opp.potential_savings:,.2f} potential savings")
        print(f"  Security Impact: {opp.security_impact}")

See Step 1 for the complete implementation. The CloudCostOptimizer class includes the find_optimization_opportunities() method that comprehensively analyzes costs and identifies optimization opportunities with security considerations.

Example usage:

optimizer = CloudCostOptimizer(region_name='us-east-1')

# Analyze costs
analysis = optimizer.analyze_costs('2024-01-01', '2024-01-31')

# Find optimization opportunities
opportunities = optimizer.find_optimization_opportunities(cost_analysis=analysis)

for opp in opportunities:
    print(f"{opp.type}: ${opp.potential_savings:,.2f} savings")
    print(f"Security Impact: {opp.security_impact}")
    print(f"Recommendations: {opp.recommendations}")