finops-expert

FinOps Expert

Expert guidance for cloud financial operations, cost optimization, resource management, and cloud economics.

Core Concepts

FinOps Fundamentals

• Cloud cost visibility
• Usage optimization
• Rate optimization
• Architecture optimization
• Cloud unit economics
• Showback and chargeback

Cost Management

• Reserved Instances (RIs)
• Savings Plans
• Spot instances
• Right-sizing resources
• Idle resource cleanup
• Storage lifecycle policies

FinOps Practices

• Tagging strategies
• Budgets and alerts
• Cost allocation
• Forecasting and planning
• Cross-team collaboration
• Continuous optimization

AWS Cost Analysis

import boto3
from datetime import datetime, timedelta
from typing import Dict, List
import pandas as pd

class AWSCostAnalyzer:
    """Analyze AWS costs using Cost Explorer API"""

    def __init__(self):
        self.ce_client = boto3.client('ce')

    def get_cost_and_usage(self, start_date: str, end_date: str,
                          granularity: str = 'DAILY',
                          metrics: List[str] = None) -> Dict:
        """Get cost and usage data"""
        if metrics is None:
            metrics = ['UnblendedCost', 'UsageQuantity']

        response = self.ce_client.get_cost_and_usage(
            TimePeriod={
                'Start': start_date,
                'End': end_date
            },
            Granularity=granularity,
            Metrics=metrics,
            GroupBy=[
                {'Type': 'DIMENSION', 'Key': 'SERVICE'}
            ]
        )

        return response['ResultsByTime']

    def get_top_services_by_cost(self, days: int = 30, top_n: int = 10) -> pd.DataFrame:
        """Get top services by cost"""
        end_date = datetime.now().strftime('%Y-%m-%d')
        start_date = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')

        results = self.get_cost_and_usage(start_date, end_date, 'MONTHLY')

        service_costs = {}
        for result in results:
            for group in result['Groups']:
                service = group['Keys'][0]
                cost = float(group['Metrics']['UnblendedCost']['Amount'])

                if service in service_costs:
                    service_costs[service] += cost
                else:
                    service_costs[service] = cost

        df = pd.DataFrame(list(service_costs.items()),
                         columns=['Service', 'Cost'])
        return df.nlargest(top_n, 'Cost')

    def get_cost_forecast(self, days_ahead: int = 30) -> Dict:
        """Get cost forecast"""
        start_date = datetime.now().strftime('%Y-%m-%d')
        end_date = (datetime.now() + timedelta(days=days_ahead)).strftime('%Y-%m-%d')

        response = self.ce_client.get_cost_forecast(
            TimePeriod={
                'Start': start_date,
                'End': end_date
            },
            Metric='UNBLENDED_COST',
            Granularity='MONTHLY'
        )

        return {
            'forecasted_cost': float(response['Total']['Amount']),
            'mean_value': float(response['ForecastResultsByTime'][0]['MeanValue'])
        }

    def get_rightsizing_recommendations(self) -> List[Dict]:
        """Get EC2 rightsizing recommendations"""
        response = self.ce_client.get_rightsizing_recommendation(
            Service='AmazonEC2'
        )

        recommendations = []
        for rec in response['RightsizingRecommendations']:
            recommendations.append({
                'instance_id': rec['CurrentInstance']['ResourceId'],
                'current_type': rec['CurrentInstance']['InstanceType'],
                'recommended_type': rec['ModifyRecommendationDetail']['TargetInstances'][0]['InstanceType']
                    if rec.get('ModifyRecommendationDetail') else None,
                'estimated_savings': float(rec['EstimatedMonthlySavings']['Value'])
                    if rec.get('EstimatedMonthlySavings') else 0
            })

        return recommendations

class CostOptimizer:
    """Optimize cloud costs"""

    def __init__(self):
        self.ec2_client = boto3.client('ec2')
        self.rds_client = boto3.client('rds')
        self.s3_client = boto3.client('s3')

    def find_idle_resources(self) -> Dict[str, List]:
        """Find idle/unused resources"""
        idle_resources = {
            'ec2_instances': [],
            'ebs_volumes': [],
            'elastic_ips': [],
            'load_balancers': []
        }

        # Idle EC2 instances (stopped for > 7 days)
        instances = self.ec2_client.describe_instances(
            Filters=[{'Name': 'instance-state-name', 'Values': ['stopped']}]
        )

        for reservation in instances['Reservations']:
            for instance in reservation['Instances']:
                idle_resources['ec2_instances'].append({
                    'id': instance['InstanceId'],
                    'type': instance['InstanceType'],
                    'state': instance['State']['Name']
                })

        # Unattached EBS volumes
        volumes = self.ec2_client.describe_volumes(
            Filters=[{'Name': 'status', 'Values': ['available']}]
        )

        for volume in volumes['Volumes']:
            idle_resources['ebs_volumes'].append({
                'id': volume['VolumeId'],
                'size': volume['Size'],
                'type': volume['VolumeType']
            })

        # Unattached Elastic IPs
        addresses = self.ec2_client.describe_addresses()

        for address in addresses['Addresses']:
            if 'InstanceId' not in address:
                idle_resources['elastic_ips'].append({
                    'allocation_id': address['AllocationId'],
                    'public_ip': address['PublicIp']
                })

        return idle_resources

    def calculate_reserved_instance_savings(self,
                                           instance_type: str,
                                           count: int,
                                           term: int = 1) -> Dict:
        """Calculate RI savings"""
        # Simplified calculation (would use actual pricing API)
        on_demand_hourly = self._get_on_demand_price(instance_type)
        ri_hourly = on_demand_hourly * 0.65  # ~35% discount

        hours_per_year = 24 * 365
        annual_on_demand = on_demand_hourly * hours_per_year * count
        annual_ri = ri_hourly * hours_per_year * count

        return {
            'instance_type': instance_type,
            'count': count,
            'annual_on_demand_cost': annual_on_demand,
            'annual_ri_cost': annual_ri,
            'annual_savings': annual_on_demand - annual_ri,
            'savings_percentage': ((annual_on_demand - annual_ri) / annual_on_demand) * 100
        }

    def _get_on_demand_price(self, instance_type: str) -> float:
        """Get on-demand hourly price (simplified)"""
        # In production, use AWS Pricing API
        prices = {
            't3.micro': 0.0104,
            't3.small': 0.0208,
            't3.medium': 0.0416,
            'm5.large': 0.096,
            'm5.xlarge': 0.192
        }
        return prices.get(instance_type, 0.10)

