# 05 · FinOps Cost Anomaly Hunter

**Format:** AWS CDK (TypeScript) · **Est. cost:** < $2/month (Lambda, S3). Cost Anomaly Detection and one budget are free.

Four layers that turn the AWS bill into action:

| Layer | Service | Result |
|---|---|---|
| Detect spikes | Cost Anomaly Detection (ML, per service) | Immediate alert when impact ≥ threshold |
| Guard the total | AWS Budgets | Alerts at 80% actual and 100% forecast |
| Explain | CUR 2.0 Data Export (Parquet, resource-level) | Query every line item with Athena |
| Fix | Weekly "waste hunter" Lambda | Finds unattached EBS, gp2, idle EIPs, old snapshots, idle EC2, Graviton candidates; estimates $/month; optional auto-fix |

## Deploy
```bash
npm install
npx cdk deploy -c alertEmail=you@example.com -c monthlyBudgetUsd=500 -c anomalyThresholdUsd=50
# Optional: auto-apply safe fixes (gp2->gp3, release idle EIPs)
npx cdk deploy -c alertEmail=you@example.com -c autoFix=true -c scanRegions=us-east-1,us-west-2,eu-west-1
```
Deploys to **us-east-1** (billing services live there); the hunter scans any regions you list.

## Run a hunt now
```bash
aws lambda invoke --function-name <WasteHunterFunction> out.json && cat out.json
aws s3 cp s3://<ReportBucketName>/reports/$(date +%F)/findings.csv - | column -s, -t | head
```

## Query CUR 2.0 with Athena
CUR files arrive within 24 h. Create a Glue crawler on `s3://<CurBucketName>/reports/`, then:
```sql
SELECT line_item_product_code AS service,
       round(sum(line_item_unblended_cost), 2) AS cost
FROM   cur2
WHERE  line_item_usage_start_date >= date_add('day', -30, current_date)
GROUP  BY 1 ORDER BY 2 DESC LIMIT 15;
```

## Show the impact
Track "monthly waste found" vs "monthly waste removed" each week. That chart is your FinOps story.

## Clean up
```bash
npx cdk destroy
```

#!/usr/bin/env node
import * as cdk from 'aws-cdk-lib';
import { FinOpsStack } from '../lib/finops-stack';

const app = new cdk.App();
new FinOpsStack(app, 'FinOpsHunter', {
  // Cost Explorer, Budgets and Data Exports are global services homed in us-east-1.
  env: { account: process.env.CDK_DEFAULT_ACCOUNT, region: 'us-east-1' },
  alertEmail: app.node.tryGetContext('alertEmail'),
  monthlyBudgetUsd: Number(app.node.tryGetContext('monthlyBudgetUsd') ?? 500),
  anomalyThresholdUsd: Number(app.node.tryGetContext('anomalyThresholdUsd') ?? 50),
  autoFix: app.node.tryGetContext('autoFix') === 'true',
  scanRegions: (app.node.tryGetContext('scanRegions') ?? 'us-east-1,us-west-2').split(','),
});

import * as path from 'path';
import * as cdk from 'aws-cdk-lib';
import { Construct } from 'constructs';
import * as bcm from 'aws-cdk-lib/aws-bcmdataexports';
import * as budgets from 'aws-cdk-lib/aws-budgets';
import * as ce from 'aws-cdk-lib/aws-ce';
import * as events from 'aws-cdk-lib/aws-events';
import * as targets from 'aws-cdk-lib/aws-events-targets';
import * as iam from 'aws-cdk-lib/aws-iam';
import * as lambda from 'aws-cdk-lib/aws-lambda';
import * as logs from 'aws-cdk-lib/aws-logs';
import * as s3 from 'aws-cdk-lib/aws-s3';
import * as sns from 'aws-cdk-lib/aws-sns';
import * as subs from 'aws-cdk-lib/aws-sns-subscriptions';

export interface FinOpsStackProps extends cdk.StackProps {
  alertEmail?: string;
  monthlyBudgetUsd: number;
  anomalyThresholdUsd: number;
  /** When true the janitor converts gp2->gp3 and releases idle Elastic IPs. Everything else is report-only. */
  autoFix: boolean;
  scanRegions: string[];
}

