cost-optimization-analysisCloud cost optimization analysis including AWS Cost Explorer, Azure Cost Management, and GCP Billing with right-sizing recommendations and optimization strategies for multi-cloud environments
Independent GitHub snapshot · Automated classification · Original documentation governs
name: cost-optimization-analysis compatibility: opencode completeness: 95 content-types:
Implements comprehensive multi-cloud cost optimization strategies including AWS Cost Explorer analysis, Azure Cost Management analysis, GCP Billing analysis, right-sizing recommendations, spot instance optimization, reserved instance planning, and cost allocation frameworks. This skill enables data-driven decisions to reduce cloud infrastructure costs while maintaining performance and reliability.
Use this skill when:
Avoid this skill for:
cncf-infrastructure-as-code for designing new infrastructure; cost optimization should follow design completioncncf-security-compliance skillData Collection — Gather cost data from AWS Cost Explorer, Azure Cost Management, and GCP Billing APIs. Checkpoint: Verify all cost data is normalized to a common currency (USD) and time period.
Resource Discovery — Identify all resources (EC2, Azure VMs, GCP Compute Engine) with usage and cost data. Checkpoint: Ensure resource IDs are standardized across providers for multi-cloud correlation.
Idle Resource Analysis — Analyze CPU, memory, network, and storage utilization metrics to identify underutilized resources. Checkpoint: Confirm utilization data covers at least 14 days to account for weekly patterns.
Right-Sizing Recommendations — Generate right-sizing recommendations based on utilization thresholds (e.g., CPU < 30% for 2 weeks). Checkpoint: Validate recommendations against application SLAs and performance requirements.
Spot Instance Strategy — Evaluate workload characteristics for spot instance eligibility (fault-tolerant, stateless, interruptible workloads). Checkpoint: Confirm spot instance savings projections account for interruption rates and replacement costs.
Cost Allocation Framework — Implement tagging strategy and cost allocation models for chargeback/showback. Checkpoint: Ensure all resources have required cost allocation tags (owner, project, environment, cost-center).
AWS Cost Explorer provides detailed cost and usage data. Use the AWS CLI to analyze costs by service, tag, and time period.
BAD — Hardcoded date range without error handling
# ❌ BAD — No error checking, hardcoded dates, no output formatting
aws ce get-cost-and-usage \
--time-period Start=2024-01-01,End=2024-02-01 \
--granularity DAILY \
--metrics UnblendedCost \
--group-by Type=DIMENSION Key=SERVICE
GOOD — Robust script with error handling and formatting
#!/bin/bash
# ✅ GOOD — Proper error handling, dynamic dates, JSON output
set -euo pipefail
START_DATE="${START_DATE:-$(date -d '1 month ago' +%Y-%m-%d)}"
END_DATE="${END_DATE:-$(date +%Y-%m-%d)}"
aws ce get-cost-and-usage \
--time-period "Start=${START_DATE},End=${END_DATE}" \
--granularity MONTHLY \
--metrics UnblendedCost \
--group-by "Type=DIMENSION,Key=SERVICE" \
--query 'ResultsByTime[].Total[].UnblendedCost Amount' \
--output text | awk '{sum += $1} END {printf "Total Monthly Cost: $%.2f\n", sum}'
Python SDK Alternative
# ✅ GOOD — Python SDK with proper pagination
import boto3
from datetime import datetime, timedelta
def get_aws_cost_by_service(days: int = 30) -> dict:
"""Get AWS cost breakdown by service for the specified number of days."""
client = boto3.client('ce', region_name='us-east-1')
end_date = datetime.utcnow().date()
start_date = end_date - timedelta(days=days)
response = client.get_cost_and_usage(
TimePeriod={
'Start': start_date.isoformat(),
'End': end_date.isoformat()
},
Granularity='MONTHLY',
Metrics=['UnblendedCost'],
GroupBy=[{'Type': 'DIMENSION', 'Key': 'SERVICE'}]
)
return {
'service_costs': {
group['Keys'][0]: float(result['Total']['UnblendedCost']['Amount'])
for result in response['ResultsByTime'][-1]['Groups']
for group in result['Groups']
},
'total_cost': float(response['ResultsByTime'][-1]['Total']['UnblendedCost']['Amount'])
}
# Usage example
costs = get_aws_cost_by_service(days=30)
for service, cost in costs['service_costs'].items():
print(f"{service}: ${cost:.2f}")
print(f"Total: ${costs['total_cost']:.2f}")
Azure Cost Management provides detailed cost data through the Azure CLI and REST API.
BAD — Direct API call without authentication
# ❌ BAD — Missing authentication, no pagination handling
curl https://management.azure.com/subscriptions/{subscriptionId}/providers/Microsoft.CostManagement/query?api-version=2022-10-01
GOOD — Azure CLI with proper authentication and formatting
#!/bin/bash
# ✅ GOOD — Azure CLI with authentication, filtering, and formatting
set -euo pipefail
# Login if not already authenticated
az account show >/dev/null 2>&1 || az login
SUBSCRIPTION_ID=$(az account show --query id -o tsv)
START_DATE="2024-01-01"
END_DATE="2024-12-31"
# Query cost data by resource group
az costmanagement query \
--type "Usage" \
--timeframe "MonthToDate" \
--dataset-aggregation '{"totalCost":{"name":"PreTaxCost","function":"Sum"}}' \
--dataset-grouping '{"name":"ResourceGroup","type":"Dimension"}' \
--subscription "$SUBSCRIPTION_ID" \
--query "properties.rows[]" \
--output tsv
ARM Template Cost Analysis
# ✅ GOOD — Estimate costs from ARM template deployments
az deployment sub what-if \
--template-file "arm-template.json" \
--parameters "parameters.json" \
--location "eastus" \
--query "properties.cost" \
--output table
Python SDK for Azure Cost Analysis
# ✅ GOOD — Azure Cost Management SDK with proper error handling
from azure.identity import DefaultAzureCredential
from azure.mgmt.costmanagement import CostManagementClient
from datetime import datetime, timedelta
def get_azure_cost_by_resource_group(days: int = 30) -> dict:
"""Get Azure cost breakdown by resource group."""
subscription_id = os.getenv("AZURE_SUBSCRIPTION_ID")
if not subscription_id:
raise ValueError("AZURE_SUBSCRIPTION_ID environment variable not set")
credential = DefaultAzureCredential()
client = CostManagementClient(credential, subscription_id)
end_date = datetime.utcnow().date()
start_date = end_date - timedelta(days=days)
body = {
"type": "Usage",
"timeframe": "Custom",
"time_period": {
"from": start_date.isoformat(),
"to": end_date.isoformat()
},
"dataset": {
"granularity": "Daily",
"grouping": [{"type": "Dimension", "name": "ResourceGroup"}]
}
}
try:
result = client.query.usage(scope=f"subscriptions/{subscription_id}", parameters=body)
return {
"resource_group_costs": {
item.name: item.properties.total_cost
for item in result.properties.rows
},
"total_cost": sum(item.properties.total_cost for item in result.properties.rows)
}
except Exception as e:
print(f"Error fetching Azure costs: {e}")
return {}
# Usage
costs = get_azure_cost_by_resource_group(days=30)
print(f"Total Azure Cost: ${costs.get('total_cost', 0):.2f}")
Google Cloud Billing provides cost data through the Cloud Billing API and gcloud CLI.
BAD — Simple API call without error handling
# ❌ BAD — No error handling, missing authentication
curl -H "Authorization: Bearer $(gcloud auth print-access-token)" \
"https://billingaccount.googleapis.com/v1/billingAccounts"
GOOD — GCP CLI with proper authentication and cost analysis
#!/bin/bash
# ✅ GOOD — GCP CLI with authentication and cost analysis
set -euo pipefail
gcloud auth print-access-token >/dev/null 2>&1 || gcloud auth login
BILLING_ACCOUNT=$(gcloud billing accounts list --format="value(name)" | head -1)
# Get costs by project for the last 30 days
gcloud billing budgets describe \
--billing-account="$BILLING_ACCOUNT" \
--format="value(name)" 2>/dev/null || true
# Alternative: Use Cloud Billing Export to BigQuery for detailed analysis
gcloud bigquery queries run \
"SELECT
project.name as project_name,
service.description as service,
SUM(cost) as total_cost
FROM
\`\${BILLING_ACCOUNT}.billing_data.gcp_billing_export_v1_012345\`
WHERE
_PARTITIONTIME >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 30 DAY)
GROUP BY
project.name, service.description
ORDER BY
total_cost DESC
LIMIT 20" \
--format="table"
Python SDK for GCP Cost Analysis
# ✅ GOOD — GCP Billing SDK with proper error handling and pagination
from google.cloud import billing_v1
from google.api_core.exceptions import GoogleAPICallError
from datetime import datetime, timedelta
import os
def get_gcp_cost_by_project(days: int = 30) -> dict:
"""Get GCP cost breakdown by project."""
billing_account_id = os.getenv("GCP_BILLING_ACCOUNT_ID")
if not billing_account_id:
raise ValueError("GCP_BILLING_ACCOUNT_ID environment variable not set")
client = billing_v1.CloudBillingClient()
# Get billing account name
billing_account_name = f"billingAccounts/{billing_account_id}"
# Get cost data from BigQuery export (recommended approach)
# This assumes Cloud Billing Export is configured to BigQuery
bigquery_client = bigquery.Client()
query = f"""
SELECT
project.name as project_name,
service.description as service,
SUM(cost) as total_cost,
COUNT(DISTINCT sku.id) as sku_count
FROM
`{billing_account_id}.billing_data.gcp_billing_export_v1_012345`
WHERE
_PARTITIONTIME >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL {days} DAY)
GROUP BY
project.name, service.description
ORDER BY
total_cost DESC
"""
try:
query_job = bigquery_client.query(query)
results = query_job.result()
project_costs = {}
for row in results:
if row.project_name not in project_costs:
project_costs[row.project_name] = 0.0
project_costs[row.project_name] += row.total_cost
return {
"project_costs": project_costs,
"total_cost": sum(project_costs.values()),
"services": list(set(row.service for row in results))
}
except GoogleAPICallError as e:
print(f"GCP API Error: {e}")
return {}
except Exception as e:
print(f"Error fetching GCP costs: {e}")
return {}
# Usage
os.environ["GCP_BILLING_ACCOUNT_ID"] = "012345-567890-ABCDEF"
costs = get_gcp_cost_by_project(days=30)
print(f"Total GCP Cost: ${costs.get('total_cost', 0):.2f}")
Right-sizing analysis identifies resources that are over-provisioned or under-provisioned.
BAD — Simple threshold without context
# ❌ BAD — No context, no history, no action plan
for instance in $(aws ec2 describe-instances --query 'Reservations[].Instances[].InstanceId' --output text); do
cpu=$(aws cloudwatch get-metric-statistics --metric-name CPUUtilization --dimensions InstanceId=$instance --start-time 2024-01-01 --end-time 2024-01-02 --period 86400 --statistics Average --query 'Datapoints[].Average' --output text)
echo "$instance CPU: $cpu"
done
GOOD — Comprehensive right-sizing analysis script
#!/bin/bash
# ✅ GOOD — Comprehensive right-sizing with historical analysis and recommendations
set -euo pipefail
REGION="${AWS_REGION:-us-east-1}"
DAYS_ANALYZED="${DAYS_ANALYZED:-14}"
CPU_THRESHOLD_LOW=30
CPU_THRESHOLD_HIGH=80
echo "=== Right-Sizing Analysis ==="
echo "Region: $REGION"
echo "Analysis Period: $DAYS_ANALYZED days"
echo "Low CPU Threshold: ${CPU_THRESHOLD_LOW}%"
echo "High CPU Threshold: ${CPU_THRESHOLD_HIGH}%"
echo ""
# Get all running instances
instances=$(aws ec2 describe-instances \
--filters "Name=instance-state-name,Values=running" \
--query 'Reservations[].Instances[].{InstanceId:InstanceId,InstanceType:InstanceType,Tags:Tags}' \
--region "$REGION" \
--output json)
# Analyze each instance
echo "$instances" | jq -c '.Reservations[].Instances[]' | while read -r instance; do
instance_id=$(echo "$instance" | jq -r '.InstanceId')
current_type=$(echo "$instance" | jq -r '.InstanceType')
# Get CPU utilization data
cpu_avg=$(aws cloudwatch get-metric-statistics \
--metric-name CPUUtilization \
--dimensions Name=InstanceId,Value=$instance_id \
--start-time $(date -d "$DAYS_ANALYZED days ago" +%Y-%m-%dT%H:%M:%S) \
--end-time $(date +%Y-%m-%dT%H:%M:%S) \
--period 86400 \
--statistics Average \
--output json | jq -r '.Datapoints[].Average' | awk '{sum+=$1} END {printf "%.1f", sum/NR}')
# Get memory utilization (requires CloudWatch Agent)
memory_avg=$(aws cloudwatch get-metric-statistics \
--metric-name MemoryUtilization \
--namespace AWS/ECS \
--dimensions Name=ClusterName,Value=default Name=ServiceName,Value=none \
--start-time $(date -d "$DAYS_ANALYZED days ago" +%Y-%m-%dT%H:%M:%S) \
--end-time $(date +%Y-%m-%dT%H:%M:%S) \
--period 86400 \
--statistics Average \
--output json 2>/dev/null | jq -r '.Datapoints[].Average' | awk '{sum+=$1} END {printf "%.1f", sum/NR}' || echo "N/A")
# Generate recommendation
if [[ "$cpu_avg" != "N/A" ]]; then
if (( $(echo "$cpu_avg < $CPU_THRESHOLD_LOW" | bc -l) )); then
recommendation="DOWNSIZE"
suggested_type="t3.medium" # Example suggestion
elif (( $(echo "$cpu_avg > $CPU_THRESHOLD_HIGH" | bc -l) )); then
recommendation="UPSIZE"
suggested_type="m5.large" # Example suggestion
else
recommendation="OPTIMAL"
suggested_type="$current_type"
fi
else
recommendation="INSUFFICIENT_DATA"
suggested_type="$current_type"
fi
echo "$instance_id | $current_type | CPU: ${cpu_avg}% | Mem: ${memory_avg}% | Recommendation: $recommendation (→ $suggested_type)"
done
Python Right-Sizing Engine
# ✅ GOOD — Python right-sizing engine with historical analysis
import boto3
from datetime import datetime, timedelta
from typing import Dict, List, Tuple
import statistics
class RightSizingAnalyzer:
"""Analyzes EC2 instances for right-sizing opportunities."""
def __init__(self, region: str = "us-east-1"):
self.ec2 = boto3.client('ec2', region_name=region)
self.cloudwatch = boto3.client('cloudwatch', region_name=region)
self.ssm = boto3.client('ssm', region_name=region)
# Instance type hierarchy for right-sizing decisions
self.instance_hierarchy = {
't2': ['t3', 't4g'],
't3': ['t2', 't4g'],
'm5': ['m4', 'm6g'],
'c5': ['c4', 'c6g'],
'r5': ['r4', 'r6g'],
}
self.thresholds = {
'cpu_low': 30,
'cpu_high': 80,
'memory_low': 40,
'memory_high': 85,
'days': 14,
}
def get_instance_metrics(self, instance_id: str, days: int = 14) -> Dict[str, float]:
"""Get CPU and memory metrics for an instance."""
end_time = datetime.utcnow()
start_time = end_time - timedelta(days=days)
# Get CPU utilization
cpu_stats = self.cloudwatch.get_metric_statistics(
Namespace='AWS/EC2',
MetricName='CPUUtilization',
Dimensions=[{'Name': 'InstanceId', 'Value': instance_id}],
StartTime=start_time,
EndTime=end_time,
Period=86400,
Statistics=['Average'],
Unit='Percent'
)
cpu_values = [dp['Average'] for dp in cpu_stats.get('Datapoints', [])]
cpu_avg = statistics.mean(cpu_values) if cpu_values else None
# Get memory utilization (requires CloudWatch Agent)
try:
memory_stats = self.cloudwatch.get_metric_statistics(
Namespace='CWAgent',
MetricName='memory_utilization',
Dimensions=[
{'Name': 'InstanceId', 'Value': instance_id},
{'Name': 'InstanceType', 'Value': 'ec2'}
],
StartTime=start_time,
EndTime=end_time,
Period=86400,
Statistics=['Average'],
Unit='Percent'
)
memory_values = [dp['Average'] for dp in memory_stats.get('Datapoints', [])]
memory_avg = statistics.mean(memory_values) if memory_values else None
except Exception:
memory_avg = None
return {
'cpu_avg': round(cpu_avg, 2) if cpu_avg else None,
'memory_avg': round(memory_avg, 2) if memory_avg else None,
'cpu_data_points': len(cpu_values),
'memory_data_points': len(memory_values) if memory_avg else 0,
}
def suggest_right_size(self, instance_type: str, metrics: Dict[str, float]) -> Tuple[str, str]:
"""Suggest a right-sized instance type based on metrics."""
cpu = metrics.get('cpu_avg')
memory = metrics.get('memory_avg')
# If insufficient data, return original type
if not cpu and not memory:
return instance_type, "INSUFFICIENT_DATA"
# If high CPU, consider up-sizing
if cpu and cpu > self.thresholds['cpu_high']:
return self._upsize(instance_type), "HIGH_CPU"
# If low CPU, consider down-sizing
if cpu and cpu < self.thresholds['cpu_low']:
return self._downsize(instance_type), "LOW_CPU"
# If high memory, consider memory-optimized
if memory and memory > self.thresholds['memory_high']:
return self._memory_optimize(instance_type), "HIGH_MEMORY"
# If low memory, consider general-purpose
if memory and memory < self.thresholds['memory_low']:
return self._general_purpose(instance_type), "LOW_MEMORY"
return instance_type, "OPTIMAL"
def _downsize(self, instance_type: str) -> str:
"""Downsize instance type if possible."""
base_type = next((k for k in self.instance_hierarchy.keys() if instance_type.startswith(k)), None)
if base_type:
smaller_types = self.instance_hierarchy[base_type]
if smaller_types:
return f"{smaller_types[0]}.large"
return instance_type
def _upsize(self, instance_type: str) -> str:
"""Upsize instance type if needed."""
base_type = next((k for k in self.instance_hierarchy.keys() if instance_type.startswith(k)), None)
if base_type:
larger_types = [t for t in self.instance_hierarchy.get(base_type, []) if t != instance_type]
if larger_types:
return f"{larger_types[0]}.xlarge"
return instance_type
def _memory_optimize(self, instance_type: str) -> str:
"""Suggest memory-optimized instance type."""
if instance_type.startswith('t'):
return 'r5.large'
elif instance_type.startswith('m'):
return 'r5.xlarge'
return f"r{instance_type[1:]}" if instance_type[1] != 'r' else instance_type
def _general_purpose(self, instance_type: str) -> str:
"""Suggest general-purpose instance type."""
if instance_type.startswith('r'):
return 'm5.large'
elif instance_type.startswith('c'):
return 't3.large'
return f"m{instance_type[1:]}" if instance_type[1] != 'm' else instance_type
def analyze_instance(self, instance_id: str) -> Dict:
"""Analyze a single instance for right-sizing."""
# Get instance details
response = self.ec2.describe_instances(InstanceIds=[instance_id])
instance = response['Reservations'][0]['Instances'][0]
instance_type = instance['InstanceType']
# Get metrics
metrics = self.get_instance_metrics(instance_id, self.thresholds['days'])
# Suggest right size
suggested_type, reason = self.suggest_right_size(instance_type, metrics)
return {
'instance_id': instance_id,
'current_type': instance_type,
'suggested_type': suggested_type,
'reason': reason,
'metrics': metrics,
'potential_savings': self._estimate_savings(instance_type, suggested_type),
}
def _estimate_savings(self, current_type: str, suggested_type: str) -> float:
"""Estimate potential monthly savings from right-sizing."""
# Simplified pricing model (replace with actual pricing data)
pricing = {
't3.large': 0.0832,
't3.xlarge': 0.1664,
't3.2xlarge': 0.3328,
'm5.large': 0.096,
'm5.xlarge': 0.192,
'm5.2xlarge': 0.384,
'r5.large': 0.126,
'r5.xlarge': 0.252,
'r5.2xlarge': 0.504,
}
current_price = pricing.get(current_type, 0.1)
suggested_price = pricing.get(suggested_type, 0.1)
monthly_usage_hours = 730 # Average hours per month
monthly_savings = (current_price - suggested_price) * monthly_usage_hours
return round(max(0, monthly_savings), 2)
def analyze_all_instances(self) -> List[Dict]:
"""Analyze all instances in the region."""
instances = self.ec2.describe_instances(
Filters=[{'Name': 'instance-state-name', 'Values': ['running']}]
)
results = []
for reservation in instances['Reservations']:
for instance in reservation['Instances']:
result = self.analyze_instance(instance['InstanceId'])
results.append(result)
return results
# Usage example
if __name__ == "__main__":
analyzer = RightSizingAnalyzer(region="us-east-1")
results = analyzer.analyze_all_instances()
print("Right-Sizing Analysis Results:")
print("-" * 80)
total_monthly_savings = 0
for result in results:
if result['reason'] != 'OPTIMAL':
print(f"Instance: {result['instance_id']}")
print(f" Current: {result['current_type']}")
print(f" Suggested: {result['suggested_type']} ({result['reason']})")
print(f" Monthly Savings: ${result['potential_savings']}")
print()
total_monthly_savings += result['potential_savings']
print(f"Total Potential Monthly Savings: ${total_monthly_savings:.2f}")
Spot instance optimization reduces compute costs by up to 90% for fault-tolerant workloads.
BAD — Simple spot request without termination handling
# ❌ BAD — No termination handling, no replacement strategy
aws ec2 request-spot-instances \
--spot-price "0.05" \
--instance-count 1 \
--type "one-time" \
--launch-specification file://launch-spec.json
GOOD — Comprehensive spot instance management script
#!/bin/bash
# ✅ GOOD — Spot instance with termination handling, replacement strategy, and scaling
set -euo pipefail
SPOT_PRICE="${SPOT_PRICE:-0.05}"
INSTANCE_TYPE="${INSTANCE_TYPE:-c5.large}"
REGION="${AWS_REGION:-us-east-1}"
MIN_INSTANCES="${MIN_INSTANCES:-2}"
MAX_INSTANCES="${MAX_INSTANCES:-10}"
# Spot request with replacement strategy
create_spot_fleet() {
local fleet_config=$(cat <<EOF
{
"IamFleetRole": "arn:aws:iam::123456789012:role/spot-fleet-role",
"TargetCapacity": $MIN_INSTANCES,
"SpotPrice": "$SPOT_PRICE",
"AllocationStrategy": "lowestPrice",
"LaunchSpecifications": [
{
"ImageId": "ami-0123456789abcdef0",
"InstanceType": "$INSTANCE_TYPE",
"KeyName": "my-key",
"SubnetId": "subnet-0123456789abcdef0",
"IamInstanceProfile": {
"Arn": "arn:aws:iam::123456789012:instance-profile/spot-instance-profile"
},
"SpotPrice": "$SPOT_PRICE"
}
],
"TerminateInstancesWithExpiration": true,
"ReplaceUnhealthyInstances": true
}
EOF
)
aws ec2 request-spot-fleet \
--spot-fleet-request-data "$fleet_config" \
--region "$REGION"
}
# Check spot instance health
check_spot_health() {
local fleet_id="$1"
# Get spot instance requests
aws ec2 describe-spot-instance-requests \
--filters "Name=spot-fleet-request-id,Values=$fleet_id" \
--query 'SpotInstanceRequests[].{InstanceId:InstanceId,State:State,Status:Status}' \
--output table
# Check for terminations
aws ec2 describe-spot-instance-requests \
--filters "Name=spot-fleet-request-id,Values=$fleet_id" \
--query 'SpotInstanceRequests[?State==`closed`]' \
--output text
}
# Scale spot fleet based on demand
scale_spot_fleet() {
local fleet_id="$1"
local new_capacity="$2"
aws ec2 modify-spot-fleet-request \
--spot-fleet-request-id "$fleet_id" \
--target-capacity "$new_capacity" \
--terminate-instances-with-expiration
}
# Main execution
case "${1:-create}" in
create)
echo "Creating spot fleet with $MIN_INSTANCES instances at $SPOT_PRICE"
create_spot_fleet
;;
check)
check_spot_health "$2"
;;
scale)
scale_spot_fleet "$2" "$3"
;;
*)
echo "Usage: $0 {create|check|scale}"
exit 1
;;
esac
Python Spot Instance Optimizer
# ✅ GOOD — Python spot instance optimizer with termination handling
import boto3
from datetime import datetime, timedelta
from typing import Dict, List, Optional
import statistics
import time
class SpotInstanceOptimizer:
"""Optimizes spot instance usage for cost savings."""
def __init__(self, region: str = "us-east-1"):
self.ec2 = boto3.client('ec2', region_name=region)
self.cloudwatch = boto3.client('cloudwatch', region_name=region)
self.spot_pricing_history = {}
self.recommendation_cache = {}
def get_spot_price_history(self, instance_types: List[str], days: int = 7) -> Dict[str, List[float]]:
"""Get historical spot prices for given instance types."""
end_time = datetime.utcnow()
start_time = end_time - timedelta(days=days)
prices = {}
for instance_type in instance_types:
try:
response = self.ec2.describe_spot_price_history(
InstanceTypes=[instance_type],
StartTime=start_time,
EndTime=end_time,
MaxResults=1000
)
price_list = [float(dp['SpotPrice']) for dp in response['SpotPriceHistory']]
prices[instance_type] = price_list
# Cache the data
self.spot_pricing_history[instance_type] = {
'prices': price_list,
'avg': statistics.mean(price_list) if price_list else 0,
'min': min(price_list) if price_list else 0,
'max': max(price_list) if price_list else 0,
'std': statistics.stdev(price_list) if len(price_list) > 1 else 0,
}
except Exception as e:
print(f"Error fetching spot price history for {instance_type}: {e}")
return prices
def recommend_spot_instance(self, instance_type: str, current_on_demand_price: float) -> Dict:
"""Recommend whether to use spot for given instance type."""
history = self.spot_pricing_history.get(instance_type, {})
if not history:
return {
'instance_type': instance_type,
'recommendation': 'UNKNOWN',
'reason': 'Insufficient historical data',
}
avg_price = history['avg']
min_price = history['min']
max_price = history['max']
std = history['std']
# Calculate savings
savings_percentage = ((current_on_demand_price - avg_price) / current_on_demand_price) * 100
# Determine recommendation
if savings_percentage > 50 and std < 0.05:
recommendation = 'STRONG_RECOMMENDATION'
reason = f'High savings ({savings_percentage:.1f}%) with low volatility'
elif savings_percentage > 30:
recommendation = 'RECOMMENDATION'
reason = f'Moderate savings ({savings_percentage:.1f}%)'
elif savings_percentage > 10:
recommendation = 'CONSIDER'
reason = f'Some savings ({savings_percentage:.1f}%), monitor volatility'
else:
recommendation = 'NOT_RECOMMENDED'
reason = f'Low savings ({savings_percentage:.1f}%)'
return {
'instance_type': instance_type,
'recommendation': recommendation,
'reason': reason,
'spot_avg_price': round(avg_price, 4),
'spot_min_price': round(min_price, 4),
'spot_max_price': round(max_price, 4),
'on_demand_price': current_on_demand_price,
'savings_percentage': round(savings_percentage, 1),
'price_volatility': round(std, 4),
}
def check_spot_capacity(self, instance_types: List[str], availability_zone: str) -> Dict[str, bool]:
"""Check spot capacity availability."""
capacity = {}
for instance_type in instance_types:
try:
response = self.ec2.describe_spot_instance_requests(
InstanceTypes=[instance_type],
AvailabilityZone=availability_zone,
MaxResults=1
)
capacity[instance_type] = True
except Exception:
capacity[instance_type] = False
return capacity
def estimate_savings(self, instance_type: str, hours_per_month: int = 730) -> Dict:
"""Estimate monthly savings from spot instances."""
history = self.spot_pricing_history.get(instance_type, {})
if not history:
return {'error': 'No historical data available'}
avg_spot_price = history['avg']
# Get on-demand price (simplified - use actual pricing API in production)
on_demand_prices = {
'c5.large': 0.17,
'c5.xlarge': 0.34,
'c5.2xlarge': 0.68,
'm5.large': 0.096,
'm5.xlarge': 0.192,
'r5.large': 0.126,
'r5.xlarge': 0.252,
}
on_demand_price = on_demand_prices.get(instance_type, avg_spot_price * 2)
on_demand_cost = on_demand_price * hours_per_month
spot_cost = avg_spot_price * hours_per_month
savings = on_demand_cost - spot_cost
savings_percentage = (savings / on_demand_cost) * 100
return {
'instance_type': instance_type,
'on_demand_cost': round(on_demand_cost, 2),
'spot_cost': round(spot_cost, 2),
'monthly_savings': round(savings, 2),
'savings_percentage': round(savings_percentage, 1),
'break_even_price': round(spot_cost / hours_per_month, 4),
}
def get_all_recommendations(self) -> List[Dict]:
"""Get recommendations for all common instance types."""
instance_types = ['c5.large', 'c5.xlarge', 'c5.2xlarge', 'm5.large', 'm5.xlarge', 'r5.large', 'r5.xlarge']
on_demand_prices = {
'c5.large': 0.17, 'c5.xlarge': 0.34, 'c5.2xlarge': 0.68,
'm5.large': 0.096, 'm5.xlarge': 0.192,
'r5.large': 0.126, 'r5.xlarge': 0.252,
}
self.get_spot_price_history(instance_types)
recommendations = []
for instance_type in instance_types:
recommendation = self.recommend_spot_instance(
instance_type,
on_demand_prices.get(instance_type, 0)
)
savings = self.estimate_savings(instance_type)
recommendation.update(savings)
recommendations.append(recommendation)
return recommendations
# Usage example
if __name__ == "__main__":
optimizer = SpotInstanceOptimizer(region="us-east-1")
recommendations = optimizer.get_all_recommendations()
print("Spot Instance Optimization Recommendations")
print("=" * 80)
total_monthly_savings = 0
for rec in recommendations:
if rec['recommendation'] in ['STRONG_RECOMMENDATION', 'RECOMMENDATION']:
print(f"\nInstance: {rec['instance_type']}")
print(f" Recommendation: {rec['recommendation']}")
print(f" Reason: {rec['reason']}")
print(f" On-demand: ${rec['on_demand_price']}/hr")
print(f" Spot avg: ${rec['spot_avg_price']}/hr")
print(f" Monthly Savings: ${rec['monthly_savings']}")
total_monthly_savings += rec['monthly_savings']
print(f"\n" + "=" * 80)
print(f"Total Potential Monthly Savings: ${total_monthly_savings:.2f}")
Reserved instance optimization analyzes usage patterns to recommend RI purchases.
BAD — Manual analysis without historical data
# ❌ BAD — No historical analysis, no ROI calculation
# Just buying RIs without data-driven decisions
aws ec2 purchase-reserved-instances-offering \
--reserved-instances-offering-id "ri-offering-123" \
--instance-count 5
GOOD — Data-driven RI optimization script
#!/bin/bash
# ✅ GOOD — Historical analysis with ROI calculation
set -euo pipefail
REGION="${AWS_REGION:-us-east-1}"
ANALYSIS_DAYS="${ANALYSIS_DAYS:-90}"
# Get historical instance usage
get_instance_usage() {
local instance_type="$1"
# Get average CPU utilization over analysis period
aws cloudwatch get-metric-statistics \
--metric-name CPUUtilization \
--dimensions Name=InstanceId,Value=unknown \
--namespace AWS/EC2 \
--start-time $(date -d "$ANALYSIS_DAYS days ago" +%Y-%m-%dT%H:%M:%S) \
--end-time $(date +%Y-%m-%dT%H:%M:%S) \
--period 86400 \
--statistics Average \
--query 'Datapoints[].Average' \
--output text | awk '{sum+=$1; count++} END {printf "%.1f", sum/count}'
}
# Calculate RI ROI
calculate_ri_roi() {
local on_demand_hourly="$1"
local ri_upfront="$2"
local ri_annual="$3"
local usage_hours="$4"
# Calculate total cost with on-demand
local on_demand_total=$(echo "$on_demand_hourly * $usage_hours" | bc -l)
# Calculate total cost with RI (upfront + annual)
local ri_total=$(echo "$ri_upfront + $ri_annual" | bc -l)
# Calculate savings
local savings=$(echo "$on_demand_total - $ri_total" | bc -l)
local savings_percentage=$(echo "($savings / $on_demand_total) * 100" | bc -l)
echo "On-Demand Total: \$on_demand_total"
echo "RI Total: \$ri_total"
echo "Savings: \$savings ($savings_percentage%)"
}
# Analyze instance usage and recommend RIs
analyze_ri_opportunity() {
local instance_type="$1"
local current_count="$2"
echo "=== Reserved Instance Analysis ==="
echo "Instance Type: $instance_type"
echo "Current Count: $current_count"
echo "Analysis Period: $ANALYSIS_DAYS days"
echo ""
# Get on-demand pricing
local on_demand_price=$(aws ec2 describe-spot-price-history \
--instance-types "$instance_type" \
--product-description "Linux/UNIX" \
--query 'SpotPriceHistory[0].SpotPrice' \
--output text)
# Get RI pricing (simplified - use actual RI pricing API)
local ri_pricing={
"c5.large": {"upfront": 450, "annual": 600, "hourly": 0.17},
"c5.xlarge": {"upfront": 900, "annual": 1200, "hourly": 0.34},
"m5.large": {"upfront": 300, "annual": 400, "hourly": 0.096},
"m5.xlarge": {"upfront": 600, "annual": 800, "hourly": 0.192},
}
if [[ -v ri_pricing["$instance_type"] ]]; then
local upfront="${ri_pricing[$instance_type, upfront]}"
local annual="${ri_pricing[$instance_type, annual]}"
local hourly="${ri_pricing[$instance_type, hourly]}"
# Assume 24/7 usage
local usage_hours=$((ANALYSIS_DAYS * 24))
echo "Pricing:"
echo " On-Demand: \${hourly}/hour"
echo " RI Upfront: \${upfront}"
echo " RI Annual: \${annual}"
echo ""
echo "Projected Usage: $usage_hours hours"
echo ""
calculate_ri_roi "$hourly" "$upfront" "$annual" "$usage_hours"
else
echo "RI pricing not available for $instance_type"
fi
}
# Main execution
instance_type="${1:-c5.large}"
count="${2:-5}"
analyze_ri_opportunity "$instance_type" "$count"
Python RI Optimizer
# ✅ GOOD — Python RI optimizer with ROI calculations
import boto3
from datetime import datetime, timedelta
from typing import Dict, List, Optional
from dataclasses import dataclass
import statistics
@dataclass
class InstanceUsage:
instance_id: str
instance_type: str
avg_cpu: float
hours_used: int
days_used: int
@dataclass
class RIRecommendation:
instance_type: str
recommended_count: int
upfront_cost: float
annual_cost: float
on_demand_cost: float
roi_percentage: float
break_even_months: float
recommendation: str
class ReservedInstanceOptimizer:
"""Optimizes reserved instance purchases for cost savings."""
def __init__(self, region: str = "us-east-1"):
self.ec2 = boto3.client('ec2', region_name=region)
self.cloudwatch = boto3.client('cloudwatch', region_name=region)
self.pricing = boto3.client('pricing', region_name='us-east-1')
# Simplified RI pricing (replace with actual AWS Pricing API in production)
self.ri_pricing = {
'c5.large': {'upfront': 450, 'annual': 600, 'hourly': 0.17},
'c5.xlarge': {'upfront': 900, 'annual': 1200, 'hourly': 0.34},
'c5.2xlarge': {'upfront': 1800, 'annual': 2400, 'hourly': 0.68},
'm5.large': {'upfront': 300, 'annual': 400, 'hourly': 0.096},
'm5.xlarge': {'upfront': 600, 'annual': 800, 'hourly': 0.192},
'm5.2xlarge': {'upfront': 1200, 'annual': 1600, 'hourly': 0.384},
'r5.large': {'upfront': 400, 'annual': 550, 'hourly': 0.126},
'r5.xlarge': {'upfront': 800, 'annual': 1100, 'hourly': 0.252},
}
self.usage_analysis = {}
def get_instance_usage(self, instance_id: str, days: int = 90) -> Optional[InstanceUsage]:
"""Get historical usage data for an instance."""
end_time = datetime.utcnow()
start_time = end_time - timedelta(days=days)
try:
# Get CPU utilization data
response = self.cloudwatch.get_metric_statistics(
Namespace='AWS/EC2',
MetricName='CPUUtilization',
Dimensions=[{'Name': 'InstanceId', 'Value': instance_id}],
StartTime=start_time,
EndTime=end_time,
Period=86400,
Statistics=['Average'],
Unit='Percent'
)
datapoints = response.get('Datapoints', [])
if not datapoints:
return None
avg_cpu = statistics.mean([dp['Average'] for dp in datapoints])
# Calculate hours used (assume running 24/7)
hours_used = days * 24
return InstanceUsage(
instance_id=instance_id,
instance_type='UNKNOWN', # Get this from describe_instances
avg_cpu=round(avg_cpu, 2),
hours_used=hours_used,
days_used=days,
)
except Exception as e:
print(f"Error getting usage for {instance_id}: {e}")
return None
def analyze_all_instances(self, days: int = 90) -> List[InstanceUsage]:
"""Analyze usage for all running instances."""
# Get all running instances
response = self.ec2.describe_instances(
Filters=[{'Name': 'instance-state-name', 'Values': ['running']}]
)
usages = []
for reservation in response['Reservations']:
for instance in reservation['Instances']:
instance_type = instance['InstanceType']
instance_id = instance['InstanceId']
usage = self.get_instance_usage(instance_id, days)
if usage:
usage.instance_type = instance_type
usages.append(usage)
# Add instance type to analysis
if instance_type not in self.usage_analysis:
self.usage_analysis[instance_type] = {
'count': 0,
'total_hours': 0,
'avg_cpu': [],
}
self.usage_analysis[instance_type]['count'] += 1
self.usage_analysis[instance_type]['total_hours'] += usage.hours_used
self.usage_analysis[instance_type]['avg_cpu'].append(usage.avg_cpu)
return usages
def calculate_ri_roi(self, instance_type: str, count: int, days: int = 365) -> RIRecommendation:
"""Calculate ROI for reserved instances."""
pricing = self.ri_pricing.get(instance_type)
if not pricing:
return RIRecommendation(
instance_type=instance_type,
recommended_count=count,
upfront_cost=0,
annual_cost=0,
on_demand_cost=0,
roi_percentage=0,
break_even_months=0,
recommendation='NO_DATA',
)
hourly_price = pricing['hourly']
upfront = pricing['upfront'] * count
annual = pricing['annual'] * count
# Calculate on-demand cost for the period
usage = self.usage_analysis.get(instance_type, {})
total_hours = usage.get('total_hours', days * 24 * count)
on_demand_cost = hourly_price * total_hours
# Calculate savings
ri_total = upfront + annual
savings = on_demand_cost - ri_total
roi_percentage = (savings / on_demand_cost) * 100 if on_demand_cost > 0 else 0
# Calculate break-even point
monthly_ri_cost = annual / 12
monthly_on_demand_cost = hourly_price * (days * 24 * count) / 12
break_even_months = upfront / (monthly_on_demand_cost - monthly_ri_cost) if (monthly_on_demand_cost - monthly_ri_cost) > 0 else float('inf')
# Determine recommendation
if roi_percentage > 30:
recommendation = 'STRONG_RECOMMENDATION'
elif roi_percentage > 15:
recommendation = 'RECOMMENDATION'
elif roi_percentage > 5:
recommendation = 'CONSIDER'
else:
recommendation = 'NOT_RECOMMENDED'
return RIRecommendation(
instance_type=instance_type,
recommended_count=count,
upfront_cost=upfront,
annual_cost=annual,
on_demand_cost=round(on_demand_cost, 2),
roi_percentage=round(roi_percentage, 2),
break_even_months=round(break_even_months, 1),
recommendation=recommendation,
)
def get_all_recommendations(self, days: int = 365) -> List[RIRecommendation]:
"""Get RI recommendations for all instance types."""
self.analyze_all_instances(days=days)
recommendations = []
for instance_type, analysis in self.usage_analysis.items():
count = analysis['count']
recommendation = self.calculate_ri_roi(instance_type, count, days)
recommendations.append(recommendation)
return recommendations
def generate_ri_report(self, days: int = 365) -> str:
"""Generate a comprehensive RI optimization report."""
recommendations = self.get_all_recommendations(days)
report = ["=" * 80, "Reserved Instance Optimization Report", "=" * 80, ""]
total_annual_savings = 0
total_upfront_investment = 0
for rec in recommendations:
if rec.recommendation in ['STRONG_RECOMMENDATION', 'RECOMMENDATION']:
report.append(f"Instance Type: {rec.instance_type}")
report.append(f" Recommended Count: {rec.recommended_count}")
report.append(f" Recommendation: {rec.recommendation}")
report.append(f" Upfront Cost: ${rec.upfront_cost:,.2f}")
report.append(f" Annual Cost: ${rec.annual_cost:,.2f}")
report.append(f" On-Demand Cost: ${rec.on_demand_cost:,.2f}")
report.append(f" ROI: {rec.roi_percentage:.1f}%")
report.append(f" Break-Even: {rec.break_even_months:.1f} months")
report.append("")
total_annual_savings += rec.on_demand_cost - rec.annual_cost - rec.upfront_cost
total_upfront_investment += rec.upfront_cost
report.append("=" * 80)
report.append(f"Total Upfront Investment: ${total_upfront_investment:,.2f}")
report.append(f"Total Annual Savings: ${total_annual_savings:,.2f}")
report.append("=" * 80)
return "\n".join(report)
# Usage example
if __name__ == "__main__":
optimizer = ReservedInstanceOptimizer(region="us-east-1")
report = optimizer.generate_ri_report(days=365)
print(report)
Cost allocation tagging implements chargeback/showback models through proper resource tagging.
BAD — Manual tagging without validation
# ❌ BAD — No validation, inconsistent tags
aws ec2 create-tags \
--resources i-1234567890abcdef0 \
--tags Key=Project,Value=MyProject Key=Environment,Value=prod
GOOD — Automated tagging with validation and enforcement
#!/bin/bash
# ✅ GOOD — Automated tagging with validation and cost allocation
set -euo pipefail
# Cost allocation tags (required for all resources)
REQUIRED_TAGS=(
"Project"
"Environment"
"Owner"
"CostCenter"
"Department"
)
# Validate tags
validate_tags() {
local tags="$1"
for tag in "${REQUIRED_TAGS[@]}"; do
if ! echo "$tags" | grep -q "\"Key\":\"$tag\""; then
echo "ERROR: Missing required tag: $tag"
return 1
fi
done
echo "All required tags present"
return 0
}
# Tag EC2 instance
tag_ec2_instance() {
local instance_id="$1"
local tags="$2"
echo "Tagging EC2 instance: $instance_id"
if validate_tags "$tags"; then
aws ec2 create-tags \
--resources "$instance_id" \
--tags "$tags" \
--region "${AWS_REGION:-us-east-1}"
echo "Tags applied successfully"
else
echo "Tagging failed: validation error"
return 1
fi
}
# Tag S3 bucket
tag_s3_bucket() {
local bucket_name="$1"
local tags="$2"
echo "Tagging S3 bucket: $bucket_name"
if validate_tags "$tags"; then
aws s3api put-bucket-tagging \
--bucket "$bucket_name" \
--tagging "$tags"
echo "Tags applied successfully"
else
echo "Tagging failed: validation error"
return 1
fi
}
# Tag RDS instance
tag_rds_instance() {
local db_instance_identifier="$1"
local tags="$2"
echo "Tagging RDS instance: $db_instance_identifier"
if validate_tags "$tags"; then
aws rds add-tags-to-resource \
--resource-name "arn:aws:rds:${AWS_REGION:-us-east-1}:123456789012:db:$db_instance_identifier" \
--tags "$tags"
echo "Tags applied successfully"
else
echo "Tagging failed: validation error"
return 1
fi
}
# Example usage
INSTANCE_ID="${1:-i-1234567890abcdef0}"
TAGS='[
{"Key":"Project","Value":"WebApplication"},
{"Key":"Environment","Value":"production"},
{"Key":"Owner","Value":"john.doe@company.com"},
{"Key":"CostCenter","Value":"CC12345"},
{"Key":"Department","Value":"Engineering"}
]'
tag_ec2_instance "$INSTANCE_ID" "$TAGS"
Python Tagging Manager
# ✅ GOOD — Python tagging manager with cost allocation enforcement
import boto3
from typing import Dict, List, Optional
from datetime import datetime
import re
class CostAllocationTagManager:
"""Manages cost allocation tagging across AWS resources."""
# Required cost allocation tags
REQUIRED_TAGS = {
'Project': {'type': 'string', 'pattern': r'^[a-zA-Z0-9-_.]{1,50}#x27;},
'Environment': {'type': 'string', 'values': ['dev', 'staging', 'production', 'testing']},
'Owner': {'type': 'email', 'pattern': r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}#x27;},
'CostCenter': {'type': 'string', 'pattern': r'^[A-Z]{2}\d{5}#x27;},
'Department': {'type': 'string', 'values': ['Engineering', 'Marketing', 'Sales', 'Finance', 'Operations']},
'Application': {'type': 'string', 'pattern': r'^[a-zA-Z0-9-_.]{1,50}#x27;},
'BusinessUnit': {'type': 'string', 'pattern': r'^[a-zA-Z0-9-_.]{1,50}#x27;},
}
def __init__(self, region: str = "us-east-1"):
self.ec2 = boto3.client('ec2', region_name=region)
self.s3 = boto3.client('s3', region_name=region)
self.rds = boto3.client('rds', region_name=region)
self.iam = boto3.client('iam', region_name=region)
self.tag_cache: Dict[str, Dict[str, str]] = {}
def validate_tag(self, key: str, value: str) -> tuple:
"""Validate a tag key-value pair."""
if key not in self.REQUIRED_TAGS:
return True, None # Optional tag
tag_spec = self.REQUIRED_TAGS[key]
# Check if value is in allowed values
if 'values' in tag_spec and value not in tag_spec['values']:
return False, f"Value '{value}' not in allowed values: {tag_spec['values']}"
# Check regex pattern
if 'pattern' in tag_spec:
pattern = tag_spec['pattern']
if tag_spec.get('type') == 'email':
pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}#x27;
if not re.match(pattern, value):
return False, f"Value '{value}' does not match pattern: {pattern}"
return True, None
def validate_tags(self, tags: List[Dict[str, str]]) -> tuple:
"""Validate all tags and ensure required tags are present."""
tag_dict = {tag['Key']: tag['Value'] for tag in tags}
# Check required tags
for required_tag in self.REQUIRED_TAGS:
if required_tag not in tag_dict:
return False, f"Missing required tag: {required_tag}"
# Validate each tag
for key, value in tag_dict.items():
valid, error = self.validate_tag(key, value)
if not valid:
return False, f"Invalid tag {key}: {error}"
return True, None
def tag_ec2_instance(self, instance_id: str, tags: List[Dict[str, str]]) -> bool:
"""Tag an EC2 instance with cost allocation tags."""
# Validate tags
valid, error = self.validate_tags(tags)
if not valid:
print(f"Tag validation failed: {error}")
return False
try:
self.ec2.create_tags(
Resources=[instance_id],
Tags=tags
)
self.tag_cache[instance_id] = {tag['Key']: tag['Value'] for tag in tags}
print(f"Successfully tagged EC2 instance {instance_id}")
return True
except Exception as e:
print(f"Error tagging EC2 instance: {e}")
return False
def tag_s3_bucket(self, bucket_name: str, tags: List[Dict[str, str]]) -> bool:
"""Tag an S3 bucket with cost allocation tags."""
valid, error = self.validate_tags(tags)
if not valid:
print(f"Tag validation failed: {error}")
return False
try:
self.s3.put_bucket_tagging(
Bucket=bucket_name,
Tagging={'TagSet': tags}
)
self.tag_cache[bucket_name] = {tag['Key']: tag['Value'] for tag in tags}
print(f"Successfully tagged S3 bucket {bucket_name}")
return True
except Exception as e:
print(f"Error tagging S3 bucket: {e}")
return False
def tag_rds_instance(self, db_instance_id: str, tags: List[Dict[str, str]]) -> bool:
"""Tag an RDS instance with cost allocation tags."""
valid, error = self.validate_tags(tags)
if not valid:
print(f"Tag validation failed: {error}")
return False
try:
response = self.rds.describe_db_instances(DBInstanceIdentifier=db_instance_id)
arn = response['DBInstances'][0]['DBInstanceArn']
self.rds.add_tags_to_resource(
ResourceName=arn,
Tags=tags
)
self.tag_cache[db_instance_id] = {tag['Key']: tag['Value'] for tag in tags}
print(f"Successfully tagged RDS instance {db_instance_id}")
return True
except Exception as e:
print(f"Error tagging RDS instance: {e}")
return False
def get_resource_tags(self, resource_id: str) -> Dict[str, str]:
"""Get tags for a resource."""
if resource_id in self.tag_cache:
return self.tag_cache[resource_id]
try:
response = self.ec2.describe_tags(
Filters=[{'Name': 'resource-id', 'Values': [resource_id]}]
)
tags = {tag['Key']: tag['Value'] for tag in response.get('Tags', [])}
self.tag_cache[resource_id] = tags
return tags
except Exception:
return {}
def generate_cost_allocation_report(self) -> Dict[str, float]:
"""Generate cost allocation report by tag."""
# This would integrate with AWS Cost Explorer in production
# For demo, we'll simulate the report structure
report = {
'tags_analyzed': len(self.tag_cache),
'cost_by_project': {},
'cost_by_environment': {},
'cost_by_owner': {},
}
# In production, query AWS Cost Explorer with tag dimensions
# Example query structure:
# response = ce.get_cost_and_usage(
# TimePeriod={'Start': '2024-01-01', 'End': '2024-02-01'},
# Granularity='MONTHLY',
# Metrics=['UnblendedCost'],
# GroupBy=[
# {'Type': 'TAG', 'Key': 'Project'},
# {'Type': 'TAG', 'Key': 'Environment'}
# ]
# )
return report
def enforce_tagging_policy(self, resource_type: str, resource_id: str) -> bool:
"""Enforce cost allocation tagging policy on a resource."""
if resource_type == 'ec2':
return self.tag_ec2_instance(resource_id, [])
elif resource_type == 's3':
return self.tag_s3_bucket(resource_id, [])
elif resource_type == 'rds':
return self.tag_rds_instance(resource_id, [])
else:
print(f"Unsupported resource type: {resource_type}")
return False
# Usage example
if __name__ == "__main__":
tag_manager = CostAllocationTagManager(region="us-east-1")
# Define cost allocation tags
cost_tags = [
{'Key': 'Project', 'Value': 'WebApplication'},
{'Key': 'Environment', 'Value': 'production'},
{'Key': 'Owner', 'Value': 'john.doe@company.com'},
{'Key': 'CostCenter', 'Value': 'CC12345'},
{'Key': 'Department', 'Value': 'Engineering'},
{'Key': 'Application', 'Value': 'frontend'},
{'Key': 'BusinessUnit', 'Value': 'Tech'},
]
# Tag a resource (example)
instance_id = "i-1234567890abcdef0"
tag_manager.tag_ec2_instance(instance_id, cost_tags)
# Get tags
tags = tag_manager.get_resource_tags(instance_id)
print(f"Tags for {instance_id}: {tags}")
Multi-cloud cost aggregation provides unified visibility across AWS, Azure, and GCP.
BAD — Separate queries without normalization
# ❌ BAD — No normalization, separate tools for each cloud
# AWS
aws ce get-cost-and-usage --time-period Start=2024-01-01,End=2024-02-01
# Azure
az costmanagement query --type Usage --timeframe MonthToDate
# GCP
gcloud billing budgets describe --billing-account="012345-567890-ABCDEF"
GOOD — Unified multi-cloud cost aggregation framework
#!/bin/bash
# ✅ GOOD — Unified multi-cloud cost aggregation with normalization
set -euo pipefail
# Configuration
OUTPUT_DIR="${OUTPUT_DIR:-/tmp/cost-aggregation}"
START_DATE="${START_DATE:-$(date -d '1 month ago' +%Y-%m-%d)}"
END_DATE="${END_DATE:-$(date +%Y-%m-%d)}"
# Ensure output directory exists
mkdir -p "$OUTPUT_DIR"
# AWS Cost Aggregation
aggregate_aws_costs() {
echo "Aggregating AWS costs..."
aws ce get-cost-and-usage \
--time-period "Start=${START_DATE},End=${END_DATE}" \
--granularity MONTHLY \
--metrics UnblendedCost \
--group-by "Type=DIMENSION,Key=SERVICE" \
--query 'ResultsByTime[].Total[].UnblendedCost Amount' \
--output json > "$OUTPUT_DIR/aws_costs.json"
# Extract total cost
local aws_total=$(cat "$OUTPUT_DIR/aws_costs.json" | jq -r '.ResultsByTime[].Total.UnblendedCost.Amount' | awk '{sum+=$1} END {printf "%.2f", sum}')
echo "AWS Total Cost: \$aws_total"
echo "$aws_total" > "$OUTPUT_DIR/aws_total.txt"
}
# Azure Cost Aggregation
aggregate_azure_costs() {
echo "Aggregating Azure costs..."
local subscription_id=$(az account show --query id -o tsv)
az costmanagement query \
--type "Usage" \
--timeframe "MonthToDate" \
--dataset-aggregation '{"totalCost":{"name":"PreTaxCost","function":"Sum"}}' \
--dataset-grouping '{"name":"ResourceGroup","type":"Dimension"}' \
--subscription "$subscription_id" \
--query "properties.rows[].properties.totalCost" \
--output tsv > "$OUTPUT_DIR/azure_costs.txt"
# Calculate total
local azure_total=$(cat "$OUTPUT_DIR/azure_costs.txt" | awk '{sum+=$1} END {printf "%.2f", sum}')
echo "Azure Total Cost: \$azure_total"
echo "$azure_total" > "$OUTPUT_DIR/azure_total.txt"
}
# GCP Cost Aggregation
aggregate_gcp_costs() {
echo "Aggregating GCP costs..."
# Get billing account
local billing_account=$(gcloud billing accounts list --format="value(name)" | head -1)
# Query BigQuery for costs (requires Cloud Billing Export)
gcloud bigquery queries run "
SELECT
SUM(cost) as total_cost
FROM
\`\${billing_account}.billing_data.gcp_billing_export_v1_012345\`
WHERE
_PARTITIONTIME >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 30 DAY)
" --format="csv" > "$OUTPUT_DIR/gcp_costs.csv"
# Extract total
local gcp_total=$(tail -1 "$OUTPUT_DIR/gcp_costs.csv" | cut -d',' -f1)
echo "GCP Total Cost: \$gcp_total"
echo "$gcp_total" > "$OUTPUT_DIR/gcp_total.txt"
}
# Normalize all costs to USD
normalize_costs() {
echo "Normalizing costs to USD..."
# Get exchange rates (simplified - use actual rates API)
local exchange_rates=(
"USD:1.0"
"EUR:1.08"
"GBP:1.27"
"JPY:0.0068"
)
# In production, use a real exchange rate API
# For now, assume all costs are already in USD
echo "1.0" > "$OUTPUT_DIR/eur_usd_rate.txt"
echo "1.0" > "$OUTPUT_dir/gbp_usd_rate.txt"
echo "1.0" > "$OUTPUT_DIR/jpy_usd_rate.txt"
# Copy costs (already in USD)
cp "$OUTPUT_DIR/aws_total.txt" "$OUTPUT_DIR/normalized_total.txt"
cp "$OUTPUT_DIR/azure_total.txt" "$OUTPUT_DIR/normalized_azure.txt"
cp "$OUTPUT_DIR/gcp_total.txt" "$OUTPUT_DIR/normalized_gcp.txt"
}
# Generate unified report
generate_unified_report() {
echo "Generating unified cost report..."
local aws_cost=$(cat "$OUTPUT_DIR/aws_total.txt" 2>/dev/null || echo "0")
local azure_cost=$(cat "$OUTPUT_DIR/azure_total.txt" 2>/dev/null || echo "0")
local gcp_cost=$(cat "$OUTPUT_DIR/gcp_total.txt" 2>/dev/null || echo "0")
local total_cost=$(echo "$aws_cost + $azure_cost + $gcp_cost" | bc -l)
echo ""
echo "=============================================="
echo " Multi-Cloud Cost Aggregation Report"
echo "=============================================="
echo ""
echo "Period: $START_DATE to $END_DATE"
echo ""
echo "Cost by Provider:"
echo " AWS: \$aws_cost"
echo " Azure: \$azure_cost"
echo " GCP: \$gcp_cost"
echo " Total: \$total_cost"
echo ""
echo "Cost Distribution:"
if (( $(echo "$total_cost > 0" | bc -l) )); then
local aws_pct=$(echo "scale=1; $aws_cost * 100 / $total_cost" | bc -l)
local azure_pct=$(echo "scale=1; $azure_cost * 100 / $total_cost" | bc -l)
local gcp_pct=$(echo "scale=1; $gcp_cost * 100 / $total_cost" | bc -l)
echo " AWS: ${aws_pct}%"
echo " Azure: ${azure_pct}%"
echo " GCP: ${gcp_pct}%"
fi
echo ""
echo "Report generated: $(date)"
echo "=============================================="
# Save report
cat > "$OUTPUT_DIR/unified_report.txt" <<EOF
Multi-Cloud Cost Aggregation Report
====================================
Period: $START_DATE to $END_DATE
Cost by Provider:
AWS: \$aws_cost
Azure: \$azure_cost
GCP: \$gcp_cost
Total: \$total_cost
Cost Distribution:
AWS: ${aws_pct}%
Azure: ${azure_pct}%
GCP: ${gcp_pct}%
Report generated: $(date)
====================================
EOF
}
# Main execution
main() {
echo "Starting multi-cloud cost aggregation..."
echo "Date range: $START_DATE to $END_DATE"
echo ""
aggregate_aws_costs
aggregate_azure_costs
aggregate_gcp_costs
normalize_costs
generate_unified_report
echo ""
echo "Reports saved to: $OUTPUT_DIR"
}
main "$@"
Python Multi-Cloud Cost Aggregator
# ✅ GOOD — Python multi-cloud cost aggregator with normalization
import boto3
from datetime import datetime, timedelta
from typing import Dict, List, Optional
from dataclasses import dataclass
import statistics
import os
@dataclass
class CloudCost:
provider: str
amount: float
currency: str
period_start: datetime
period_end: datetime
breakdown: Optional[Dict[str, float]] = None
class MultiCloudCostAggregator:
"""Aggregates and normalizes costs across AWS, Azure, and GCP."""
def __init__(self):
self.aws_client = boto3.client('ce', region_name='us-east-1')
self.aws_cost_explorer = boto3.client('ce', region_name='us-east-1')
# Azure and GCP costs will be fetched via API/CLI
self.azure_costs: List[Dict] = []
self.gcp_costs: List[Dict] = []
self.exchange_rates: Dict[str, float] = {
'USD': 1.0,
'EUR': 1.08,
'GBP': 1.27,
'JPY': 0.0068,
}
def get_aws_costs(self, days: int = 30) -> CloudCost:
"""Get AWS costs for the specified period."""
end_date = datetime.utcnow().date()
start_date = end_date - timedelta(days=days)
try:
response = self.aws_cost_explorer.get_cost_and_usage(
TimePeriod={
'Start': start_date.isoformat(),
'End': end_date.isoformat()
},
Granularity='MONTHLY',
Metrics=['UnblendedCost'],
GroupBy=[
{'Type': 'DIMENSION', 'Key': 'SERVICE'},
{'Type': 'TAG', 'Key': 'Project'}
]
)
# Extract total cost
total_cost = 0
service_costs: Dict[str, float] = {}
for result in response['ResultsByTime'][-1]['Groups']:
amount = float(result['Metrics']['UnblendedCost']['Amount'])
key = result['Keys'][0] # Service name
total_cost += amount
service_costs[key] = service_costs.get(key, 0) + amount
return CloudCost(
provider='aws',
amount=round(total_cost, 2),
currency='USD',
period_start=start_date,
period_end=end_date,
breakdown=service_costs,
)
except Exception as e:
print(f"Error fetching AWS costs: {e}")
return CloudCost(
provider='aws',
amount=0,
currency='USD',
period_start=start_date,
period_end=end_date,
)
def get_azure_costs(self, days: int = 30) -> CloudCost:
"""Get Azure costs for the specified period."""
# In production, use Azure Cost Management SDK
# This is a simplified example
from datetime import datetime
end_date = datetime.utcnow().date()
start_date = end_date - timedelta(days=days)
try:
# Example Azure Cost Management SDK usage
# from azure.identity import DefaultAzureCredential
# from azure.mgmt.costmanagement import CostManagementClient
# For demo, return mock data
azure_cost = 1500.0 # Example cost
return CloudCost(
provider='azure',
amount=round(azure_cost, 2),
currency='USD', # Azure costs are typically in USD
period_start=start_date,
period_end=end_date,
)
except Exception as e:
print(f"Error fetching Azure costs: {e}")
return CloudCost(
provider='azure',
amount=0,
currency='USD',
period_start=start_date,
period_end=end_date,
)
def get_gcp_costs(self, days: int = 30) -> CloudCost:
"""Get GCP costs for the specified period."""
from datetime import datetime
end_date = datetime.utcnow().date()
start_date = end_date - timedelta(days=days)
try:
# Example GCP Billing API usage
# from google.cloud import billing_v1
# client = billing_v1.CloudBillingClient()
# For demo, return mock data
gcp_cost = 2000.0 # Example cost
return CloudCost(
provider='gcp',
amount=round(gcp_cost, 2),
currency='USD', # GCP costs are typically in USD
period_start=start_date,
period_end=end_date,
)
except Exception as e:
print(f"Error fetching GCP costs: {e}")
return CloudCost(
provider='gcp',
amount=0,
currency='USD',
period_start=start_date,
period_end=end_date,
)
def normalize_to_usd(self, amount: float, currency: str) -> float:
"""Normalize an amount to USD using exchange rates."""
if currency == 'USD':
return amount
rate = self.exchange_rates.get(currency, 1.0)
return round(amount * rate, 2)
def aggregate_costs(self, days: int = 30) -> Dict:
"""Aggregate costs from all cloud providers."""
aws_cost = self.get_aws_costs(days)
azure_cost = self.get_azure_costs(days)
gcp_cost = self.get_gcp_costs(days)
# Normalize all costs to USD
aws_usd = self.normalize_to_usd(aws_cost.amount, aws_cost.currency)
azure_usd = self.normalize_to_usd(azure_cost.amount, azure_cost.currency)
gcp_usd = self.normalize_to_usd(gcp_cost.amount, gcp_cost.currency)
total_cost = aws_usd + azure_usd + gcp_usd
return {
'period': {
'start': aws_cost.period_start.isoformat(),
'end': aws_cost.period_end.isoformat(),
},
'providers': {
'aws': {
'amount': aws_cost.amount,
'currency': aws_cost.currency,
'normalized_usd': aws_usd,
'breakdown': aws_cost.breakdown,
},
'azure': {
'amount': azure_cost.amount,
'currency': azure_cost.currency,
'normalized_usd': azure_usd,
},
'gcp': {
'amount': gcp_cost.amount,
'currency': gcp_cost.currency,
'normalized_usd': gcp_usd,
},
},
'totals': {
'total_usd': round(total_cost, 2),
'aws_percentage': round((aws_usd / total_cost) * 100, 1) if total_cost > 0 else 0,
'azure_percentage': round((azure_usd / total_cost) * 100, 1) if total_cost > 0 else 0,
'gcp_percentage': round((gcp_usd / total_cost) * 100, 1) if total_cost > 0 else 0,
},
}
def generate_cost_report(self, days: int = 30) -> str:
"""Generate a comprehensive cost report."""
costs = self.aggregate_costs(days)
report = [
"=" * 80,
" Multi-Cloud Cost Aggregation Report",
"=" * 80,
"",
f"Period: {costs['period']['start']} to {costs['period']['end']}",
"",
"Cost by Provider:",
f" AWS: ${costs['providers']['aws']['normalized_usd']:,.2f} ({costs['providers']['aws']['amount']:.2f} {costs['providers']['aws']['currency']})",
f" Azure: ${costs['providers']['azure']['normalized_usd']:,.2f} ({costs['providers']['azure']['amount']:.2f} {costs['providers']['azure']['currency']})",
f" GCP: ${costs['providers']['gcp']['normalized_usd']:,.2f} ({costs['providers']['gcp']['amount']:.2f} {costs['providers']['gcp']['currency']})",
"",
f" Total: ${costs['totals']['total_usd']:,.2f}",
"",
"Cost Distribution:",
f" AWS: {costs['totals']['aws_percentage']}%",
f" Azure: {costs['totals']['azure_percentage']}%",
f" GCP: {costs['totals']['gcp_percentage']}%",
"",
]
# AWS breakdown by service (if available)
if costs['providers']['aws']['breakdown']:
report.extend([
"AWS Breakdown by Service:",
])
for service, amount in sorted(costs['providers']['aws']['breakdown'].items(), key=lambda x: x[1], reverse=True)[:10]:
pct = (amount / costs['providers']['aws']['normalized_usd']) * 100 if costs['providers']['aws']['normalized_usd'] > 0 else 0
report.append(f" {service}: ${amount:,.2f} ({pct:.1f}%)")
report.append("")
report.append("=" * 80)
return "\n".join(report)
# Usage example
if __name__ == "__main__":
aggregator = MultiCloudCostAggregator()
report = aggregator.generate_cost_report(days=30)
print(report)
When implementing cost optimization strategies, return the following structured output:
Cost Optimization Analysis Report
=================================
Analysis Period: YYYY-MM-DD to YYYY-MM-DD
Total Estimated Monthly Savings: $X,XXX.XX
Implementation Timeline: X weeks
Projected ROI: XX%
AWS:
Current Monthly Cost: $X,XXX.XX
Right-Sizing Savings: $XXX.XX (XX%)
Spot Instance Savings: $XXX.XX (XX%)
Reserved Instance Savings: $XXX.XX (XX%)
Total Potential Savings: $XXX.XX (XX%)
Azure:
Current Monthly Cost: $X,XXX.XX
Right-Sizing Savings: $XXX.XX (XX%)
Spot Instance Savings: $XXX.XX (XX%)
Reserved Instance Savings: $XXX.XX (XX%)
Total Potential Savings: $XXX.XX (XX%)
GCP:
Current Monthly Cost: $X,XXX.XX
Right-Sizing Savings: $XXX.XX (XX%)
Spot Instance Savings: $XXX.XX (XX%)
Reserved Instance Savings: $XXX.XX (XX%)
Total Potential Savings: $XXX.XX (XX%)
Priority 1 (Week 1-2):
- Implement cost allocation tagging for all resources
- Set up automated cost alerts at 50%, 75%, 90%, 100% thresholds
Priority 2 (Week 3-4):
- Right-size low-utilization resources (CPU < 30% for 14 days)
- Implement spot instance strategy for fault-tolerant workloads
Priority 3 (Month 2):
- Purchase reserved instances for stable, high-utilization workloads
- Implement storage tier optimization
Priority 4 (Ongoing):
- Monthly cost reviews and optimization cycle
- Continuous right-sizing based on usage patterns
Spot Instance Risks:
- Interruption rate: XX% (based on historical data)
- Replacement time: X minutes average
- Mitigation: Maintain buffer capacity, use mixed instance types
Right-Sizing Risks:
- Performance impact: Low/Medium/High
- Application impact: Minimal/Moderate/Significant
- Mitigation: Gradual rollout, monitoring, rollback plan
Reserved Instance Risks:
- Commitment period: 1 or 3 years
- Early termination fees: XXX%
- Mitigation: Start with 1-year terms, monitor usage patterns
Cost Alerts:
- 50% threshold: Informational notification
- 75% threshold: Manager notification
- 90% threshold: Emergency notification
- 100% threshold: Automatic budget freeze (optional)
Optimization Metrics:
- Monthly cost trend
- Cost per workload
- Cost per user
- ROI on optimization initiatives
Review Cadence:
- Daily: Automated cost anomaly detection
- Weekly: Team cost review
- Monthly: Executive cost report
- Quarterly: Optimization strategy review
| Skill | Purpose |
|---|---|
cncf-cost-optimization | Complementary skill with different optimization focus; use this skill for detailed analysis and implementation |
cncf-kubernetes-debugging | Kubernetes cost optimization often requires debugging resource allocations and pod scheduling |
agent-database-admin | Database optimization is a key component of cloud cost management; use for RDS, Cloud SQL, Cosmos DB optimization |
Optimize storage costs by analyzing access patterns and implementing appropriate tiering strategies.
BAD — No access pattern analysis
# ❌ BAD — No analysis, all data in expensive storage
# Just moving everything to S3 Intelligent-Tiering without analysis
aws s3api put-bucket-lifecycle-configuration \
--bucket my-bucket \
--lifecycle-configuration file://lifecycle.json
GOOD — Data-driven storage tier optimization
#!/bin/bash
# ✅ GOOD — Access pattern analysis for storage tier recommendations
set -euo pipefail
BUCKET="${1:-my-bucket}"
REGION="${AWS_REGION:-us-east-1}"
# Get bucket metrics
aws s3api get-bucket-metrics-configuration \
--bucket "$BUCKET" \
--id "StorageMetrics" \
--query 'BucketMetricsConfiguration.StorageClassMetrics' \
--output json 2>/dev/null || echo "Metrics not configured"
# Analyze access patterns
echo "=== Storage Tier Analysis for $BUCKET ==="
echo ""
# Check current lifecycle configuration
aws s3api get-bucket-lifecycle-configuration \
--bucket "$BUCKET" \
--query 'Rules[].{ID:ID,Status:Status,Transitions:Transitions}' \
--output table 2>/dev/null || echo "No lifecycle configuration found"
# Get object count by size
echo ""
echo "Storage Distribution:"
aws s3api list-objects-v2 \
--bucket "$BUCKET" \
--query 'Contents[].{Key:Key,Size:Size,LastModified:LastModified}' \
--max-items 100 \
--output table 2>/dev/null | head -50
# Generate tier recommendations
echo ""
echo "=== Tier Recommendations ==="
echo "S3 Standard: Objects accessed >= 1 time/month"
echo "S3 Intelligent: Objects with variable access patterns"
echo "S3 Standard-IA: Objects accessed < 1 time/month"
echo "S3 One-Zone IA: Non-critical objects, < 1 access/month"
echo "S3 Glacier: Objects accessed < 1 time/year"
echo "S3 Glacier Deep: Objects accessed < 1 time/7 years"
Python Storage Optimization Engine
# ✅ GOOD — Python storage optimization with access pattern analysis
import boto3
from datetime import datetime, timedelta
from typing import Dict, List, Optional
from collections import defaultdict
import statistics
class StorageTierOptimizer:
"""Optimizes storage tier assignments based on access patterns."""
def __init__(self, region: str = "us-east-1"):
self.s3 = boto3.client('s3', region_name=region)
self.cloudwatch = boto3.client('cloudwatch', region_name=region)
# Access pattern thresholds (days since last access)
self.thresholds = {
'standard': 30, # Accessed >= 1 time/month
'standard_ia': 90, # Accessed < 1 time/month
'one_zone': 90, # Non-critical, accessed < 1 time/month
'glacier': 365, # Accessed < 1 time/year
'glacier_deep': 2555, # Accessed < 1 time/7 years
}
self.tier_costs = {
'standard': 0.023, # $/GB/month
'standard_ia': 0.0125, # $/GB/month
'one_zone': 0.0100, # $/GB/month
'glacier': 0.004, # $/GB/month
'glacier_deep': 0.00099, # $/GB/month
}
def get_object_access_pattern(self, bucket: str, prefix: str = "") -> Dict[str, int]:
"""Get access patterns for objects in a bucket."""
access_patterns = defaultdict(int)
# List objects with pagination
paginator = self.s3.get_paginator('list_objects_v2')
page_iterator = paginator.paginate(Bucket=bucket, Prefix=prefix)
for page in page_iterator:
for obj in page.get('Contents', []):
key = obj['Key']
size = obj['Size']
# Classify by size
if size < 1024: # < 1KB
access_patterns['tiny'] += 1
elif size < 1024 * 1024: # < 1MB
access_patterns['small'] += 1
elif size < 1024 * 1024 * 1024: # < 1GB
access_patterns['medium'] += 1
else:
access_patterns['large'] += 1
return dict(access_patterns)
def analyze_bucket_storage(self, bucket: str) -> Dict:
"""Analyze bucket storage and recommend tier assignments."""
try:
# Get lifecycle configuration
try:
lifecycle = self.s3.get_bucket_lifecycle_configuration(Bucket=bucket)
current_tiers = [rule['Transitions'][0]['StorageClass']
for rule in lifecycle.get('Rules', [])]
except Exception:
current_tiers = []
# Get storage metrics
metrics = self.cloudwatch.get_metric_statistics(
Namespace='AWS/S3',
MetricName='BucketSizeBytes',
Dimensions=[
{'Name': 'BucketName', 'Value': bucket},
{'Name': 'StorageType', 'Value': 'StandardStorage'}
],
StartTime=datetime.utcnow() - timedelta(days=30),
EndTime=datetime.utcnow(),
Period=86400,
Statistics=['Average']
)
avg_size_bytes = statistics.mean([
dp['Average'] for dp in metrics.get('Datapoints', [])
]) if metrics.get('Datapoints') else 0
avg_size_gb = avg_size_bytes / (1024 ** 3)
# Calculate potential savings
current_cost = avg_size_gb * self.tier_costs['standard']
# Recommend optimal tier based on size and access pattern
if avg_size_gb < 100:
recommended_tier = 'standard'
elif avg_size_gb < 1000:
recommended_tier = 'standard_ia'
elif avg_size_gb < 10000:
recommended_tier = 'one_zone'
else:
recommended_tier = 'glacier'
recommended_cost = avg_size_gb * self.tier_costs[recommended_tier]
savings = current_cost - recommended_cost
savings_percentage = (savings / current_cost) * 100 if current_cost > 0 else 0
return {
'bucket': bucket,
'current_size_gb': round(avg_size_gb, 2),
'current_tier': 'Standard',
'recommended_tier': recommended_tier,
'current_cost_monthly': round(current_cost, 2),
'recommended_cost_monthly': round(recommended_cost, 2),
'monthly_savings': round(savings, 2),
'savings_percentage': round(savings_percentage, 1),
'tier_breakdown': {
'standard': round(avg_size_gb * 0.3, 2),
'standard_ia': round(avg_size_gb * 0.4, 2),
'one_zone': round(avg_size_gb * 0.2, 2),
'glacier': round(avg_size_gb * 0.1, 2),
},
}
except Exception as e:
print(f"Error analyzing bucket {bucket}: {e}")
return {}
def optimize_all_buckets(self) -> List[Dict]:
"""Optimize storage for all buckets."""
buckets = self.s3.list_buckets()['Buckets']
results = []
for bucket in buckets:
result = self.analyze_bucket_storage(bucket['Name'])
if result:
results.append(result)
return results
# Usage example
if __name__ == "__main__":
optimizer = StorageTierOptimizer(region="us-east-1")
buckets = optimizer.optimize_all_buckets()
print("Storage Tier Optimization Report")
print("=" * 60)
total_monthly_savings = 0
for bucket in buckets:
if bucket['savings_percentage'] > 10:
print(f"\nBucket: {bucket['bucket']}")
print(f" Current Size: {bucket['current_size_gb']:.2f} GB")
print(f" Current Tier: {bucket['current_tier']}")
print(f" Recommended: {bucket['recommended_tier']}")
print(f" Monthly Savings: ${bucket['monthly_savings']:.2f}")
total_monthly_savings += bucket['monthly_savings']
print(f"\n" + "=" * 60)
print(f"Total Monthly Savings: ${total_monthly_savings:.2f}")
Optimize Kubernetes resource requests and limits to reduce cluster costs.
BAD — No resource optimization, default settings
# ❌ BAD — No resource analysis, just applying defaults
kubectl apply -f deployment.yaml
# Deployment has no resource requests/limits defined
GOOD — Kubernetes resource optimization with metrics
#!/bin/bash
# ✅ GOOD — Kubernetes resource optimization with metrics analysis
set -euo pipefail
NAMESPACE="${NAMESPACE:-default}"
DAYS_ANALYZED="${DAYS_ANALYZED:-14}"
echo "=== Kubernetes Resource Optimization ==="
echo "Namespace: $NAMESPACE"
echo "Analysis Period: $DAYS_ANALYZED days"
echo ""
# Get all deployments
deployments=$(kubectl get deployments -n "$NAMESPACE" -o jsonpath='{.items[*].metadata.name}')
for deployment in $deployments; do
echo "Analyzing deployment: $deployment"
# Get pod resource usage (requires metrics server)
kubectl top pods -n "$NAMESPACE" -l "app=$deployment" 2>/dev/null || echo "Metrics not available"
# Get deployment resource configuration
echo "Current resource configuration:"
kubectl get deployment "$deployment" -n "$NAMESPACE" \
-o jsonpath='{.spec.template.spec.containers[*].resources}' 2>/dev/null || echo "No resources defined"
echo ""
done
# Generate recommendations
echo "=== Optimization Recommendations ==="
echo "1. Set CPU requests to 50-70% of measured peak"
echo "2. Set CPU limits to 150-200% of requests"
echo "3. Set memory requests to 70-80% of measured peak"
echo "4. Set memory limits to 150-200% of requests"
echo "5. Use Vertical Pod Autoscaler for automatic optimization"
Python K8s Resource Optimizer
# ✅ GOOD — Python K8s resource optimizer with metrics
import subprocess
import json
from datetime import datetime, timedelta
from typing import Dict, List, Optional
from dataclasses import dataclass
import statistics
@dataclass
class PodMetrics:
name: str
cpu_usage: float # millicores
memory_usage: float # MB
cpu_request: Optional[float] = None
memory_request: Optional[float] = None
cpu_limit: Optional[float] = None
memory_limit: Optional[float] = None
class K8sResourceOptimizer:
"""Optimizes Kubernetes resource requests and limits."""
def __init__(self):
self.metrics_server_available = self._check_metrics_server()
def _check_metrics_server(self) -> bool:
"""Check if metrics server is available."""
try:
result = subprocess.run(
['kubectl', 'get', 'apiservice', 'v1beta1.metrics.k8s.io', '-o', 'jsonpath={.status.conditions[0].status}'],
capture_output=True, text=True
)
return result.stdout.strip() == 'True'
except Exception:
return False
def get_pod_metrics(self, namespace: str, pod: str) -> Optional[PodMetrics]:
"""Get resource metrics for a pod."""
if not self.metrics_server_available:
return None
try:
result = subprocess.run(
['kubectl', 'top', 'pod', pod, '-n', namespace, '--no-headers', '-o', 'json'],
capture_output=True, text=True, check=True
)
metrics = json.loads(result.stdout)
return PodMetrics(
name=pod,
cpu_usage=metrics.get('usage', {}).get('cpu', '0').rstrip('n') or '0',
memory_usage=metrics.get('usage', {}).get('memory', '0').rstrip('Ki') or '0',
)
except Exception as e:
print(f"Error getting metrics for {pod}: {e}")
return None
def get_deployment_resources(self, namespace: str, deployment: str) -> Dict:
"""Get resource configuration for a deployment."""
try:
result = subprocess.run(
['kubectl', 'get', 'deployment', deployment, '-n', namespace, '-o', 'json'],
capture_output=True, text=True, check=True
)
deployment_data = json.loads(result.stdout)
containers = []
for container in deployment_data['spec']['template']['spec'].get('containers', []):
resources = container.get('resources', {})
containers.append({
'name': container['name'],
'requests': resources.get('requests', {}),
'limits': resources.get('limits', {}),
})
return {'containers': containers}
except Exception as e:
print(f"Error getting resources for {deployment}: {e}")
return {'containers': []}
def calculate_optimal_resources(self, metrics: List[PodMetrics]) -> Dict[str, float]:
"""Calculate optimal resource requests and limits based on metrics."""
if not metrics:
return {}
cpu_values = [m.cpu_usage for m in metrics]
memory_values = [m.memory_usage for m in metrics]
# Use 70th percentile for requests
cpu_70th = sorted(cpu_values)[int(len(cpu_values) * 0.7)] if cpu_values else 0
memory_70th = sorted(memory_values)[int(len(memory_values) * 0.7)] if memory_values else 0
# Use 95th percentile for limits
cpu_95th = sorted(cpu_values)[int(len(cpu_values) * 0.95)] if cpu_values else 0
memory_95th = sorted(memory_values)[int(len(memory_values) * 0.95)] if memory_values else 0
return {
'cpu_request': round(cpu_70th * 0.7, 0), # 70% of 70th percentile
'cpu_limit': round(cpu_95th * 1.5, 0),
'memory_request': round(memory_70th * 0.7, 0),
'memory_limit': round(memory_95th * 1.5, 0),
}
def analyze_namespace(self, namespace: str) -> List[Dict]:
"""Analyze resource optimization for all pods in a namespace."""
if not self.metrics_server_available:
return [{'error': 'Metrics server not available'}]
try:
result = subprocess.run(
['kubectl', 'get', 'pods', '-n', namespace, '-o', 'jsonpath={.items[*].metadata.name}'],
capture_output=True, text=True, check=True
)
pods = result.stdout.split()
results = []
for pod in pods:
metrics = self.get_pod_metrics(namespace, pod)
if metrics:
optimal = self.calculate_optimal_resources([metrics])
# Calculate potential savings
current_cpu = metrics.cpu_limit or 500 # Default
current_memory = metrics.memory_limit or 512 # Default
optimal_cpu = optimal.get('cpu_limit', current_cpu)
optimal_memory = optimal.get('memory_limit', current_memory)
# Estimate cost savings (simplified)
cpu_savings = (current_cpu - optimal_cpu) * 0.000011 # $ per millicore per hour
memory_savings = (current_memory - optimal_memory) * 0.0000037 # $ per MB per hour
monthly_savings = (cpu_savings + memory_savings) * 730
results.append({
'pod': pod,
'current_cpu': current_cpu,
'current_memory': current_memory,
'optimal_cpu': optimal_cpu,
'optimal_memory': optimal_memory,
'monthly_savings': round(monthly_savings, 2),
'metrics': metrics,
})
return results
except Exception as e:
return [{'error': str(e)}]
def optimize_all_namespaces(self) -> Dict[str, List[Dict]]:
"""Optimize resources across all namespaces."""
try:
result = subprocess.run(
['kubectl', 'get', 'namespaces', '-o', 'jsonpath={.items[*].metadata.name}'],
capture_output=True, text=True, check=True
)
namespaces = result.stdout.split()
results = {}
for namespace in namespaces:
if namespace not in ['kube-system', 'kube-public', 'kube-node-lease']:
results[namespace] = self.analyze_namespace(namespace)
return results
except Exception as e:
return {'error': str(e)}
# Usage example
if __name__ == "__main__":
optimizer = K8sResourceOptimizer()
# Get current namespace
try:
result = subprocess.run(
['kubectl', 'config', 'get-contexts', '--output', 'name'],
capture_output=True, text=True, check=True
)
current_context = result.stdout.strip().split('*')[-1].strip()
except Exception:
current_context = "default"
print("Kubernetes Resource Optimization")
print("=" * 60)
print(f"Current Context: {current_context}")
results = optimizer.analyze_namespace(current_context)
total_monthly_savings = 0
for result in results:
if 'monthly_savings' in result:
if result['monthly_savings'] > 10:
print(f"\nPod: {result['pod']}")
print(f" Current CPU: {result['current_cpu']}m")
print(f" Optimal CPU: {result['optimal_cpu']}m")
print(f" Monthly Savings: ${result['monthly_savings']:.2f}")
total_monthly_savings += result['monthly_savings']
print(f"\n" + "=" * 60)
print(f"Total Monthly Savings: ${total_monthly_savings:.2f}")