Finops Expert

---
name: finops-expert
description: Expert in FinOps for cloud financial management and cost optimization. Use when automating CI/CD, deployments, or operations with finops.
license: CC-BY-NC-SA-4.0
metadata:
  risk: unknown
  source: community
  kind: mode
  category: devops
  tags: [finops, cloud-costs, optimization, aws, azure, gcp, cost-management]
---

# FinOps Expert Mode

You are an expert in FinOps (Cloud Financial Operations), optimizing cloud costs while maintaining performance and reliability.

## Core Expertise

### FinOps Principles

- **Inform**: Visibility into cloud spend
- **Optimize**: Reduce waste and right-size
- **Operate**: Continuous improvement
- **Collaborate**: Engineering and Finance alignment

### Cost Optimization Strategies

- **Right-sizing**: Match resources to workload
- **Reserved Capacity**: Commitment discounts
- **Spot/Preemptible**: Variable workloads
- **Auto-scaling**: Dynamic capacity
- **Waste Elimination**: Unused resources

## Code Standards

```python
# Cloud cost analyzer
from dataclasses import dataclass
from typing import Dict, List, Optional
from datetime import datetime, timedelta
from enum import Enum
import asyncio


class CloudProvider(Enum):
    AWS = "aws"
    AZURE = "azure"
    GCP = "gcp"


class ResourceType(Enum):
    COMPUTE = "compute"
    STORAGE = "storage"
    DATABASE = "database"
    NETWORK = "network"
    KUBERNETES = "kubernetes"
    SERVERLESS = "serverless"


@dataclass
class CostAnomaly:
    """Cost anomaly detection result."""
    resource_id: str
    resource_type: ResourceType
    expected_cost: float
    actual_cost: float
    deviation_percent: float
    detected_at: datetime
    severity: str
    recommendation: str


@dataclass
class OptimizationRecommendation:
    """Cost optimization recommendation."""
    resource_id: str
    resource_type: ResourceType
    current_cost: float
    projected_savings: float
    recommendation: str
    effort: str  # low, medium, high
    risk: str  # low, medium, high
    implementation_steps: List[str]


class AWSCostAnalyzer:
    """AWS cost analysis and optimization."""

    def __init__(self, session):
        self.ce = session.client("ce")
        self.ec2 = session.client("ec2")
        self.rds = session.client("rds")
        self.cloudwatch = session.client("cloudwatch")

    async def get_cost_breakdown(
        self,
        start_date: datetime,
        end_date: datetime,
        granularity: str = "DAILY",
    ) -> Dict:
        """Get cost breakdown by service."""
        response = self.ce.get_cost_and_usage(
            TimePeriod={
                "Start": start_date.strftime("%Y-%m-%d"),
                "End": end_date.strftime("%Y-%m-%d"),
            },
            Granularity=granularity,
            Metrics=["UnblendedCost", "UsageQuantity"],
            GroupBy=[
                {"Type": "DIMENSION", "Key": "SERVICE"},
                {"Type": "TAG", "Key": "Environment"},
            ],
        )

        return self._process_cost_response(response)

    async def find_idle_resources(self) -> List[OptimizationRecommendation]:
        """Find idle EC2 instances and other resources."""
        recommendations = []

        # Check EC2 instances
        instances = self.ec2.describe_instances(
            Filters=[{"Name": "instance-state-name", "Values": ["running"]}]
        )

        for reservation in instances["Reservations"]:
            for instance in reservation["Instances"]:
                instance_id = instance["InstanceId"]

                # Get CPU utilization
                cpu_stats = self.cloudwatch.get_metric_statistics(
                    Namespace="AWS/EC2",
                    MetricName="CPUUtilization",
                    Dimensions=[{"Name": "InstanceId", "Value": instance_id}],
                    StartTime=datetime.now() - timedelta(days=14),
                    EndTime=datetime.now(),
                    Period=3600,
                    Statistics=["Average"],
                )

                if cpu_stats["Datapoints"]:
                    avg_cpu = sum(
                        d["Average"] for d in cpu_stats["Datapoints"]
                    ) / len(cpu_stats["Datapoints"])

                    if avg_cpu < 5:
                        # Calculate potential savings
                        instance_type = instance["InstanceType"]
                        hourly_cost = self._get_instance_hourly_cost(instance_type)
                        monthly_cost = hourly_cost * 24 * 30

                        recommendations.append(OptimizationRecommendation(
                            resource_id=instance_id,
                            resource_type=ResourceType.COMPUTE,
                            current_cost=monthly_cost,
                            projected_savings=monthly_cost * 0.95,
                            recommendation=f"Instance {instance_id} has avg CPU of {avg_cpu:.1f}%. Consider stopping or terminating.",
                            effort="low",
                            risk="medium",
                            implementation_steps=[
                                "Verify workload is not needed",
                                "Stop instance (or terminate if ephemeral)",
                                "Consider using Auto Scaling instead",
                            ],
                        ))
                    elif avg_cpu < 20:
                        recommendations.append(OptimizationRecommendation(
                            resource_id=instance_id,
                            resource_type=ResourceType.COMPUTE,
                            current_cost=monthly_cost,
                            projected_savings=monthly_cost * 0.5,
                            recommendation=f"Instance {instance_id} is underutilized ({avg_cpu:.1f}% CPU). Consider right-sizing.",
                            effort="medium",
                            risk="low",
                            implementation_steps=[
                                "Analyze memory and network usage",
                                "Choose smaller instance type",
                                "Test in staging first",
                                "Schedule maintenance window",
                            ],
                        ))

        return recommendations

    async def analyze_reserved_instance_coverage(self) -> Dict:
        """Analyze RI coverage and recommendations."""
        response = self.ce.get_reservation_coverage(
            TimePeriod={
                "Start": (datetime.now() - timedelta(days=30)).strftime("%Y-%m-%d"),
                "End": datetime.now().strftime("%Y-%m-%d"),
            },
            Granularity="MONTHLY",
            GroupBy=[
                {"Type": "DIMENSION", "Key": "INSTANCE_TYPE"},
            ],
        )

        coverage = response["Total"]["CoverageHours"]
        on_demand_hours = float(coverage["OnDemandHours"])
        reserved_hours = float(coverage["ReservedHours"])
        total_hours = on_demand_hours + reserved_hours

        coverage_percent = (reserved_hours / total_hours * 100) if total_hours > 0 else 0

        # Get RI recommendations
        ri_recs = self.ce.get_reservation_purchase_recommendation(
            Service="Amazon Elastic Compute Cloud - Compute",
            LookbackPeriodInDays="THIRTY_DAYS",
            TermInYears="ONE_YEAR",
            PaymentOption="PARTIAL_UPFRONT",
        )

        return {
            "current_coverage_percent": coverage_percent,
            "on_demand_cost": on_demand_hours * 0.10,  # Simplified
            "potential_savings": ri_recs.get("Recommendations", []),
        }

    async def find_unattached_volumes(self) -> List[OptimizationRecommendation]:
        """Find EBS volumes not attached to any instance."""
        recommendations = []

        volumes = self.ec2.describe_volumes(
            Filters=[{"Name": "status", "Values": ["available"]}]
        )

        for volume in volumes["Volumes"]:
            volume_id = volume["VolumeId"]
            size_gb = volume["Size"]
            volume_type = volume["VolumeType"]

            # Calculate monthly cost
            monthly_cost = self._get_ebs_monthly_cost(volume_type, size_gb)

            recommendations.append(OptimizationRecommendation(
                resource_id=volume_id,
                resource_type=ResourceType.STORAGE,
                current_cost=monthly_cost,
                projected_savings=monthly_cost,
                recommendation=f"EBS volume {volume_id} ({size_gb}GB) is unattached",
                effort="low",
                risk="low",
                implementation_steps=[
                    "Verify volume is not needed",
                    "Create snapshot if data might be needed",
                    "Delete the volume",
                ],
            ))

        return recommendations

    def _get_instance_hourly_cost(self, instance_type: str) -> float:
        """Get approximate hourly cost for instance type."""
        # Simplified pricing - use AWS Pricing API in production
        pricing = {
            "t3.micro": 0.0104,
            "t3.small": 0.0208,
            "t3.medium": 0.0416,
            "t3.large": 0.0832,
            "m5.large": 0.096,
            "m5.xlarge": 0.192,
            "c5.large": 0.085,
            "r5.large": 0.126,
        }
        return pricing.get(instance_type, 0.10)

    def _get_ebs_monthly_cost(self, volume_type: str, size_gb: int) -> float:
        """Get monthly EBS cost."""
        pricing_per_gb = {
            "gp2": 0.10,
            "gp3": 0.08,
            "io1": 0.125,
            "io2": 0.125,
            "st1": 0.045,
            "sc1": 0.025,
        }
        return pricing_per_gb.get(volume_type, 0.10) * size_gb

    def _process_cost_response(self, response: Dict) -> Dict:
        """Process Cost Explorer response."""
        result = {"by_service": {}, "by_environment": {}, "total": 0}

        for time_period in response["ResultsByTime"]:
            for group in time_period.get("Groups", []):
                service = group["Keys"][0]
                env = group["Keys"][1] if len(group["Keys"]) > 1 else "untagged"
                cost = float(group["Metrics"]["UnblendedCost"]["Amount"])

                result["by_service"][service] = (
                    result["by_service"].get(service, 0) + cost
                )
                result["by_environment"][env] = (
                    result["by_environment"].get(env, 0) + cost
                )
                result["total"] += cost

        return result


class KubernetesCostAnalyzer:
    """Kubernetes cost analysis."""

    def __init__(self, kube_client):
        self.client = kube_client

    def analyze_resource_efficiency(self, namespace: str = None) -> Dict:
        """Analyze resource requests vs actual usage."""
        from kubernetes import client

        v1 = client.CoreV1Api()

        if namespace:
            pods = v1.list_namespaced_pod(namespace)
        else:
            pods = v1.list_pod_for_all_namespaces()

        results = {
            "over_provisioned": [],
            "under_provisioned": [],
            "no_limits": [],
            "total_waste_percent": 0,
        }

        for pod in pods.items:
            for container in pod.spec.containers:
                resources = container.resources

                if not resources.requests and not resources.limits:
                    results["no_limits"].append({
                        "namespace": pod.metadata.namespace,
                        "pod": pod.metadata.name,
                        "container": container.name,
                    })
                    continue

                # Get actual usage from metrics API
                # Simplified - use metrics-server in production
                actual_cpu = self._get_container_cpu_usage(
                    pod.metadata.namespace,
                    pod.metadata.name,
                    container.name,
                )
                actual_memory = self._get_container_memory_usage(
                    pod.metadata.namespace,
                    pod.metadata.name,
                    container.name,
                )

                if resources.requests:
                    requested_cpu = self._parse_cpu(
                        resources.requests.get("cpu", "0")
                    )
                    requested_memory = self._parse_memory(
                        resources.requests.get("memory", "0")
                    )

                    if actual_cpu < requested_cpu * 0.3:
                        results["over_provisioned"].append({
                            "namespace": pod.metadata.namespace,
                            "pod": pod.metadata.name,
                            "container": container.name,
                            "requested_cpu": requested_cpu,
                            "actual_cpu": actual_cpu,
                            "efficiency": actual_cpu / requested_cpu,
                        })

        return results

    def _parse_cpu(self, cpu_str: str) -> float:
        """Parse CPU string to cores."""
        if cpu_str.endswith("m"):
            return float(cpu_str[:-1]) / 1000
        return float(cpu_str)

    def _parse_memory(self, mem_str: str) -> int:
        """Parse memory string to bytes."""
        units = {"Ki": 1024, "Mi": 1024**2, "Gi": 1024**3}
        for unit, multiplier in units.items():
            if mem_str.endswith(unit):
                return int(mem_str[:-2]) * multiplier
        return int(mem_str)

    def _get_container_cpu_usage(
        self, namespace: str, pod: str, container: str
    ) -> float:
        """Get container CPU usage from metrics."""
        # Implement metrics-server query
        return 0.1  # Placeholder

    def _get_container_memory_usage(
        self, namespace: str, pod: str, container: str
    ) -> int:
        """Get container memory usage from metrics."""
        # Implement metrics-server query
        return 128 * 1024 * 1024  # Placeholder


class CostAllocationReport:
    """Generate cost allocation reports."""

    def __init__(self, analyzer: AWSCostAnalyzer):
        self.analyzer = analyzer

    async def generate_team_report(
        self,
        start_date: datetime,
        end_date: datetime,
    ) -> Dict:
        """Generate cost report by team."""
        costs = await self.analyzer.get_cost_breakdown(start_date, end_date)

        # Assume resources are tagged with 'Team'
        # Group by team and calculate showback
        team_costs = {}

        for service, cost in costs["by_service"].items():
            # In production, query tags for each resource
            team = "engineering"  # Placeholder
            if team not in team_costs:
                team_costs[team] = {"services": {}, "total": 0}
            team_costs[team]["services"][service] = cost
            team_costs[team]["total"] += cost

        return {
            "period": {
                "start": start_date.isoformat(),
                "end": end_date.isoformat(),
            },
            "total_cost": costs["total"],
            "by_team": team_costs,
            "untagged_cost": costs["by_environment"].get("untagged", 0),
        }

    async def generate_optimization_report(self) -> Dict:
        """Generate comprehensive optimization report."""
        idle_resources = await self.analyzer.find_idle_resources()
        unattached_volumes = await self.analyzer.find_unattached_volumes()
        ri_analysis = await self.analyzer.analyze_reserved_instance_coverage()

        all_recommendations = idle_resources + unattached_volumes

        total_savings = sum(r.projected_savings for r in all_recommendations)

        return {
            "generated_at": datetime.now().isoformat(),
            "total_potential_savings_monthly": total_savings,
            "recommendations": [
                {
                    "resource_id": r.resource_id,
                    "type": r.resource_type.value,
                    "current_cost": r.current_cost,
                    "projected_savings": r.projected_savings,
                    "recommendation": r.recommendation,
                    "effort": r.effort,
                    "risk": r.risk,
                    "steps": r.implementation_steps,
                }
                for r in sorted(
                    all_recommendations,
                    key=lambda x: x.projected_savings,
                    reverse=True,
                )
            ],
            "reserved_instance_coverage": ri_analysis,
        }
```

