Cloud Cost Optimization

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196

"""
Right-Sizing Analysis: Collect and Analyze EC2 Utilization
 
Uses CloudWatch metrics to identify right-sizing opportunities.
"""
 
import boto3
from datetime import datetime, timedelta
from dataclasses import dataclass
from typing import List, Dict
import statistics
 
@dataclass
class InstanceUtilization:
    instance_id: str
    instance_type: str
    avg_cpu: float
    max_cpu: float
    p95_cpu: float
    avg_memory: float  # Requires CloudWatch agent
    p95_memory: float
    current_vcpus: int
    current_memory_gb: float
    recommendation: str
    potential_savings: float
 
def get_cpu_metrics(
    cloudwatch: boto3.client,
    instance_id: str,
    days: int = 14
) -> List[float]:
    """Fetch CPU utilization data points."""
    
    end_time = datetime.utcnow()
    start_time = end_time - timedelta(days=days)
    
    response = cloudwatch.get_metric_statistics(
        Namespace='AWS/EC2',
        MetricName='CPUUtilization',
        Dimensions=[{'Name': 'InstanceId', 'Value': instance_id}],
        StartTime=start_time,
        EndTime=end_time,
        Period=300,  # 5-minute granularity
        Statistics=['Average']
    )
    
    return [dp['Average'] for dp in response['Datapoints']]
 
def analyze_instance(
    instance: dict,
    cpu_data: List[float],
    memory_data: List[float],
    instance_prices: Dict[str, float]
) -> InstanceUtilization:
    """Analyze a single instance for right-sizing opportunities."""
    
    instance_id = instance['InstanceId']
    instance_type = instance['InstanceType']
    
    # Calculate CPU statistics
    avg_cpu = statistics.mean(cpu_data) if cpu_data else 0
    max_cpu = max(cpu_data) if cpu_data else 0
    p95_cpu = sorted(cpu_data)[int(len(cpu_data) * 0.95)] if cpu_data else 0
    
    # Calculate memory statistics (requires CloudWatch agent)
    avg_memory = statistics.mean(memory_data) if memory_data else 0
    p95_memory = sorted(memory_data)[int(len(memory_data) * 0.95)] if memory_data else 0
    
    # Get instance capacity (simplified lookup)
    vcpus, memory_gb = get_instance_specs(instance_type)
    
    # Determine recommendation
    recommendation, new_type = determine_recommendation(
        instance_type, avg_cpu, p95_cpu, avg_memory, p95_memory
    )
    
    # Calculate potential savings
    current_price = instance_prices.get(instance_type, 0)
    new_price = instance_prices.get(new_type, current_price)
    monthly_savings = (current_price - new_price) * 730  # hours/month
    
    return InstanceUtilization(
        instance_id=instance_id,
        instance_type=instance_type,
        avg_cpu=round(avg_cpu, 2),
        max_cpu=round(max_cpu, 2),
        p95_cpu=round(p95_cpu, 2),
        avg_memory=round(avg_memory, 2),
        p95_memory=round(p95_memory, 2),
        current_vcpus=vcpus,
        current_memory_gb=memory_gb,
        recommendation=recommendation,
        potential_savings=round(monthly_savings, 2)
    )
 
def determine_recommendation(
    current_type: str,
    avg_cpu: float,
    p95_cpu: float,
    avg_memory: float,
    p95_memory: float
) -> tuple:
    """
    Recommendation logic based on utilization thresholds.
    
    Conservative approach:
    - Only recommend downsize if P95 < 50%
    - Recommend upsize if P95 > 80%
    - Keep buffer for unexpected peaks
    """
    
    # Define thresholds (adjustable based on risk tolerance)
    DOWNSIZE_THRESHOLD = 50  # P95 below this = downsize candidate
    UPSIZE_THRESHOLD = 80   # P95 above this = upsize candidate
    
    cpu_under = p95_cpu < DOWNSIZE_THRESHOLD
    memory_under = p95_memory < DOWNSIZE_THRESHOLD if p95_memory > 0 else True
    
    cpu_over = p95_cpu > UPSIZE_THRESHOLD
    memory_over = p95_memory > UPSIZE_THRESHOLD
    
    if cpu_under and memory_under:
        new_type = get_smaller_type(current_type)
        return f"DOWNSIZE to {new_type}", new_type
    elif cpu_over or memory_over:
        new_type = get_larger_type(current_type)
        return f"UPSIZE to {new_type}", new_type
    else:
        return "OPTIMAL - No change recommended", current_type
 
def get_instance_specs(instance_type: str) -> tuple:
    """Return (vCPUs, memory_GB) for instance type."""
    # Simplified lookup - production would use AWS pricing API
    specs = {
        'm5.large': (2, 8),
        'm5.xlarge': (4, 16),
        'm5.2xlarge': (8, 32),
        'm5.4xlarge': (16, 64),
        'c5.large': (2, 4),
        'c5.xlarge': (4, 8),
        'r5.large': (2, 16),
        'r5.xlarge': (4, 32),
    }
    return specs.get(instance_type, (4, 16))
 
