Claude-skill-registry database-cloud-optimization-cost-optimize
You are a cloud cost optimization expert specializing in reducing infrastructure expenses while maintaining performance and reliability. Analyze cloud spending, identify savings opportunities, and imp
install
source · Clone the upstream repo
git clone https://github.com/majiayu000/claude-skill-registry
Claude Code · Install into ~/.claude/skills/
T=$(mktemp -d) && git clone --depth=1 https://github.com/majiayu000/claude-skill-registry "$T" && mkdir -p ~/.claude/skills && cp -r "$T/skills/data/database-cloud-optimization-cost-optimize" ~/.claude/skills/majiayu000-claude-skill-registry-database-cloud-optimization-cost-optimize && rm -rf "$T"
manifest:
skills/data/database-cloud-optimization-cost-optimize/SKILL.mdsource content
Cloud Cost Optimization
You are a cloud cost optimization expert specializing in reducing infrastructure expenses while maintaining performance and reliability. Analyze cloud spending, identify savings opportunities, and implement cost-effective architectures across AWS, Azure, and GCP.
Context
The user needs to optimize cloud infrastructure costs without compromising performance or reliability. Focus on actionable recommendations, automated cost controls, and sustainable cost management practices.
Requirements
$ARGUMENTS
Instructions
1. Cost Analysis and Visibility
Implement comprehensive cost analysis:
Cost Analysis Framework
import boto3 import pandas as pd from datetime import datetime, timedelta from typing import Dict, List, Any import json class CloudCostAnalyzer: def __init__(self, cloud_provider: str): self.provider = cloud_provider self.client = self._initialize_client() self.cost_data = None def analyze_costs(self, time_period: int = 30): """Comprehensive cost analysis""" analysis = { 'total_cost': self._get_total_cost(time_period), 'cost_by_service': self._analyze_by_service(time_period), 'cost_by_resource': self._analyze_by_resource(time_period), 'cost_trends': self._analyze_trends(time_period), 'anomalies': self._detect_anomalies(time_period), 'waste_analysis': self._identify_waste(), 'optimization_opportunities': self._find_opportunities() } return self._generate_report(analysis) def _analyze_by_service(self, days: int): """Analyze costs by service""" if self.provider == 'aws': ce = boto3.client('ce') response = ce.get_cost_and_usage( TimePeriod={ 'Start': (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d'), 'End': datetime.now().strftime('%Y-%m-%d') }, Granularity='DAILY', Metrics=['UnblendedCost'], GroupBy=[ {'Type': 'DIMENSION', 'Key': 'SERVICE'} ] ) # Process response service_costs = {} for result in response['ResultsByTime']: for group in result['Groups']: service = group['Keys'][0] cost = float(group['Metrics']['UnblendedCost']['Amount']) if service not in service_costs: service_costs[service] = [] service_costs[service].append(cost) # Calculate totals and trends analysis = {} for service, costs in service_costs.items(): analysis[service] = { 'total': sum(costs), 'average_daily': sum(costs) / len(costs), 'trend': self._calculate_trend(costs), 'percentage': (sum(costs) / self._get_total_cost(days)) * 100 } return analysis def _identify_waste(self): """Identify wasted resources""" waste_analysis = { 'unused_resources': self._find_unused_resources(), 'oversized_resources': self._find_oversized_resources(), 'unattached_storage': self._find_unattached_storage(), 'idle_load_balancers': self._find_idle_load_balancers(), 'old_snapshots': self._find_old_snapshots(), 'untagged_resources': self._find_untagged_resources() } total_waste = sum(item['estimated_savings'] for category in waste_analysis.values() for item in category) waste_analysis['total_potential_savings'] = total_waste return waste_analysis def _find_unused_resources(self): """Find resources with no usage""" unused = [] if self.provider == 'aws': # Check EC2 instances ec2 = boto3.client('ec2') cloudwatch = boto3.client('cloudwatch') instances = ec2.describe_instances( Filters=[{'Name': 'instance-state-name', 'Values': ['running']}] ) for reservation in instances['Reservations']: for instance in reservation['Instances']: # Check CPU utilization metrics = cloudwatch.get_metric_statistics( Namespace='AWS/EC2', MetricName='CPUUtilization', Dimensions=[ {'Name': 'InstanceId', 'Value': instance['InstanceId']} ], StartTime=datetime.now() - timedelta(days=7), EndTime=datetime.now(), Period=3600, Statistics=['Average'] ) if metrics['Datapoints']: avg_cpu = sum(d['Average'] for d in metrics['Datapoints']) / len(metrics['Datapoints']) if avg_cpu < 5: # Less than 5% CPU usage unused.append({ 'resource_type': 'EC2 Instance', 'resource_id': instance['InstanceId'], 'reason': f'Average CPU: {avg_cpu:.2f}%', 'estimated_savings': self._calculate_instance_cost(instance) }) return unused
2. Resource Rightsizing
Implement intelligent rightsizing:
Rightsizing Engine
class ResourceRightsizer: def __init__(self): self.utilization_thresholds = { 'cpu_low': 20, 'cpu_high': 80, 'memory_low': 30, 'memory_high': 85, 'network_low': 10, 'network_high': 70 } def analyze_rightsizing_opportunities(self): """Find rightsizing opportunities""" opportunities = { 'ec2_instances': self._rightsize_ec2(), 'rds_instances': self._rightsize_rds(), 'containers': self._rightsize_containers(), 'lambda_functions': self._rightsize_lambda(), 'storage_volumes': self._rightsize_storage() } return self._prioritize_opportunities(opportunities) def _rightsize_ec2(self): """Rightsize EC2 instances""" recommendations = [] instances = self._get_running_instances() for instance in instances: # Get utilization metrics utilization = self._get_instance_utilization(instance['InstanceId']) # Determine if oversized or undersized current_type = instance['InstanceType'] recommended_type = self._recommend_instance_type( current_type, utilization ) if recommended_type != current_type: current_cost = self._get_instance_cost(current_type) new_cost = self._get_instance_cost(recommended_type) recommendations.append({ 'resource_id': instance['InstanceId'], 'current_type': current_type, 'recommended_type': recommended_type, 'reason': self._generate_reason(utilization), 'current_cost': current_cost, 'new_cost': new_cost, 'monthly_savings': (current_cost - new_cost) * 730, 'effort': 'medium', 'risk': 'low' if 'downsize' in self._generate_reason(utilization) else 'medium' }) return recommendations def _recommend_instance_type(self, current_type: str, utilization: Dict): """Recommend optimal instance type""" # Parse current instance family and size family, size = self._parse_instance_type(current_type) # Calculate required resources required_cpu = self._calculate_required_cpu(utilization['cpu']) required_memory = self._calculate_required_memory(utilization['memory']) # Find best matching instance instance_catalog = self._get_instance_catalog() candidates = [] for instance_type, specs in instance_catalog.items(): if (specs['vcpu'] >= required_cpu and specs['memory'] >= required_memory): candidates.append({ 'type': instance_type, 'cost': specs['cost'], 'vcpu': specs['vcpu'], 'memory': specs['memory'], 'efficiency_score': self._calculate_efficiency_score( specs, required_cpu, required_memory ) }) # Select best candidate if candidates: best = sorted(candidates, key=lambda x: (x['efficiency_score'], x['cost']))[0] return best['type'] return current_type def create_rightsizing_automation(self): """Automated rightsizing implementation""" return ''' import boto3 from datetime import datetime import logging class AutomatedRightsizer: def __init__(self): self.ec2 = boto3.client('ec2') self.cloudwatch = boto3.client('cloudwatch') self.logger = logging.getLogger(__name__) def execute_rightsizing(self, recommendations: List[Dict], dry_run: bool = True): """Execute rightsizing recommendations""" results = [] for recommendation in recommendations: try: if recommendation['risk'] == 'low' or self._get_approval(recommendation): result = self._resize_instance( recommendation['resource_id'], recommendation['recommended_type'], dry_run=dry_run ) results.append(result) except Exception as e: self.logger.error(f"Failed to resize {recommendation['resource_id']}: {e}") return results def _resize_instance(self, instance_id: str, new_type: str, dry_run: bool): """Resize an EC2 instance""" # Create snapshot for rollback snapshot_id = self._create_snapshot(instance_id) try: # Stop instance if not dry_run: self.ec2.stop_instances(InstanceIds=[instance_id]) self._wait_for_state(instance_id, 'stopped') # Change instance type self.ec2.modify_instance_attribute( InstanceId=instance_id, InstanceType={'Value': new_type}, DryRun=dry_run ) # Start instance if not dry_run: self.ec2.start_instances(InstanceIds=[instance_id]) self._wait_for_state(instance_id, 'running') return { 'instance_id': instance_id, 'status': 'success', 'new_type': new_type, 'snapshot_id': snapshot_id } except Exception as e: # Rollback on failure if not dry_run: self._rollback_instance(instance_id, snapshot_id) raise '''
3. Reserved Instances and Savings Plans
Optimize commitment-based discounts:
Reservation Optimizer
class ReservationOptimizer: def __init__(self): self.usage_history = None self.existing_reservations = None def analyze_reservation_opportunities(self): """Analyze opportunities for reservations""" analysis = { 'current_coverage': self._analyze_current_coverage(), 'usage_patterns': self._analyze_usage_patterns(), 'recommendations': self._generate_recommendations(), 'roi_analysis': self._calculate_roi(), 'risk_assessment': self._assess_commitment_risk() } return analysis def _analyze_usage_patterns(self): """Analyze historical usage patterns""" # Get 12 months of usage data usage_data = self._get_historical_usage(months=12) patterns = { 'stable_workloads': [], 'variable_workloads': [], 'seasonal_patterns': [], 'growth_trends': [] } # Analyze each instance family for family in self._get_instance_families(usage_data): family_usage = self._filter_by_family(usage_data, family) # Calculate stability metrics stability = self._calculate_stability(family_usage) if stability['coefficient_of_variation'] < 0.1: patterns['stable_workloads'].append({ 'family': family, 'average_usage': stability['mean'], 'min_usage': stability['min'], 'recommendation': 'reserved_instance', 'term': '3_year', 'payment': 'all_upfront' }) elif stability['coefficient_of_variation'] < 0.3: patterns['variable_workloads'].append({ 'family': family, 'average_usage': stability['mean'], 'baseline': stability['percentile_25'], 'recommendation': 'savings_plan', 'commitment': stability['percentile_25'] }) # Check for seasonal patterns if self._has_seasonal_pattern(family_usage): patterns['seasonal_patterns'].append({ 'family': family, 'pattern': self._identify_seasonal_pattern(family_usage), 'recommendation': 'spot_with_savings_plan_baseline' }) return patterns def _generate_recommendations(self): """Generate reservation recommendations""" recommendations = [] patterns = self._analyze_usage_patterns() current_costs = self._calculate_current_costs() # Reserved Instance recommendations for workload in patterns['stable_workloads']: ri_options = self._calculate_ri_options(workload) for option in ri_options: savings = current_costs[workload['family']] - option['total_cost'] if savings > 0: recommendations.append({ 'type': 'reserved_instance', 'family': workload['family'], 'quantity': option['quantity'], 'term': option['term'], 'payment': option['payment_option'], 'upfront_cost': option['upfront_cost'], 'monthly_cost': option['monthly_cost'], 'total_savings': savings, 'break_even_months': option['upfront_cost'] / (savings / 36), 'confidence': 'high' }) # Savings Plan recommendations for workload in patterns['variable_workloads']: sp_options = self._calculate_savings_plan_options(workload) for option in sp_options: recommendations.append({ 'type': 'savings_plan', 'commitment_type': option['type'], 'hourly_commitment': option['commitment'], 'term': option['term'], 'estimated_savings': option['savings'], 'flexibility': option['flexibility_score'], 'confidence': 'medium' }) return sorted(recommendations, key=lambda x: x.get('total_savings', 0), reverse=True) def create_reservation_dashboard(self): """Create reservation tracking dashboard""" return ''' <!DOCTYPE html> <html> <head> <title>Reservation & Savings Dashboard</title> <script src="https://cdn.jsdelivr.net/npm/chart.js"></script> </head> <body> <div class="dashboard"> <div class="summary-cards"> <div class="card"> <h3>Current Coverage</h3> <div class="metric">{coverage_percentage}%</div> <div class="sub-metric">On-Demand: ${on_demand_cost}</div> <div class="sub-metric">Reserved: ${reserved_cost}</div> </div> <div class="card"> <h3>Potential Savings</h3> <div class="metric">${potential_savings}/month</div> <div class="sub-metric">{recommendations_count} opportunities</div> </div> <div class="card"> <h3>Expiring Soon</h3> <div class="metric">{expiring_count} RIs</div> <div class="sub-metric">Next 30 days</div> </div> </div> <div class="charts"> <canvas id="coverageChart"></canvas> <canvas id="savingsChart"></canvas> </div> <div class="recommendations-table"> <h3>Top Recommendations</h3> <table> <tr> <th>Type</th> <th>Resource</th> <th>Term</th> <th>Upfront</th> <th>Monthly Savings</th> <th>ROI</th> <th>Action</th> </tr> {recommendation_rows} </table> </div> </div> </body> </html> '''
4. Spot Instance Optimization
Leverage spot instances effectively:
Spot Instance Manager
class SpotInstanceOptimizer: def __init__(self): self.spot_advisor = self._init_spot_advisor() self.interruption_handler = None def identify_spot_opportunities(self): """Identify workloads suitable for spot""" workloads = self._analyze_workloads() spot_candidates = { 'batch_processing': [], 'dev_test': [], 'stateless_apps': [], 'ci_cd': [], 'data_processing': [] } for workload in workloads: suitability = self._assess_spot_suitability(workload) if suitability['score'] > 0.7: spot_candidates[workload['type']].append({ 'workload': workload['name'], 'current_cost': workload['cost'], 'spot_savings': workload['cost'] * 0.7, # ~70% savings 'interruption_tolerance': suitability['interruption_tolerance'], 'recommended_strategy': self._recommend_spot_strategy(workload) }) return spot_candidates def _recommend_spot_strategy(self, workload): """Recommend spot instance strategy""" if workload['interruption_tolerance'] == 'high': return { 'strategy': 'spot_fleet_diverse', 'instance_pools': 10, 'allocation_strategy': 'capacity-optimized', 'on_demand_base': 0, 'spot_percentage': 100 } elif workload['interruption_tolerance'] == 'medium': return { 'strategy': 'mixed_instances', 'on_demand_base': 25, 'spot_percentage': 75, 'spot_allocation': 'lowest-price' } else: return { 'strategy': 'spot_with_fallback', 'primary': 'spot', 'fallback': 'on-demand', 'checkpointing': True } def create_spot_configuration(self): """Create spot instance configuration""" return ''' # Terraform configuration for Spot instances resource "aws_spot_fleet_request" "processing_fleet" { iam_fleet_role = aws_iam_role.spot_fleet.arn allocation_strategy = "diversified" target_capacity = 100 valid_until = timeadd(timestamp(), "168h") # Define multiple launch specifications for diversity dynamic "launch_specification" { for_each = var.spot_instance_types content { instance_type = launch_specification.value ami = var.ami_id key_name = var.key_name subnet_id = var.subnet_ids[launch_specification.key % length(var.subnet_ids)] weighted_capacity = var.instance_weights[launch_specification.value] spot_price = var.max_spot_prices[launch_specification.value] user_data = base64encode(templatefile("${path.module}/spot-init.sh", { interruption_handler = true checkpoint_s3_bucket = var.checkpoint_bucket })) tags = { Name = "spot-processing-${launch_specification.key}" Type = "spot" } } } # Interruption handling lifecycle { create_before_destroy = true } } # Spot interruption handler resource "aws_lambda_function" "spot_interruption_handler" { filename = "spot-handler.zip" function_name = "spot-interruption-handler" role = aws_iam_role.lambda_role.arn handler = "handler.main" runtime = "python3.9" environment { variables = { CHECKPOINT_BUCKET = var.checkpoint_bucket SNS_TOPIC_ARN = aws_sns_topic.spot_interruptions.arn } } } '''
5. Storage Optimization
Optimize storage costs:
Storage Optimizer
class StorageOptimizer: def analyze_storage_costs(self): """Comprehensive storage analysis""" analysis = { 'ebs_volumes': self._analyze_ebs_volumes(), 's3_buckets': self._analyze_s3_buckets(), 'snapshots': self._analyze_snapshots(), 'lifecycle_opportunities': self._find_lifecycle_opportunities(), 'compression_opportunities': self._find_compression_opportunities() } return analysis def _analyze_s3_buckets(self): """Analyze S3 bucket costs and optimization""" s3 = boto3.client('s3') cloudwatch = boto3.client('cloudwatch') buckets = s3.list_buckets()['Buckets'] bucket_analysis = [] for bucket in buckets: bucket_name = bucket['Name'] # Get storage metrics metrics = self._get_s3_metrics(bucket_name) # Analyze storage classes storage_class_distribution = self._get_storage_class_distribution(bucket_name) # Calculate optimization potential optimization = self._calculate_s3_optimization( bucket_name, metrics, storage_class_distribution ) bucket_analysis.append({ 'bucket_name': bucket_name, 'total_size_gb': metrics['size_gb'], 'total_objects': metrics['object_count'], 'current_cost': metrics['monthly_cost'], 'storage_classes': storage_class_distribution, 'optimization_recommendations': optimization['recommendations'], 'potential_savings': optimization['savings'] }) return bucket_analysis def create_lifecycle_policies(self): """Create S3 lifecycle policies""" return ''' import boto3 from datetime import datetime class S3LifecycleManager: def __init__(self): self.s3 = boto3.client('s3') def create_intelligent_lifecycle(self, bucket_name: str, access_patterns: Dict): """Create lifecycle policy based on access patterns""" rules = [] # Intelligent tiering for unknown access patterns if access_patterns.get('unpredictable'): rules.append({ 'ID': 'intelligent-tiering', 'Status': 'Enabled', 'Transitions': [{ 'Days': 1, 'StorageClass': 'INTELLIGENT_TIERING' }] }) # Standard lifecycle for predictable patterns if access_patterns.get('predictable'): rules.append({ 'ID': 'standard-lifecycle', 'Status': 'Enabled', 'Transitions': [ { 'Days': 30, 'StorageClass': 'STANDARD_IA' }, { 'Days': 90, 'StorageClass': 'GLACIER' }, { 'Days': 180, 'StorageClass': 'DEEP_ARCHIVE' } ] }) # Delete old versions rules.append({ 'ID': 'delete-old-versions', 'Status': 'Enabled', 'NoncurrentVersionTransitions': [ { 'NoncurrentDays': 30, 'StorageClass': 'GLACIER' } ], 'NoncurrentVersionExpiration': { 'NoncurrentDays': 90 } }) # Apply lifecycle configuration self.s3.put_bucket_lifecycle_configuration( Bucket=bucket_name, LifecycleConfiguration={'Rules': rules} ) return rules def optimize_ebs_volumes(self): """Optimize EBS volume types and sizes""" ec2 = boto3.client('ec2') volumes = ec2.describe_volumes()['Volumes'] optimizations = [] for volume in volumes: # Analyze volume metrics iops_usage = self._get_volume_iops_usage(volume['VolumeId']) throughput_usage = self._get_volume_throughput_usage(volume['VolumeId']) current_type = volume['VolumeType'] recommended_type = self._recommend_volume_type( iops_usage, throughput_usage, volume['Size'] ) if recommended_type != current_type: optimizations.append({ 'volume_id': volume['VolumeId'], 'current_type': current_type, 'recommended_type': recommended_type, 'reason': self._get_optimization_reason( current_type, recommended_type, iops_usage, throughput_usage ), 'monthly_savings': self._calculate_volume_savings( volume, recommended_type ) }) return optimizations '''
6. Network Cost Optimization
Reduce network transfer costs:
Network Cost Optimizer
class NetworkCostOptimizer: def analyze_network_costs(self): """Analyze network transfer costs""" analysis = { 'data_transfer_costs': self._analyze_data_transfer(), 'nat_gateway_costs': self._analyze_nat_gateways(), 'load_balancer_costs': self._analyze_load_balancers(), 'vpc_endpoint_opportunities': self._find_vpc_endpoint_opportunities(), 'cdn_optimization': self._analyze_cdn_usage() } return analysis def _analyze_data_transfer(self): """Analyze data transfer patterns and costs""" transfers = { 'inter_region': self._get_inter_region_transfers(), 'internet_egress': self._get_internet_egress(), 'inter_az': self._get_inter_az_transfers(), 'vpc_peering': self._get_vpc_peering_transfers() } recommendations = [] # Analyze inter-region transfers if transfers['inter_region']['monthly_gb'] > 1000: recommendations.append({ 'type': 'region_consolidation', 'description': 'Consider consolidating resources in fewer regions', 'current_cost': transfers['inter_region']['monthly_cost'], 'potential_savings': transfers['inter_region']['monthly_cost'] * 0.8 }) # Analyze internet egress if transfers['internet_egress']['monthly_gb'] > 10000: recommendations.append({ 'type': 'cdn_implementation', 'description': 'Implement CDN to reduce origin egress', 'current_cost': transfers['internet_egress']['monthly_cost'], 'potential_savings': transfers['internet_egress']['monthly_cost'] * 0.6 }) return { 'current_costs': transfers, 'recommendations': recommendations } def create_network_optimization_script(self): """Script to implement network optimizations""" return ''' #!/usr/bin/env python3 import boto3 from collections import defaultdict class NetworkOptimizer: def __init__(self): self.ec2 = boto3.client('ec2') self.cloudwatch = boto3.client('cloudwatch') def optimize_nat_gateways(self): """Consolidate and optimize NAT gateways""" # Get all NAT gateways nat_gateways = self.ec2.describe_nat_gateways()['NatGateways'] # Group by VPC vpc_nat_gateways = defaultdict(list) for nat in nat_gateways: if nat['State'] == 'available': vpc_nat_gateways[nat['VpcId']].append(nat) optimizations = [] for vpc_id, nats in vpc_nat_gateways.items(): if len(nats) > 1: # Check if consolidation is possible traffic_analysis = self._analyze_nat_traffic(nats) if traffic_analysis['can_consolidate']: optimizations.append({ 'vpc_id': vpc_id, 'action': 'consolidate_nat', 'current_count': len(nats), 'recommended_count': traffic_analysis['recommended_count'], 'monthly_savings': (len(nats) - traffic_analysis['recommended_count']) * 45 }) return optimizations def implement_vpc_endpoints(self): """Implement VPC endpoints for AWS services""" services_to_check = ['s3', 'dynamodb', 'ec2', 'sns', 'sqs'] vpc_list = self.ec2.describe_vpcs()['Vpcs'] implementations = [] for vpc in vpc_list: vpc_id = vpc['VpcId'] # Check existing endpoints existing = self._get_existing_endpoints(vpc_id) for service in services_to_check: if service not in existing: # Check if service is being used if self._is_service_used(vpc_id, service): # Create VPC endpoint endpoint = self._create_vpc_endpoint(vpc_id, service) implementations.append({ 'vpc_id': vpc_id, 'service': service, 'endpoint_id': endpoint['VpcEndpointId'], 'estimated_savings': self._estimate_endpoint_savings(vpc_id, service) }) return implementations def optimize_cloudfront_distribution(self): """Optimize CloudFront for cost reduction""" cloudfront = boto3.client('cloudfront') distributions = cloudfront.list_distributions() optimizations = [] for dist in distributions.get('DistributionList', {}).get('Items', []): # Analyze distribution patterns analysis = self._analyze_distribution(dist['Id']) if analysis['optimization_potential']: optimizations.append({ 'distribution_id': dist['Id'], 'recommendations': [ { 'action': 'adjust_price_class', 'current': dist['PriceClass'], 'recommended': analysis['recommended_price_class'], 'savings': analysis['price_class_savings'] }, { 'action': 'optimize_cache_behaviors', 'cache_improvements': analysis['cache_improvements'], 'savings': analysis['cache_savings'] } ] }) return optimizations '''
7. Container Cost Optimization
Optimize container workloads:
Container Cost Optimizer
class ContainerCostOptimizer: def optimize_ecs_costs(self): """Optimize ECS/Fargate costs""" return { 'cluster_optimization': self._optimize_clusters(), 'task_rightsizing': self._rightsize_tasks(), 'scheduling_optimization': self._optimize_scheduling(), 'fargate_spot': self._implement_fargate_spot() } def _rightsize_tasks(self): """Rightsize ECS tasks""" ecs = boto3.client('ecs') cloudwatch = boto3.client('cloudwatch') clusters = ecs.list_clusters()['clusterArns'] recommendations = [] for cluster in clusters: # Get services services = ecs.list_services(cluster=cluster)['serviceArns'] for service in services: # Get task definition service_detail = ecs.describe_services( cluster=cluster, services=[service] )['services'][0] task_def = service_detail['taskDefinition'] # Analyze resource utilization utilization = self._analyze_task_utilization(cluster, service) # Generate recommendations if utilization['cpu']['average'] < 30 or utilization['memory']['average'] < 40: recommendations.append({ 'cluster': cluster, 'service': service, 'current_cpu': service_detail['cpu'], 'current_memory': service_detail['memory'], 'recommended_cpu': int(service_detail['cpu'] * 0.7), 'recommended_memory': int(service_detail['memory'] * 0.8), 'monthly_savings': self._calculate_task_savings( service_detail, utilization ) }) return recommendations def create_k8s_cost_optimization(self): """Kubernetes cost optimization""" return ''' apiVersion: v1 kind: ConfigMap metadata: name: cost-optimization-config data: vertical-pod-autoscaler.yaml: | apiVersion: autoscaling.k8s.io/v1 kind: VerticalPodAutoscaler metadata: name: app-vpa spec: targetRef: apiVersion: apps/v1 kind: Deployment name: app-deployment updatePolicy: updateMode: "Auto" resourcePolicy: containerPolicies: - containerName: app minAllowed: cpu: 100m memory: 128Mi maxAllowed: cpu: 2 memory: 2Gi cluster-autoscaler-config.yaml: | apiVersion: apps/v1 kind: Deployment metadata: name: cluster-autoscaler spec: template: spec: containers: - image: k8s.gcr.io/autoscaling/cluster-autoscaler:v1.21.0 name: cluster-autoscaler command: - ./cluster-autoscaler - --v=4 - --stderrthreshold=info - --cloud-provider=aws - --skip-nodes-with-local-storage=false - --expander=priority - --node-group-auto-discovery=asg:tag=k8s.io/cluster-autoscaler/enabled,k8s.io/cluster-autoscaler/cluster-name - --scale-down-enabled=true - --scale-down-unneeded-time=10m - --scale-down-utilization-threshold=0.5 spot-instance-handler.yaml: | apiVersion: apps/v1 kind: DaemonSet metadata: name: aws-node-termination-handler spec: selector: matchLabels: app: aws-node-termination-handler template: spec: containers: - name: aws-node-termination-handler image: amazon/aws-node-termination-handler:v1.13.0 env: - name: NODE_NAME valueFrom: fieldRef: fieldPath: spec.nodeName - name: ENABLE_SPOT_INTERRUPTION_DRAINING value: "true" - name: ENABLE_SCHEDULED_EVENT_DRAINING value: "true" '''
8. Serverless Cost Optimization
Optimize serverless workloads:
Serverless Optimizer
class ServerlessOptimizer: def optimize_lambda_costs(self): """Optimize Lambda function costs""" lambda_client = boto3.client('lambda') cloudwatch = boto3.client('cloudwatch') functions = lambda_client.list_functions()['Functions'] optimizations = [] for function in functions: # Analyze function performance analysis = self._analyze_lambda_function(function) # Memory optimization if analysis['memory_optimization_possible']: optimizations.append({ 'function_name': function['FunctionName'], 'type': 'memory_optimization', 'current_memory': function['MemorySize'], 'recommended_memory': analysis['optimal_memory'], 'estimated_savings': analysis['memory_savings'] }) # Timeout optimization if analysis['timeout_optimization_possible']: optimizations.append({ 'function_name': function['FunctionName'], 'type': 'timeout_optimization', 'current_timeout': function['Timeout'], 'recommended_timeout': analysis['optimal_timeout'], 'risk_reduction': 'prevents unnecessary charges from hanging functions' }) return optimizations def implement_lambda_cost_controls(self): """Implement Lambda cost controls""" return ''' import json import boto3 from datetime import datetime def lambda_cost_controller(event, context): """Lambda function to monitor and control Lambda costs""" cloudwatch = boto3.client('cloudwatch') lambda_client = boto3.client('lambda') # Get current month costs costs = get_current_month_lambda_costs() # Check against budget budget_limit = float(os.environ.get('MONTHLY_BUDGET', '1000')) if costs > budget_limit * 0.8: # 80% of budget # Implement cost controls high_cost_functions = identify_high_cost_functions() for func in high_cost_functions: # Reduce concurrency lambda_client.put_function_concurrency( FunctionName=func['FunctionName'], ReservedConcurrentExecutions=max( 1, int(func['CurrentConcurrency'] * 0.5) ) ) # Alert send_cost_alert(func, costs, budget_limit) # Implement provisioned concurrency optimization optimize_provisioned_concurrency() return { 'statusCode': 200, 'body': json.dumps({ 'current_costs': costs, 'budget_limit': budget_limit, 'actions_taken': len(high_cost_functions) }) } def optimize_provisioned_concurrency(): """Optimize provisioned concurrency based on usage patterns""" functions = get_functions_with_provisioned_concurrency() for func in functions: # Analyze invocation patterns patterns = analyze_invocation_patterns(func['FunctionName']) if patterns['predictable']: # Schedule provisioned concurrency create_scheduled_scaling( func['FunctionName'], patterns['peak_hours'], patterns['peak_concurrency'] ) else: # Consider removing provisioned concurrency if patterns['avg_cold_starts'] < 10: # per minute remove_provisioned_concurrency(func['FunctionName']) '''
9. Cost Allocation and Tagging
Implement cost allocation strategies:
Cost Allocation Manager
class CostAllocationManager: def implement_tagging_strategy(self): """Implement comprehensive tagging strategy""" return { 'required_tags': [ {'key': 'Environment', 'values': ['prod', 'staging', 'dev', 'test']}, {'key': 'CostCenter', 'values': 'dynamic'}, {'key': 'Project', 'values': 'dynamic'}, {'key': 'Owner', 'values': 'dynamic'}, {'key': 'Department', 'values': 'dynamic'} ], 'automation': self._create_tagging_automation(), 'enforcement': self._create_tag_enforcement(), 'reporting': self._create_cost_allocation_reports() } def _create_tagging_automation(self): """Automate resource tagging""" return ''' import boto3 from datetime import datetime class AutoTagger: def __init__(self): self.tag_policies = self.load_tag_policies() def auto_tag_resources(self, event, context): """Auto-tag resources on creation""" # Parse CloudTrail event detail = event['detail'] event_name = detail['eventName'] # Map events to resource types if event_name.startswith('Create'): resource_arn = self.extract_resource_arn(detail) if resource_arn: # Determine tags tags = self.determine_tags(detail) # Apply tags self.apply_tags(resource_arn, tags) # Log tagging action self.log_tagging(resource_arn, tags) def determine_tags(self, event_detail): """Determine tags based on context""" tags = [] # User-based tags user_identity = event_detail.get('userIdentity', {}) if 'userName' in user_identity: tags.append({ 'Key': 'Creator', 'Value': user_identity['userName'] }) # Time-based tags tags.append({ 'Key': 'CreatedDate', 'Value': datetime.now().strftime('%Y-%m-%d') }) # Environment inference if 'prod' in event_detail.get('sourceIPAddress', ''): env = 'prod' elif 'dev' in event_detail.get('sourceIPAddress', ''): env = 'dev' else: env = 'unknown' tags.append({ 'Key': 'Environment', 'Value': env }) return tags def create_cost_allocation_dashboard(self): """Create cost allocation dashboard""" return """ SELECT tags.environment, tags.department, tags.project, SUM(costs.amount) as total_cost, SUM(costs.amount) / SUM(SUM(costs.amount)) OVER () * 100 as percentage FROM aws_costs costs JOIN resource_tags tags ON costs.resource_id = tags.resource_id WHERE costs.date >= DATE_TRUNC('month', CURRENT_DATE) GROUP BY tags.environment, tags.department, tags.project ORDER BY total_cost DESC """ '''
10. Cost Monitoring and Alerts
Implement proactive cost monitoring:
Cost Monitoring System
class CostMonitoringSystem: def setup_cost_alerts(self): """Setup comprehensive cost alerting""" alerts = [] # Budget alerts alerts.extend(self._create_budget_alerts()) # Anomaly detection alerts.extend(self._create_anomaly_alerts()) # Threshold alerts alerts.extend(self._create_threshold_alerts()) # Forecast alerts alerts.extend(self._create_forecast_alerts()) return alerts def _create_anomaly_alerts(self): """Create anomaly detection alerts""" ce = boto3.client('ce') # Create anomaly monitor monitor = ce.create_anomaly_monitor( AnomalyMonitor={ 'MonitorName': 'ServiceCostMonitor', 'MonitorType': 'DIMENSIONAL', 'MonitorDimension': 'SERVICE' } ) # Create anomaly subscription subscription = ce.create_anomaly_subscription( AnomalySubscription={ 'SubscriptionName': 'CostAnomalyAlerts', 'Threshold': 100.0, # Alert on anomalies > $100 'Frequency': 'DAILY', 'MonitorArnList': [monitor['MonitorArn']], 'Subscribers': [ { 'Type': 'EMAIL', 'Address': 'team@company.com' }, { 'Type': 'SNS', 'Address': 'arn:aws:sns:us-east-1:123456789012:cost-alerts' } ] } ) return [monitor, subscription] def create_cost_dashboard(self): """Create executive cost dashboard""" return ''' <!DOCTYPE html> <html> <head> <title>Cloud Cost Dashboard</title> <script src="https://d3js.org/d3.v7.min.js"></script> <style> .metric-card { background: #f5f5f5; padding: 20px; margin: 10px; border-radius: 8px; box-shadow: 0 2px 4px rgba(0,0,0,0.1); } .alert { color: #d32f2f; } .warning { color: #f57c00; } .success { color: #388e3c; } </style> </head> <body> <div id="dashboard"> <h1>Cloud Cost Optimization Dashboard</h1> <div class="summary-row"> <div class="metric-card"> <h3>Current Month Spend</h3> <div class="metric">${current_spend}</div> <div class="trend ${spend_trend_class}">${spend_trend}% vs last month</div> </div> <div class="metric-card"> <h3>Projected Month End</h3> <div class="metric">${projected_spend}</div> <div class="budget-status">Budget: ${budget}</div> </div> <div class="metric-card"> <h3>Optimization Opportunities</h3> <div class="metric">${total_savings_identified}</div> <div class="count">{opportunity_count} recommendations</div> </div> <div class="metric-card"> <h3>Realized Savings</h3> <div class="metric">${realized_savings_mtd}</div> <div class="count">YTD: ${realized_savings_ytd}</div> </div> </div> <div class="charts-row"> <div id="spend-trend-chart"></div> <div id="service-breakdown-chart"></div> <div id="optimization-progress-chart"></div> </div> <div class="recommendations-section"> <h2>Top Optimization Recommendations</h2> <table id="recommendations-table"> <thead> <tr> <th>Priority</th> <th>Service</th> <th>Recommendation</th> <th>Monthly Savings</th> <th>Effort</th> <th>Action</th> </tr> </thead> <tbody> ${recommendation_rows} </tbody> </table> </div> </div> <script> // Real-time updates setInterval(updateDashboard, 60000); // Initialize charts initializeCharts(); </script> </body> </html> '''
Output Format
- Cost Analysis Report: Comprehensive breakdown of current cloud costs
- Optimization Recommendations: Prioritized list of cost-saving opportunities
- Implementation Scripts: Automated scripts for implementing optimizations
- Monitoring Dashboards: Real-time cost tracking and alerting
- ROI Calculations: Detailed savings projections and payback periods
- Risk Assessment: Analysis of risks associated with each optimization
- Implementation Roadmap: Phased approach to cost optimization
- Best Practices Guide: Long-term cost management strategies
Focus on delivering immediate cost savings while establishing sustainable cost optimization practices that maintain performance and reliability standards.