export class FinOpsStack extends cdk.Stack {
  constructor(scope: Construct, id: string, props: FinOpsStackProps) {
    super(scope, id, props);

    const topic = new sns.Topic(this, 'FinOpsTopic', { displayName: 'FinOps alerts' });
    topic.addToResourcePolicy(
      new iam.PolicyStatement({
        sid: 'AllowCostAnomalyDetection',
        principals: [new iam.ServicePrincipal('costalerts.amazonaws.com')],
        actions: ['sns:Publish'],
        resources: [topic.topicArn],
      }),
    );
    topic.addToResourcePolicy(
      new iam.PolicyStatement({
        sid: 'AllowBudgets',
        principals: [new iam.ServicePrincipal('budgets.amazonaws.com')],
        actions: ['sns:Publish'],
        resources: [topic.topicArn],
      }),
    );
    if (props.alertEmail) topic.addSubscription(new subs.EmailSubscription(props.alertEmail));

    // ---------------------------------------------------------------- 1. detect spikes (ML)
    const monitor = new ce.CfnAnomalyMonitor(this, 'ServiceMonitor', {
      monitorName: 'finops-by-service',
      monitorType: 'DIMENSIONAL',
      monitorDimension: 'SERVICE',
    });
    new ce.CfnAnomalySubscription(this, 'AnomalySubscription', {
      subscriptionName: 'finops-immediate',
      frequency: 'IMMEDIATE',
      monitorArnList: [monitor.attrMonitorArn],
      subscribers: [{ type: 'SNS', address: topic.topicArn }],
      thresholdExpression: JSON.stringify({
        Dimensions: {
          Key: 'ANOMALY_TOTAL_IMPACT_ABSOLUTE',
          MatchOptions: ['GREATER_THAN_OR_EQUAL'],
          Values: [String(props.anomalyThresholdUsd)],
        },
      }),
    });

    // ---------------------------------------------------------------- 2. guard the total
    new budgets.CfnBudget(this, 'MonthlyBudget', {
      budget: {
        budgetName: 'finops-monthly-total',
        budgetType: 'COST',
        timeUnit: 'MONTHLY',
        budgetLimit: { amount: props.monthlyBudgetUsd, unit: 'USD' },
      },
      notificationsWithSubscribers: [
        {
          notification: { notificationType: 'FORECASTED', comparisonOperator: 'GREATER_THAN', threshold: 100 },
          subscribers: [{ subscriptionType: 'SNS', address: topic.topicArn }],
        },
        {
          notification: { notificationType: 'ACTUAL', comparisonOperator: 'GREATER_THAN', threshold: 80 },
          subscribers: [{ subscriptionType: 'SNS', address: topic.topicArn }],
        },
      ],
    });

    // ---------------------------------------------------------------- 3. CUR 2.0 for Athena
    const curBucket = new s3.Bucket(this, 'CurBucket', {
      encryption: s3.BucketEncryption.S3_MANAGED,
      blockPublicAccess: s3.BlockPublicAccess.BLOCK_ALL,
      enforceSSL: true,
      lifecycleRules: [{ prefix: 'reports/', expiration: cdk.Duration.days(400) }],
    });
    curBucket.addToResourcePolicy(
      new iam.PolicyStatement({
        sid: 'AllowDataExports',
        principals: [
          new iam.ServicePrincipal('bcm-data-exports.amazonaws.com'),
          new iam.ServicePrincipal('billingreports.amazonaws.com'),
        ],
        actions: ['s3:PutObject', 's3:GetBucketPolicy'],
        resources: [curBucket.bucketArn, curBucket.arnForObjects('*')],
        conditions: {
          StringLike: {
            'aws:SourceAccount': this.account,
            'aws:SourceArn': [
              `arn:${this.partition}:cur:us-east-1:${this.account}:definition/*`,
              `arn:${this.partition}:bcm-data-exports:us-east-1:${this.account}:export/*`,
            ],
          },
        },
      }),
    );
    const curExport = new bcm.CfnExport(this, 'Cur2Export', {
      export: {
        name: 'finops-cur2',
        description: 'CUR 2.0 daily, resource-level, Parquet - query with Athena',
        dataQuery: {
          queryStatement:
            'SELECT bill_billing_period_start_date, line_item_usage_start_date, line_item_usage_account_id, ' +
            'line_item_product_code, line_item_resource_id, line_item_usage_type, line_item_line_item_type, ' +
            'line_item_unblended_cost, line_item_net_unblended_cost, savings_plan_savings_plan_effective_cost, ' +
            'reservation_effective_cost, product_region_code, resource_tags FROM COST_AND_USAGE_REPORT',
          tableConfigurations: {
            COST_AND_USAGE_REPORT: {
              TIME_GRANULARITY: 'DAILY',
              INCLUDE_RESOURCES: 'TRUE',
              INCLUDE_SPLIT_COST_ALLOCATION_DATA: 'FALSE',
              INCLUDE_MANUAL_DISCOUNT_COMPATIBILITY: 'FALSE',
            },
          },
        },
        destinationConfigurations: {
          s3Destination: {
            s3Bucket: curBucket.bucketName,
            s3Prefix: 'reports',
            s3Region: 'us-east-1',
            s3OutputConfigurations: {
              compression: 'PARQUET',
              format: 'PARQUET',
              outputType: 'CUSTOM',
              overwrite: 'OVERWRITE_REPORT',
            },
          },
        },
        refreshCadence: { frequency: 'SYNCHRONOUS' },
      },
    });
    curExport.node.addDependency(curBucket.policy!);

    // ---------------------------------------------------------------- 4. find + fix waste
    const reportBucket = new s3.Bucket(this, 'ReportBucket', {
      encryption: s3.BucketEncryption.S3_MANAGED,
      blockPublicAccess: s3.BlockPublicAccess.BLOCK_ALL,
      enforceSSL: true,
      lifecycleRules: [{ expiration: cdk.Duration.days(365) }],
    });

    const janitor = new lambda.Function(this, 'WasteHunter', {
      runtime: lambda.Runtime.PYTHON_3_12,
      architecture: lambda.Architecture.ARM_64,
      handler: 'waste_hunter.handler',
      code: lambda.Code.fromAsset(path.join(__dirname, '..', 'lambda')),
      timeout: cdk.Duration.minutes(10),
      memorySize: 512,
      environment: {
        TOPIC_ARN: topic.topicArn,
        REPORT_BUCKET: reportBucket.bucketName,
        AUTO_FIX: String(props.autoFix),
        SCAN_REGIONS: props.scanRegions.join(','),
        IDLE_CPU_PCT: '5',
        SNAPSHOT_MAX_AGE_DAYS: '90',
      },
      logGroup: new logs.LogGroup(this, 'WasteHunterLogs', {
        retention: logs.RetentionDays.ONE_MONTH,
        removalPolicy: cdk.RemovalPolicy.DESTROY,
      }),
    });
    topic.grantPublish(janitor);
    reportBucket.grantPut(janitor);
    janitor.addToRolePolicy(
      new iam.PolicyStatement({
        actions: [
          'ec2:DescribeVolumes',
          'ec2:DescribeAddresses',
          'ec2:DescribeSnapshots',
          'ec2:DescribeImages',
          'ec2:DescribeInstances',
          'cloudwatch:GetMetricStatistics',
          'ce:GetCostAndUsage',
          'ce:GetSavingsPlansCoverage',
        ],
        resources: ['*'],
      }),
    );
    if (props.autoFix) {
      janitor.addToRolePolicy(
        new iam.PolicyStatement({ actions: ['ec2:ModifyVolume', 'ec2:ReleaseAddress'], resources: ['*'] }),
      );
    }

    new events.Rule(this, 'WeeklyHunt', {
      description: 'Weekly waste hunt (Mondays 13:00 UTC)',
      schedule: events.Schedule.cron({ weekDay: 'MON', hour: '13', minute: '0' }),
      targets: [new targets.LambdaFunction(janitor)],
    });

    new cdk.CfnOutput(this, 'TopicArn', { value: topic.topicArn });
    new cdk.CfnOutput(this, 'CurBucketName', { value: curBucket.bucketName });
    new cdk.CfnOutput(this, 'ReportBucketName', { value: reportBucket.bucketName });
    new cdk.CfnOutput(this, 'WasteHunterFunction', { value: janitor.functionName });
  }
}

