Tracking IOPS vs baseline storage spend in RDS #

import asyncio
import logging
import json
import os
from datetime import datetime, timedelta, timezone
from typing import Dict, List, Optional, Tuple

import aiobotocore
import boto3
from aiobotocore.session import get_session
from botocore.config import Config
from botocore.exceptions import ClientError, BotoCoreError

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("rds_iops_cost_tracker")

REGION = os.getenv("AWS_DEFAULT_REGION", "us-east-1")
MAX_CONCURRENCY = int(os.getenv("MAX_CONCURRENCY", "10"))
COST_EXPLORER_FALLBACK_CACHE = "/tmp/rds_cost_cache.json"

class CostAttributionError(Exception):
    """Raised when cost/metric correlation fails irrecoverably."""
    pass

def calculate_baseline_iops(storage_gb: int, storage_type: str, provisioned_iops: Optional[int]) -> int:
    """Derive baseline IOPS from RDS storage configuration."""
    if storage_type == "gp2":
        return min(storage_gb * 3, 16000)  # gp2 caps at 16k
    elif storage_type == "gp3":
        return provisioned_iops if provisioned_iops else 3000
    elif storage_type in ("io1", "io2"):
        return provisioned_iops or 0
    else:
        raise CostAttributionError(f"Unsupported storage type: {storage_type}")

async def fetch_rds_instances(session: aiobotocore.client.AioBaseClient) -> List[Dict]:
    """Retrieve RDS instance metadata with pagination handling."""
    instances = []
    paginator = session.get_paginator("describe_db_instances")
    async for page in paginator.paginate():
        for db in page.get("DBInstances", []):
            instances.append({
                "db_instance_id": db["DBInstanceIdentifier"],
                "storage_type": db.get("StorageType", "gp2"),
                "allocated_storage_gb": db["AllocatedStorage"],
                "provisioned_iops": db.get("Iops"),
                "engine": db["Engine"]
            })
    return instances

async def fetch_cloudwatch_iops(session: aiobotocore.client.AioBaseClient, instance_id: str) -> Dict[str, float]:
    """Pull 30-day average Read/Write IOPS from CloudWatch."""
    end_time = datetime.now(timezone.utc)
    start_time = end_time - timedelta(days=30)
    
    try:
        response = await session.get_metric_data(
            MetricDataQueries=[
                {
                    "Id": "read_ops",
                    "MetricStat": {
                        "Metric": {"Namespace": "AWS/RDS", "MetricName": "ReadIOPS", "Dimensions": [{"Name": "DBInstanceIdentifier", "Value": instance_id}]},
                        "Period": 86400,
                        "Stat": "Average"
                    }
                },
                {
                    "Id": "write_ops",
                    "MetricStat": {
                        "Metric": {"Namespace": "AWS/RDS", "MetricName": "WriteIOPS", "Dimensions": [{"Name": "DBInstanceIdentifier", "Value": instance_id}]},
                        "Period": 86400,
                        "Stat": "Average"
                    }
                }
            ],
            StartTime=start_time,
            EndTime=end_time
        )
        metrics = {m["Id"]: m.get("Values", [0.0])[0] for m in response.get("MetricDataResults", [])}
        return {"avg_read_iops": metrics.get("read_ops", 0.0), "avg_write_iops": metrics.get("write_ops", 0.0)}
    except ClientError as e:
        logger.warning(f"CloudWatch fetch failed for {instance_id}: {e.response['Error']['Code']}")
        return {"avg_read_iops": 0.0, "avg_write_iops": 0.0}

def get_cost_explorer_spend_sync(instance_id: str) -> Optional[float]:
    """Synchronous fallback for Cost Explorer queries."""
    try:
        ce = boto3.client("ce", region_name=REGION)
        response = ce.get_cost_and_usage(
            TimePeriod={"Start": (datetime.now(timezone.utc) - timedelta(days=30)).strftime("%Y-%m-%d"),
                        "End": datetime.now(timezone.utc).strftime("%Y-%m-%d")},
            Granularity="MONTHLY",
            Filter={"Dimensions": {"Key": "RESOURCE", "Values": [f"arn:aws:rds:{REGION}:*:db:{instance_id}"]}},
            Metrics=["UnblendedCost"]
        )
        return float(response["ResultsByTime"][0]["Total"]["UnblendedCost"]["Amount"])
    except Exception as e:
        logger.error(f"Cost Explorer fallback failed: {e}")
        if os.path.exists(COST_EXPLORER_FALLBACK_CACHE):
            with open(COST_EXPLORER_FALLBACK_CACHE, "r") as f:
                cache = json.load(f)
            return cache.get(instance_id)
        return None

async def correlate_instance_cost(session: aiobotocore.client.AioBaseClient, instance: Dict) -> Dict:
    """Execute async metric fetch and sync cost fallback."""
    db_id = instance["db_instance_id"]
    baseline = calculate_baseline_iops(
        instance["allocated_storage_gb"], 
        instance["storage_type"], 
        instance["provisioned_iops"]
    )
    telemetry = await fetch_cloudwatch_iops(session, db_id)
    total_avg_iops = telemetry["avg_read_iops"] + telemetry["avg_write_iops"]
    utilization_pct = (total_avg_iops / baseline * 100) if baseline > 0 else 0
    
    # Async cost fetch would typically go here; using sync fallback for resilience
    spend = get_cost_explorer_spend_sync(db_id)
    
    return {
        "db_instance_id": db_id,
        "storage_type": instance["storage_type"],
        "baseline_iops": baseline,
        "avg_observed_iops": round(total_avg_iops, 2),
        "utilization_pct": round(utilization_pct, 2),
        "estimated_30d_spend_usd": spend
    }

async def run_attribution_pipeline():
    """Orchestrate async concurrency with bounded semaphore."""
    config = Config(max_pool_connections=MAX_CONCURRENCY, retries={"max_attempts": 3, "mode": "adaptive"})
    session = get_session()
    async with session.create_client("rds", region_name=REGION, config=config) as rds_client, \
            session.create_client("cloudwatch", region_name=REGION, config=config) as cw_client:
        instances = await fetch_rds_instances(rds_client)
        logger.info(f"Discovered {len(instances)} RDS instances for attribution.")
        
        semaphore = asyncio.Semaphore(MAX_CONCURRENCY)
        
        async def bounded_correlate(inst):
            async with semaphore:
                return await correlate_instance_cost(cw_client, inst)
                
        tasks = [bounded_correlate(inst) for inst in instances]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        successful = [r for r in results if isinstance(r, dict)]
        logger.info(f"Successfully correlated {len(successful)} instances.")
        return successful

if __name__ == "__main__":
    asyncio.run(run_attribution_pipeline())