AWS Idle Resource Reaper with Cost Estimate and Approval

id: aws-idle-resource-reaper
namespace: company.team
description: |
  Find idle AWS resources that keep billing you: unattached EBS volumes,
  unassociated Elastic IPs, old snapshots that no AMI uses, and long-stopped
  instances. Estimate the monthly waste, report it to Slack, and, when you turn
  off dry run and approve, back up and delete the safe-to-remove resources.

triggers:
  - id: weekly_scan
    type: io.kestra.plugin.core.trigger.Schedule
    description: Scan every Monday morning. Scheduled runs use the input defaults,
      so they only report unless you change `dry_run`.
    cron: "0 8 * * 1"
    disabled: true

inputs:
  - id: region
    type: STRING
    displayName: AWS region
    defaults: us-east-1
  - id: dry_run
    type: BOOL
    displayName: Dry run
    description: When true, only report. Set to false to delete after approval.
    defaults: true
  - id: volume_min_age_days
    type: INT
    displayName: Minimum volume age (days)
    description: Unattached volumes younger than this are ignored, so a volume
      between detach and re-attach is never touched.
    defaults: 7
  - id: snapshot_min_age_days
    type: INT
    displayName: Minimum snapshot age (days)
    defaults: 90
  - id: stopped_min_days
    type: INT
    displayName: Minimum stopped time (days)
    description: Instances stopped for longer than this are reported. They are never
      terminated automatically.
    defaults: 30
  - id: keep_tag_key
    type: STRING
    displayName: Keep tag
    description: Resources carrying a tag with this key are always skipped.
    defaults: keep
  - id: snapshot_before_delete
    type: BOOL
    displayName: Snapshot volumes before deleting
    description: Take a tagged backup snapshot of each volume before it is deleted.
    defaults: true
  - id: endpoint_url
    type: STRING
    displayName: Endpoint URL (optional)
    description: Leave empty for real AWS. Set it to try the flow against an
      emulator such as Moto, for example http://host.docker.internal:5050.
    required: false

variables:
  # Monthly list prices in USD for us-east-1. Adjust them for your region and pricing.
  prices:
    ebs_gb_month:
      gp3: 0.08
      gp2: 0.10
      io1: 0.125
      io2: 0.125
      st1: 0.045
      sc1: 0.015
      standard: 0.05
    snapshot_gb_month: 0.05
    public_ipv4_month: 3.65