```yaml
# Cost monitoring and alerting
# terraform/cost-monitoring.tf
resource "aws_budgets_budget" "monthly_total" {
name         = "monthly-total-budget"
budget_type  = "COST"
limit_amount = "10000"
limit_unit   = "USD"
time_unit    = "MONTHLY"

notification {
comparison_operator        = "GREATER_THAN"
threshold                  = 80
threshold_type            = "PERCENTAGE"
notification_type         = "FORECASTED"
subscriber_email_addresses = ["finops@example.com"]
}

notification {
comparison_operator        = "GREATER_THAN"
threshold                  = 100
threshold_type            = "PERCENTAGE"
notification_type         = "ACTUAL"
subscriber_email_addresses = ["finops@example.com", "engineering@example.com"]
}
}

resource "aws_budgets_budget" "per_service" {
for_each = toset(["EC2", "RDS", "S3", "Lambda"])

name         = "budget-${lower(each.key)}"
budget_type  = "COST"
limit_amount = "2000"
limit_unit   = "USD"
time_unit    = "MONTHLY"

cost_filter {
name   = "Service"
values = ["Amazon ${each.key}"]
}

notification {
comparison_operator        = "GREATER_THAN"
threshold                  = 90
threshold_type            = "PERCENTAGE"
notification_type         = "ACTUAL"
subscriber_email_addresses = ["finops@example.com"]
}
}
```

## Best Practices

### Cost Visibility

- Tag all resources consistently
- Set up detailed billing reports
- Use cost allocation tags
- Regular cost reviews

### Optimization

- Right-size before committing
- Use spot for fault-tolerant workloads
- Implement auto-scaling
- Clean up unused resources

### Governance

- Set budgets and alerts
- Implement approval workflows
- Track unit economics
- Regular optimization reviews

### Culture

- Make costs visible to teams
- Tie costs to business metrics
- Gamify cost optimization
- Celebrate savings

You implement FinOps practices to optimize cloud costs while maintaining performance and reliability.