def get_smaller_type(instance_type: str) -> str:
    """Return one size smaller instance type."""
    size_order = ['nano', 'micro', 'small', 'medium', 'large', 'xlarge', '2xlarge', '4xlarge']
    family, size = instance_type.rsplit('.', 1)
    
    if size in size_order:
        idx = size_order.index(size)
        if idx > 0:
            return f"{family}.{size_order[idx-1]}"
    return instance_type
 
def get_larger_type(instance_type: str) -> str:
    """Return one size larger instance type."""
    size_order = ['nano', 'micro', 'small', 'medium', 'large', 'xlarge', '2xlarge', '4xlarge']
    family, size = instance_type.rsplit('.', 1)
    
    if size in size_order:
        idx = size_order.index(size)
        if idx < len(size_order) - 1:
            return f"{family}.{size_order[idx+1]}"
    return instance_type
 
# Main execution
def run_analysis():
    ec2 = boto3.client('ec2')
    cloudwatch = boto3.client('cloudwatch')
    
    # Get running instances
    instances = ec2.describe_instances(
        Filters=[{'Name': 'instance-state-name', 'Values': ['running']}]
    )
    
    results = []
    for reservation in instances['Reservations']:
        for instance in reservation['Instances']:
            cpu_data = get_cpu_metrics(cloudwatch, instance['InstanceId'])
            memory_data = []  # Would require CloudWatch agent data
            
            analysis = analyze_instance(
                instance, cpu_data, memory_data,
                instance_prices={}  # Would load from pricing API
            )
            results.append(analysis)
            
            if 'DOWNSIZE' in analysis.recommendation:
                print(f"💰 {analysis.instance_id}: {analysis.recommendation}")
                print(f"   Current: {analysis.instance_type} ({analysis.avg_cpu}% avg, {analysis.p95_cpu}% p95)")
                print(f"   Savings: ${analysis.potential_savings}/month")
    
    total_savings = sum(r.potential_savings for r in results)
    print(f"\nTotal potential monthly savings: ${total_savings:, .2f}")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

# Karpenter NodePool with Consolidation
# Automatically consolidates under-utilized nodes
apiVersion: karpenter.sh/v1beta1
kind: NodePool
metadata:
  name: default
spec:
  template:
    spec:
      requirements:
        - key: kubernetes.io/arch
          operator: In
          values: ["amd64"]
        - key: karpenter.sh/capacity-type
          operator: In
          values: ["spot", "on-demand"]
        - key: node.kubernetes.io/instance-type
          operator: In
          values:
            - m6i.large
            - m6i.xlarge
            - m6i.2xlarge
            - m6a.large
            - m6a.xlarge
            - m6a.2xlarge
            - c6i.large
            - c6i.xlarge
      nodeClassRef:
        name: default
        
  disruption:
    # Enable consolidation - Karpenter will:
    # 1. Delete empty nodes
    # 2. Replace under-utilized nodes with smaller ones
    # 3. Consolidate pods onto fewer nodes
    consolidationPolicy: WhenUnderutilized
    consolidateAfter: 30s  # Wait 30s before consolidating
    
  # Limit total resources for cost control
  limits:
    cpu: 1000
    memory: 1000Gi

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181

"""
Automated Right-Sizing Pipeline
 
Fetches Compute Optimizer recommendations and creates Jira tickets
for human review, with option for automated implementation in non-prod.
"""
 
import boto3
import json
from datetime import datetime
from jira import JIRA
 