"""Weekly waste hunter.

Finds money leaks across regions, estimates monthly savings, writes a JSON + CSV
report to S3 and sends a summary to SNS. With AUTO_FIX=true it applies the two
safe, reversible fixes: gp2 -> gp3 conversion and releasing idle Elastic IPs.
"""
import csv
import datetime as dt
import io
import json
import os

import boto3

TOPIC_ARN = os.environ["TOPIC_ARN"]
REPORT_BUCKET = os.environ["REPORT_BUCKET"]
AUTO_FIX = os.environ.get("AUTO_FIX", "false").lower() == "true"
REGIONS = [r.strip() for r in os.environ.get("SCAN_REGIONS", "us-east-1").split(",") if r.strip()]
IDLE_CPU = float(os.environ.get("IDLE_CPU_PCT", "5"))
SNAP_AGE = int(os.environ.get("SNAPSHOT_MAX_AGE_DAYS", "90"))

# Approximate us-east-1 on-demand list prices (USD). Good enough for prioritisation.
PRICE = {
    "gp2_gb": 0.10, "gp3_gb": 0.08, "io1_gb": 0.125, "st1_gb": 0.045, "sc1_gb": 0.015, "standard_gb": 0.05,
    "snapshot_gb": 0.05, "eip_hour": 0.005,
}
GRAVITON = {"m5": "m7g", "m6i": "m7g", "c5": "c7g", "c6i": "c7g", "r5": "r7g", "r6i": "r7g", "t3": "t4g", "t2": "t4g"}