tasks:
  - id: scan
    type: io.kestra.plugin.aws.cli.AwsCLI
    description: Read volumes, Elastic IPs, snapshots, AMIs and stopped instances in
      one read-only pass. Stops at the first error, for example invalid
      credentials.
    accessKeyId: "{{ secret('AWS_ACCESS_KEY_ID') }}"
    secretKeyId: "{{ secret('AWS_SECRET_ACCESS_KEY') }}"
    region: "{{ inputs.region }}"
    env:
      ENDPOINT: "{{ inputs.endpoint_url ?? '' }}"
    outputFiles:
      - "*.json"
    commands:
      - set -e
      - EP=${ENDPOINT:+--endpoint-url $ENDPOINT}
      - aws $EP sts get-caller-identity --output json > identity.json
      - aws $EP ec2 describe-volumes --filters Name=status,Values=available
        --output json > volumes.json
      - aws $EP ec2 describe-addresses --output json > addresses.json
      - aws $EP ec2 describe-snapshots --owner-ids self --output json >
        snapshots.json
      - aws $EP ec2 describe-images --owners self --output json > images.json
      - aws $EP ec2 describe-instances --filters
        Name=instance-state-name,Values=stopped --output json > instances.json
      - aws $EP ec2 describe-volumes --output json > all_volumes.json

  - id: analyze
    type: io.kestra.plugin.scripts.python.Script
    description: Apply the rules, skip anything tagged with the keep tag, estimate
      monthly cost, and write a CSV report.
    dependencies:
      - kestra
    inputFiles:
      identity.json: "{{ read(outputs.scan.outputFiles['identity.json']) }}"
      volumes.json: "{{ read(outputs.scan.outputFiles['volumes.json']) }}"
      addresses.json: "{{ read(outputs.scan.outputFiles['addresses.json']) }}"
      snapshots.json: "{{ read(outputs.scan.outputFiles['snapshots.json']) }}"
      images.json: "{{ read(outputs.scan.outputFiles['images.json']) }}"
      instances.json: "{{ read(outputs.scan.outputFiles['instances.json']) }}"
      all_volumes.json: "{{ read(outputs.scan.outputFiles['all_volumes.json']) }}"
      prices.json: "{{ vars.prices | toJson }}"
    outputFiles:
      - idle_resources.csv
    script: |
      import csv
      import json
      import re
      from datetime import datetime, timezone
      from kestra import Kestra

      def load(name):
          with open(name) as f:
              return json.load(f)

      now = datetime.now(timezone.utc)
      keep_key = {{ inputs.keep_tag_key | toJson }}
      volume_min_age = int({{ inputs.volume_min_age_days }})
      snapshot_min_age = int({{ inputs.snapshot_min_age_days }})
      stopped_min = int({{ inputs.stopped_min_days }})
      prices = load("prices.json")
      account = load("identity.json")["Account"]

      def age_days(timestamp):
          return (now - datetime.fromisoformat(timestamp.replace("Z", "+00:00"))).days

      def kept(resource):
          return any(tag.get("Key") == keep_key for tag in resource.get("Tags") or [])

      def name(resource):
          return next((tag["Value"] for tag in resource.get("Tags") or [] if tag.get("Key") == "Name"), "")

      def volume_cost(volume_type, size):
          return round(prices["ebs_gb_month"].get(volume_type, prices["ebs_gb_month"]["gp3"]) * size, 2)

      rows, skipped = [], 0

      for volume in load("volumes.json")["Volumes"]:
          if kept(volume):
              skipped += 1
          elif age_days(volume["CreateTime"]) >= volume_min_age:
              rows.append({"type": "ebs_volume", "id": volume["VolumeId"], "name": name(volume),
                           "detail": f"{volume['Size']} GiB {volume['VolumeType']}, unattached",
                           "monthly_usd": volume_cost(volume["VolumeType"], volume["Size"]), "action": "delete"})

      for address in load("addresses.json")["Addresses"]:
          if address.get("AssociationId"):
              continue
          if kept(address):
              skipped += 1
              continue
          rows.append({"type": "elastic_ip", "id": address["AllocationId"], "name": name(address),
                       "detail": f"{address.get('PublicIp')} not associated",
                       "monthly_usd": prices["public_ipv4_month"], "action": "release"})

      ami_snapshots = {
          mapping["Ebs"]["SnapshotId"]
          for image in load("images.json")["Images"]
          for mapping in image.get("BlockDeviceMappings", [])
          if mapping.get("Ebs", {}).get("SnapshotId")
      }
      for snapshot in load("snapshots.json")["Snapshots"]:
          # Some emulators ignore the owner filter, so check the owner explicitly.
          if snapshot.get("OwnerId") != account or snapshot["SnapshotId"] in ami_snapshots:
              continue
          if kept(snapshot):
              skipped += 1
          elif age_days(snapshot["StartTime"]) >= snapshot_min_age:
              rows.append({"type": "snapshot", "id": snapshot["SnapshotId"], "name": name(snapshot),
                           "detail": f"{snapshot['VolumeSize']} GiB, {age_days(snapshot['StartTime'])} days old, not used by an AMI",
                           "monthly_usd": round(prices["snapshot_gb_month"] * snapshot["VolumeSize"], 2), "action": "delete"})

      volume_sizes = {v["VolumeId"]: v for v in load("all_volumes.json")["Volumes"]}
      for reservation in load("instances.json")["Reservations"]:
          for instance in reservation["Instances"]:
              if kept(instance):
                  skipped += 1
                  continue
              match = re.search(r"\((\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})", instance.get("StateTransitionReason", ""))
              if not match:
                  continue
              stopped_days = (now - datetime.strptime(match.group(1), "%Y-%m-%d %H:%M:%S").replace(tzinfo=timezone.utc)).days
              if stopped_days < stopped_min:
                  continue
              attached = [volume_sizes[m["Ebs"]["VolumeId"]] for m in instance.get("BlockDeviceMappings", []) if m.get("Ebs", {}).get("VolumeId") in volume_sizes]
              rows.append({"type": "stopped_instance", "id": instance["InstanceId"], "name": name(instance),
                           "detail": f"{instance['InstanceType']} stopped {stopped_days} days, {len(attached)} volume(s) still billed",
                           "monthly_usd": round(sum(volume_cost(v["VolumeType"], v["Size"]) for v in attached), 2), "action": "review"})

      with open("idle_resources.csv", "w", newline="") as f:
          writer = csv.DictWriter(f, fieldnames=["type", "id", "name", "detail", "monthly_usd", "action"])
          writer.writeheader()
          writer.writerows(rows)

      def ids(kind):
          return [row["id"] for row in rows if row["type"] == kind]

      deletable = [row for row in rows if row["action"] != "review"]
      top = sorted(rows, key=lambda row: -row["monthly_usd"])[:10]
      Kestra.outputs({
          "account": account,
          "volume_ids": ids("ebs_volume"),
          "address_ids": ids("elastic_ip"),
          "snapshot_ids": ids("snapshot"),
          "instance_ids": ids("stopped_instance"),
          "deletable_count": len(deletable),
          "review_count": len(rows) - len(deletable),
          "skipped_by_tag": skipped,
          "monthly_waste_usd": round(sum(row["monthly_usd"] for row in rows), 2),
          "deletable_monthly_usd": round(sum(row["monthly_usd"] for row in deletable), 2),
          "top_lines": "\n".join(f"- {row['type']} `{row['id']}` {row['name']} {row['detail']}: ${row['monthly_usd']}/month" for row in top),
      })

  - id: report
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Post the findings, whatever the mode.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    messageText: |
      *AWS idle resource scan* (account {{ outputs.analyze.vars.account }}, {{ inputs.region }}){% if inputs.dry_run %}, dry run{% endif %}
      Estimated waste: *${{ outputs.analyze.vars.monthly_waste_usd }}/month*. {{ outputs.analyze.vars.deletable_count }} resource(s) can be removed (${{ outputs.analyze.vars.deletable_monthly_usd }}/month), {{ outputs.analyze.vars.review_count }} stopped instance(s) need a human decision, {{ outputs.analyze.vars.skipped_by_tag }} skipped by the `{{ inputs.keep_tag_key }}` tag.
      {{ outputs.analyze.vars.top_lines }}
      Full CSV report on Kestra execution {{ execution.id }}.

  - id: cleanup
    type: io.kestra.plugin.core.flow.If
    description: Only when dry run is off and there is something to remove.
    condition: "{{ not inputs.dry_run and outputs.analyze.vars.deletable_count > 0 }}"
    then:
      - id: request_approval
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Ask for approval before anything is deleted.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        messageText: |
          *Approval needed:* delete {{ outputs.analyze.vars.deletable_count }} idle resource(s) in {{ inputs.region }} to save about ${{ outputs.analyze.vars.deletable_monthly_usd }}/month{% if inputs.snapshot_before_delete %}. Volumes get a backup snapshot first{% endif %}.
          Resume Kestra execution {{ execution.id }} within 48 hours to proceed. If nobody does, nothing is deleted.

      - id: approve_cleanup
        type: io.kestra.plugin.core.flow.Pause
        description: Wait for a human. After 48 hours the execution is cancelled and
          nothing is deleted.
        pauseDuration: PT48H
        behavior: CANCEL

      - id: delete_resources
        type: io.kestra.plugin.aws.cli.AwsCLI
        description: Back up and delete volumes, release Elastic IPs and delete
          snapshots. Each resource is handled independently, and any failure is
          listed and fails the task after the rest are processed.
        accessKeyId: "{{ secret('AWS_ACCESS_KEY_ID') }}"
        secretKeyId: "{{ secret('AWS_SECRET_ACCESS_KEY') }}"
        region: "{{ inputs.region }}"
        env:
          ENDPOINT: "{{ inputs.endpoint_url ?? '' }}"
          BACKUP: "{{ inputs.snapshot_before_delete }}"
        outputFiles:
          - cleanup_log.txt
        commands:
          - EP=${ENDPOINT:+--endpoint-url $ENDPOINT}
          - touch cleanup_log.txt failures.txt
          - |
            for volume in {{ outputs.analyze.vars.volume_ids | join(' ') }}; do
              if [ "$BACKUP" = "true" ]; then
                snapshot=$(aws $EP ec2 create-snapshot --volume-id "$volume" --description "Backup before idle-resource cleanup ({{ execution.id }})" --tag-specifications "ResourceType=snapshot,Tags=[{Key=created-by,Value=kestra-idle-resource-reaper},{Key=source-volume,Value=$volume}]" --query SnapshotId --output text) || { echo "$volume backup" >> failures.txt; continue; }
                aws $EP ec2 wait snapshot-completed --snapshot-ids "$snapshot" || { echo "$volume backup-wait" >> failures.txt; continue; }
                echo "backed up $volume as $snapshot" >> cleanup_log.txt
              fi
              aws $EP ec2 delete-volume --volume-id "$volume" && echo "deleted volume $volume" >> cleanup_log.txt || echo "$volume delete" >> failures.txt
            done
          - |
            for address in {{ outputs.analyze.vars.address_ids | join(' ') }}; do
              aws $EP ec2 release-address --allocation-id "$address" && echo "released address $address" >> cleanup_log.txt || echo "$address release" >> failures.txt
            done
          - |
            for snapshot in {{ outputs.analyze.vars.snapshot_ids | join(' ') }}; do
              aws $EP ec2 delete-snapshot --snapshot-id "$snapshot" && echo "deleted snapshot $snapshot" >> cleanup_log.txt || echo "$snapshot delete" >> failures.txt
            done
          - cat cleanup_log.txt
          - if [ -s failures.txt ]; then echo "Failed:"; cat failures.txt; cat
            failures.txt >> cleanup_log.txt; exit 1; fi

      - id: confirm_cleanup
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Confirm what was removed.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        messageText: |
          *Cleanup done* in {{ inputs.region }}: {{ outputs.analyze.vars.volume_ids | length }} volume(s), {{ outputs.analyze.vars.address_ids | length }} Elastic IP(s) and {{ outputs.analyze.vars.snapshot_ids | length }} snapshot(s) removed, saving about ${{ outputs.analyze.vars.deletable_monthly_usd }}/month. Log on Kestra execution {{ execution.id }}.

errors:
  - id: alert_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert Slack when the scan or the cleanup fails.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    messageText: |
      AWS idle resource reaper FAILED in {{ inputs.region }} (execution {{ execution.id }}). Check the task logs; a partial cleanup lists the resources that failed in `cleanup_log.txt`.

outputs:
  - id: report
    type: FILE
    displayName: Idle resource report
    description: CSV of every idle resource found, with its estimated monthly cost
      and proposed action.
    value: "{{ outputs.analyze.outputFiles['idle_resources.csv'] }}"
  - id: monthly_waste_usd
    type: FLOAT
    displayName: Estimated monthly waste (USD)
    value: "{{ outputs.analyze.vars.monthly_waste_usd }}"