class RightSizingPipeline:
    def __init__(self):
        self.compute_optimizer = boto3.client('compute-optimizer')
        self.ec2 = boto3.client('ec2')
        self.ssm = boto3.client('ssm')
        
        # Jira connection (configure in environment)
        self.jira = JIRA(
            server='https://company.atlassian.net',
            basic_auth=('user', 'api_token')
        )
    
    def get_recommendations(self, finding_status: str = 'Overprovisioned'):
        """Fetch EC2 right-sizing recommendations."""
        
        response = self.compute_optimizer.get_ec2_instance_recommendations(
            filters=[
                {'name': 'Finding', 'values': [finding_status]}
            ]
        )
        
        recommendations = []
        for rec in response.get('instanceRecommendations', []):
            instance_id = rec['instanceArn'].split('/')[-1]
            current_type = rec['currentInstanceType']
            
            # Get top recommendation
            if rec.get('recommendationOptions'):
                best_option = rec['recommendationOptions'][0]
                recommended_type = best_option['instanceType']
                estimated_savings = best_option.get('projectedUtilizationMetrics', [])
                
                recommendations.append({
                    'instance_id': instance_id,
                    'current_type': current_type,
                    'recommended_type': recommended_type,
                    'finding': rec['finding'],
                    'finding_reasons': rec.get('findingReasonCodes', []),
                    'utilization': rec.get('utilizationMetrics', []),
                    'estimated_monthly_savings': self._calculate_savings(
                        current_type, recommended_type
                    )
                })
        
        return recommendations
    
    def _calculate_savings(self, current_type: str, recommended_type: str) -> float:
        """Calculate estimated monthly savings (simplified)."""
        # Production: Use AWS Pricing API
        prices = {
            'm5.2xlarge': 0.384, 'm5.xlarge': 0.192, 'm5.large': 0.096,
            'c5.2xlarge': 0.340, 'c5.xlarge': 0.170, 'c5.large': 0.085,
        }
        current_price = prices.get(current_type, 0.20)
        new_price = prices.get(recommended_type, 0.20)
        return round((current_price - new_price) * 730, 2)  # Monthly hours
    
    def create_ticket(self, recommendation: dict, environment: str):
        """Create Jira ticket for right-sizing review."""
        
        avg_cpu = next(
            (m['value'] for m in recommendation['utilization'] 
             if m['name'] == 'CPU'), 'N/A'
        )
        
        description = f"""
*Automated Right-Sizing Recommendation*
 
|*Field*|*Value*|
|Instance ID|{recommendation['instance_id']}|
|Current Type|{recommendation['current_type']}|
|Recommended Type|{recommendation['recommended_type']}|
|Finding|{recommendation['finding']}|
|Average CPU|{avg_cpu}%|
|Est. Monthly Savings|${recommendation['estimated_monthly_savings']}|
 
*Finding Reasons:*
{chr(10).join(f'* {r}' for r in recommendation['finding_reasons'])}
 
*Next Steps:*
1. Review utilization metrics in CloudWatch
2. Verify workload can run on smaller instance
3. Schedule maintenance window for change
4. Monitor post-change performance
 
_This ticket was auto-generated by the Right-Sizing Pipeline_
        """
        
        issue = self.jira.create_issue(
            project='CLOUD',
            summary=f"Right-Size {recommendation['instance_id']}: "
                    f"{recommendation['current_type']} → {recommendation['recommended_type']}",
            description=description,
            issuetype={'name': 'Task'},
            labels=['right-sizing', 'cost-optimization', environment],
        )
        
        return issue.key
    
    def auto_implement_nonprod(self, recommendation: dict):
        """
        Automatically implement right-sizing for non-production instances.
        Uses SSM automation with proper safety checks.
        """
        
        instance_id = recommendation['instance_id']
        new_type = recommendation['recommended_type']
        
        # Safety checks
        tags = self.ec2.describe_tags(
            Filters=[{'Name': 'resource-id', 'Values': [instance_id]}]
        )
        
        env_tag = next(
            (t['Value'] for t in tags['Tags'] if t['Key'] == 'environment'),
            'unknown'
        )
        
        if env_tag not in ['development', 'staging', 'sandbox']:
            print(f"Skipping auto-implementation for {instance_id}: env={env_tag}")
            return None
        
        # Stop, modify, start using SSM Automation
        response = self.ssm.start_automation_execution(
            DocumentName='AWS-ResizeInstance',
            Parameters={
                'InstanceId': [instance_id],
                'InstanceType': [new_type],
            }
        )
        
        return response['AutomationExecutionId']
    
    def run_pipeline(self, auto_implement_nonprod: bool = False):
        """Execute full right-sizing pipeline."""
        
        recommendations = self.get_recommendations()
        print(f"Found {len(recommendations)} right-sizing opportunities")
        
        total_savings = 0
        for rec in recommendations:
            env = self._get_environment(rec['instance_id'])
            
            if auto_implement_nonprod and env in ['development', 'staging']:
                execution_id = self.auto_implement_nonprod(rec)
                print(f"Auto-implementing {rec['instance_id']}: {execution_id}")
            else:
                ticket_key = self.create_ticket(rec, env)
                print(f"Created ticket {ticket_key} for {rec['instance_id']}")
            
            total_savings += rec['estimated_monthly_savings']
        
        print(f"\nTotal potential monthly savings: ${total_savings:, .2f}")
    
    def _get_environment(self, instance_id: str) -> str:
                """Get environment from instance tags."""
        response = self.ec2.describe_instances(InstanceIds = [instance_id])
        for res in response['Reservations']:
            for inst in res['Instances']:
            for tag in inst.get('Tags', []):
            if tag['Key'] == 'environment':
            return tag['Value']
        return 'unknown'
 
 
# Execute pipeline
if __name__ == '__main__':
            pipeline = RightSizingPipeline()
    pipeline.run_pipeline(auto_implement_nonprod = True)