sns = boto3.client("sns")
s3 = boto3.client("s3")
NOW = dt.datetime.now(dt.timezone.utc)


def owner(tags):
    tags = {t["Key"].lower(): t["Value"] for t in (tags or [])}
    return tags.get("owner") or tags.get("team") or tags.get("created-by") or "untagged"


def hunt_region(region):
    ec2 = boto3.client("ec2", region_name=region)
    cw = boto3.client("cloudwatch", region_name=region)
    findings = []

    # 1. Unattached EBS volumes
    for page in ec2.get_paginator("describe_volumes").paginate():
        for v in page["Volumes"]:
            gb_price = PRICE.get(f"{v['VolumeType']}_gb", 0.10)
            if v["State"] == "available":
                findings.append({
                    "region": region, "type": "unattached-ebs", "resource": v["VolumeId"],
                    "owner": owner(v.get("Tags")), "monthly_usd": round(v["Size"] * gb_price, 2),
                    "action": "Snapshot then delete", "auto_fixed": False,
                })
            elif v["VolumeType"] == "gp2":
                saving = round(v["Size"] * (PRICE["gp2_gb"] - PRICE["gp3_gb"]), 2)
                fixed = False
                if AUTO_FIX:
                    ec2.modify_volume(VolumeId=v["VolumeId"], VolumeType="gp3")
                    fixed = True
                findings.append({
                    "region": region, "type": "gp2-to-gp3", "resource": v["VolumeId"],
                    "owner": owner(v.get("Tags")), "monthly_usd": saving,
                    "action": "Convert to gp3 (online, no downtime)", "auto_fixed": fixed,
                })

    # 2. Idle Elastic IPs
    for a in ec2.describe_addresses()["Addresses"]:
        if "AssociationId" not in a:
            fixed = False
            if AUTO_FIX and a.get("AllocationId"):
                ec2.release_address(AllocationId=a["AllocationId"])
                fixed = True
            findings.append({
                "region": region, "type": "idle-eip", "resource": a.get("PublicIp"),
                "owner": owner(a.get("Tags")), "monthly_usd": round(PRICE["eip_hour"] * 730, 2),
                "action": "Release", "auto_fixed": fixed,
            })

    # 3. Old snapshots not backing an AMI
    ami_snaps = set()
    for img in ec2.describe_images(Owners=["self"])["Images"]:
        for bdm in img.get("BlockDeviceMappings", []):
            if "Ebs" in bdm and "SnapshotId" in bdm["Ebs"]:
                ami_snaps.add(bdm["Ebs"]["SnapshotId"])
    cutoff = NOW - dt.timedelta(days=SNAP_AGE)
    for page in ec2.get_paginator("describe_snapshots").paginate(OwnerIds=["self"]):
        for s in page["Snapshots"]:
            if s["StartTime"] < cutoff and s["SnapshotId"] not in ami_snaps:
                findings.append({
                    "region": region, "type": "old-snapshot", "resource": s["SnapshotId"],
                    "owner": owner(s.get("Tags")), "monthly_usd": round(s["VolumeSize"] * PRICE["snapshot_gb"], 2),
                    "action": f"Older than {SNAP_AGE} days - archive or delete", "auto_fixed": False,
                })

    # 4. Idle instances + Graviton candidates
    for page in ec2.get_paginator("describe_instances").paginate(
        Filters=[{"Name": "instance-state-name", "Values": ["running"]}]
    ):
        for res in page["Reservations"]:
            for i in res["Instances"]:
                iid, itype = i["InstanceId"], i["InstanceType"]
                stats = cw.get_metric_statistics(
                    Namespace="AWS/EC2", MetricName="CPUUtilization",
                    Dimensions=[{"Name": "InstanceId", "Value": iid}],
                    StartTime=NOW - dt.timedelta(days=14), EndTime=NOW, Period=86400, Statistics=["Maximum"],
                )["Datapoints"]
                peak = max((d["Maximum"] for d in stats), default=None)
                if peak is not None and peak < IDLE_CPU:
                    findings.append({
                        "region": region, "type": "idle-instance", "resource": iid, "owner": owner(i.get("Tags")),
                        "monthly_usd": None, "action": f"{itype} peaked at {peak:.1f}% CPU in 14 days - stop or downsize",
                        "auto_fixed": False,
                    })
                family = itype.split(".")[0]
                if family in GRAVITON:
                    findings.append({
                        "region": region, "type": "graviton-candidate", "resource": iid, "owner": owner(i.get("Tags")),
                        "monthly_usd": None,
                        "action": f"{itype} -> {GRAVITON[family]}.{itype.split('.')[1]} (~20% cheaper, often faster)",
                        "auto_fixed": False,
                    })
    return findings