Cost Allocation and Tagging

class CostAllocation:
    """Manage cost allocation with tags"""

    def __init__(self):
        self.ec2_client = boto3.client('ec2')
        self.ce_client = boto3.client('ce')

    def define_tagging_strategy(self) -> Dict[str, List[str]]:
        """Define mandatory tags"""
        return {
            'environment': ['prod', 'staging', 'dev'],
            'team': ['engineering', 'data', 'product'],
            'cost_center': ['CC001', 'CC002', 'CC003'],
            'project': ['project-a', 'project-b'],
            'owner': ['email addresses']
        }

    def audit_resource_tags(self, resource_type: str = 'instance') -> List[Dict]:
        """Audit resources for missing tags"""
        mandatory_tags = ['environment', 'team', 'cost_center']
        untagged_resources = []

        if resource_type == 'instance':
            instances = self.ec2_client.describe_instances()

            for reservation in instances['Reservations']:
                for instance in reservation['Instances']:
                    tags = {tag['Key']: tag['Value']
                           for tag in instance.get('Tags', [])}

                    missing_tags = [tag for tag in mandatory_tags
                                  if tag not in tags]

                    if missing_tags:
                        untagged_resources.append({
                            'resource_id': instance['InstanceId'],
                            'missing_tags': missing_tags
                        })

        return untagged_resources

    def get_cost_by_tag(self, tag_key: str, start_date: str,
                       end_date: str) -> pd.DataFrame:
        """Get costs grouped by tag"""
        response = self.ce_client.get_cost_and_usage(
            TimePeriod={
                'Start': start_date,
                'End': end_date
            },
            Granularity='MONTHLY',
            Metrics=['UnblendedCost'],
            GroupBy=[
                {'Type': 'TAG', 'Key': tag_key}
            ]
        )

        costs = []
        for result in response['ResultsByTime']:
            for group in result['Groups']:
                costs.append({
                    'tag_value': group['Keys'][0].split('$')[1]
                        if '$' in group['Keys'][0] else 'Untagged',
                    'cost': float(group['Metrics']['UnblendedCost']['Amount'])
                })

        return pd.DataFrame(costs)

Budget Management

class BudgetManager:
    """Manage AWS budgets and alerts"""

    def __init__(self):
        self.budgets_client = boto3.client('budgets')
        self.account_id = boto3.client('sts').get_caller_identity()['Account']

    def create_monthly_budget(self, name: str, amount: float,
                             email: str) -> Dict:
        """Create monthly cost budget with alerts"""
        budget = {
            'BudgetName': name,
            'BudgetLimit': {
                'Amount': str(amount),
                'Unit': 'USD'
            },
            'TimeUnit': 'MONTHLY',
            'BudgetType': 'COST'
        }

        # Alert at 80% and 100%
        notifications = [
            {
                'Notification': {
                    'NotificationType': 'ACTUAL',
                    'ComparisonOperator': 'GREATER_THAN',
                    'Threshold': 80,
                    'ThresholdType': 'PERCENTAGE'
                },
                'Subscribers': [{
                    'SubscriptionType': 'EMAIL',
                    'Address': email
                }]
            },
            {
                'Notification': {
                    'NotificationType': 'ACTUAL',
                    'ComparisonOperator': 'GREATER_THAN',
                    'Threshold': 100,
                    'ThresholdType': 'PERCENTAGE'
                },
                'Subscribers': [{
                    'SubscriptionType': 'EMAIL',
                    'Address': email
                }]
            }
        ]

        response = self.budgets_client.create_budget(
            AccountId=self.account_id,
            Budget=budget,
            NotificationsWithSubscribers=notifications
        )

        return response

Best Practices

Cost Visibility

• Implement comprehensive tagging
• Enable Cost Explorer
• Set up cost allocation tags
• Create custom cost reports
• Use dashboards for visualization
• Monitor costs daily

Optimization

• Right-size resources regularly
• Use Reserved Instances/Savings Plans
• Leverage Spot instances for flexible workloads
• Implement auto-scaling
• Clean up idle resources
• Use storage lifecycle policies

Governance

• Set budgets and alerts
• Implement approval workflows
• Regular cost reviews
• Cross-team accountability
• Document cost optimization wins
• Automate cost controls

Anti-Patterns

❌ No tagging strategy ❌ Ignoring rightsizing recommendations ❌ Not using Reserved Instances ❌ No budget alerts ❌ Keeping idle resources ❌ Manual cost tracking ❌ Siloed cost responsibility

Resources

• AWS Cost Explorer: https://aws.amazon.com/aws-cost-management/aws-cost-explorer/
• FinOps Foundation: https://www.finops.org/
• AWS Well-Architected Cost Optimization: https://wa.aws.amazon.com/wat.pillar.costOptimization.en.html
• Boto3 Cost Explorer: https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ce.html