def handler(_event, _context):
    findings = []
    for region in REGIONS:
        try:
            findings += hunt_region(region)
        except Exception as exc:  # keep hunting other regions
            findings.append({"region": region, "type": "error", "resource": "-", "owner": "-",
                             "monthly_usd": None, "action": str(exc), "auto_fixed": False})

    total = round(sum(f["monthly_usd"] or 0 for f in findings), 2)
    by_type = {}
    for f in findings:
        by_type.setdefault(f["type"], {"count": 0, "monthly_usd": 0.0})
        by_type[f["type"]]["count"] += 1
        by_type[f["type"]]["monthly_usd"] = round(by_type[f["type"]]["monthly_usd"] + (f["monthly_usd"] or 0), 2)

    stamp = NOW.strftime("%Y-%m-%d")
    s3.put_object(Bucket=REPORT_BUCKET, Key=f"reports/{stamp}/findings.json",
                  Body=json.dumps({"total_monthly_usd": total, "by_type": by_type, "findings": findings}, indent=2))
    buf = io.StringIO()
    writer = csv.DictWriter(buf, fieldnames=["region", "type", "resource", "owner", "monthly_usd", "action", "auto_fixed"])
    writer.writeheader()
    writer.writerows(findings)
    s3.put_object(Bucket=REPORT_BUCKET, Key=f"reports/{stamp}/findings.csv", Body=buf.getvalue())

    lines = [f"Estimated waste: ${total}/month ({len(findings)} findings, auto-fix={'on' if AUTO_FIX else 'off'})", ""]
    lines += [f"- {k}: {v['count']} item(s), ~${v['monthly_usd']}/month" for k, v in sorted(by_type.items())]
    lines += ["", f"Full report: s3://{REPORT_BUCKET}/reports/{stamp}/findings.csv"]
    sns.publish(TopicArn=TOPIC_ARN, Subject=f"FinOps weekly hunt: ~${total}/month to save"[:100], Message="\n".join(lines))
    return {"total_monthly_usd": total, "by_type": by_type}

{
  "app": "npx ts-node --prefer-ts-exts bin/app.ts",
  "watch": { "include": ["**"], "exclude": ["README.md", "cdk*.json", "**/*.d.ts", "**/*.js", "node_modules", "cdk.out"] },
  "context": {}
}

{
  "name": "05-finops-cost-anomaly-hunter",
  "version": "1.0.0",
  "private": true,
  "bin": { "app": "bin/app.js" },
  "scripts": {
    "build": "tsc",
    "synth": "cdk synth",
    "deploy": "cdk deploy --all",
    "destroy": "cdk destroy --all"
  },
  "devDependencies": {
    "@types/node": "^22.0.0",
    "aws-cdk": "^2.1000.0",
    "ts-node": "^10.9.2",
    "typescript": "~5.6.3"
  },
  "dependencies": {
    "aws-cdk-lib": "^2.200.0",
    "constructs": "^10.4.2"
  }
}

{
  "compilerOptions": {
    "target": "ES2022",
    "module": "commonjs",
    "lib": ["es2022"],
    "strict": true,
    "esModuleInterop": true,
    "skipLibCheck": true,
    "noEmit": true,
    "types": ["node"]
  },
  "exclude": ["node_modules", "cdk.out"]
}