Find Wasted Coverage in Underutilized AWS Savings Plans

id: aws-savings-plan-waste-finder
namespace: company.team
description: |
  Find AWS Savings Plans that are paid for but barely used, so finance can
  reassign or exchange coverage before the commitment renews. A disabled
  weekly schedule and a keyed on-demand webhook both start the same run: the
  Cost Explorer coverage query executes inside the official AWS CLI
  container with a bundled two-plan fixture behind it, Python turns
  utilization and monthly commitment into a wasted dollar figure per plan,
  and anything above the floor opens a GitHub issue, alerts Slack, and
  records a recommended move for every underutilized plan. The verdict, the waste total,
  the offending plans, and the report URI are published as outputs, and any
  failure escalates to Slack.

triggers:
  - id: weekly_waste_scan
    type: io.kestra.plugin.core.trigger.Schedule
    description: Weekly coverage review every Monday at 06:00 UTC, shipped disabled
      so nothing reaches finance until an operator switches it on.
    cron: "0 6 * * 1"
    disabled: true
  - id: on_demand_audit
    type: io.kestra.plugin.core.trigger.Webhook
    description: POST here to run a waste audit on demand, for example after a
      commitment purchase or before a renewal decision.
    key: "{{ secret('SAVINGSPLAN_WEBHOOK_KEY') }}"

inputs:
  - id: utilization_threshold
    type: FLOAT
    defaults: 80.0
    description: Savings plan utilization below this percentage counts as
      underutilized, so 80.0 flags every plan burning less than 80 percent of
      its committed spend.
  - id: min_waste_usd
    type: FLOAT
    defaults: 100.0
    description: Total monthly wasted commitment above this dollar figure escalates
      to a GitHub issue and to Slack instead of a quiet all-clear log.

tasks:
  # --- 1. coverage workspace ---
  - id: coverage_workspace
    type: io.kestra.plugin.core.flow.WorkingDirectory
    description: One shared directory so the coverage query, the waste analysis, and
      the report all read and write the same files.
    tasks:
      # --- 2. coverage query ---
      - id: fetch_coverage
        type: io.kestra.plugin.scripts.shell.Commands
        description: Write a deterministic two-plan fixture with a heredoc, run the Cost
          Explorer savings plans coverage query in the official AWS CLI
          container, and fall back to the fixture whenever the API cannot be
          reached so the audit stays reproducible offline.
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
          entryPoint: []
        containerImage: amazon/aws-cli:latest
        env:
          AWS_ACCESS_KEY_ID: "{{ secret('AWS_ACCESS_KEY_ID') }}"
          AWS_SECRET_ACCESS_KEY: "{{ secret('AWS_SECRET_ACCESS_KEY') }}"
          AWS_DEFAULT_REGION: us-east-1
        outputFiles:
          - coverage.json
        commands:
          - |
            set -u
            cat > savings-plans-fixture.json <<'JSON'
            {
              "SavingsPlansCoverages": [
                {"SavingsPlanId": "sp-3f2a1c9b7d5e4a60", "SavingsPlanType": "Compute", "Region": "us-east-1", "UtilizationPercent": 95.0, "MonthlyCommitment": 2400.0},
                {"SavingsPlanId": "sp-8c7d6e5f4a3b2c10", "SavingsPlanType": "EC2Instance", "Region": "us-east-1", "UtilizationPercent": 52.0, "MonthlyCommitment": 4000.0}
              ],
              "Total": {"UtilizationPercent": 70.3, "MonthlyCommitment": 6400.0}
            }
            JSON

            echo "Querying savings plans coverage from the Cost Explorer API"
            aws ce get-savings-plans-coverage --output json > coverage.json || true

            if [ ! -s coverage.json ]; then
              cp savings-plans-fixture.json coverage.json
              echo "coverage API unreachable from this worker, emitting the bundled fixture"
            fi

            echo "coverage bytes $(wc -c < coverage.json)"

      # --- 3. waste analysis ---
      - id: analyze_coverage
        type: io.kestra.plugin.scripts.python.Script
        description: Read the coverage payload, skip plans with no commitment left,
          compute waste_usd for every plan under the utilization threshold,
          write the report file, and emit the verdict through the outputs
          protocol.
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
        containerImage: python:3.12-slim
        env:
          UTILIZATION_THRESHOLD: "{{ inputs.utilization_threshold }}"
          MIN_WASTE_USD: "{{ inputs.min_waste_usd }}"
        outputFiles:
          - report.txt
        script: |
          import json
          import os
          from datetime import datetime, timezone

          def as_float(value, fallback=0.0):
              try:
                  return float(value)
              except (TypeError, ValueError):
                  return fallback

          coverage_path = "coverage.json"
          threshold = as_float(os.environ.get("UTILIZATION_THRESHOLD"), 80.0)
          floor = as_float(os.environ.get("MIN_WASTE_USD"), 100.0)

          document = {}
          if os.path.exists(coverage_path) and os.path.getsize(coverage_path) > 0:
              try:
                  with open(coverage_path, "r") as handle:
                      document = json.load(handle)
              except ValueError:
                  document = {}

          rows = []
          if isinstance(document, dict):
              candidate = document.get("SavingsPlansCoverages")
              if isinstance(candidate, list):
                  rows = candidate
          elif isinstance(document, list):
              rows = document

          low_util_plans = []
          reviewed_plans = 0
          reviewed_commitment = 0.0

          for row in rows:
              if not isinstance(row, dict):
                  continue
              plan_id = str(row.get("SavingsPlanId") or row.get("savingsPlanId") or row.get("id") or "unknown-plan")
              raw_util = row.get("UtilizationPercent", row.get("utilization"))
              raw_commit = row.get("MonthlyCommitment", row.get("commitment"))
              utilization = as_float(raw_util, 0.0)
              commitment = as_float(raw_commit, 0.0)
              if commitment <= 0.0:
                  continue
              reviewed_plans += 1
              reviewed_commitment += commitment
              if utilization < threshold:
                  waste = round(commitment * (1.0 - utilization / 100.0), 2)
                  low_util_plans.append({
                      "id": plan_id,
                      "utilization": round(utilization, 1),
                      "waste_usd": waste,
                  })

          low_util_plans.sort(key=lambda plan: (-plan["waste_usd"], plan["id"]))
          waste_usd = round(float(sum(plan["waste_usd"] for plan in low_util_plans)), 2)
          report_time = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
          verdict = "WASTE_FOUND" if waste_usd > floor else "ALL_CLEAR"

          with open("report.txt", "w") as handle:
              handle.write("AWS Savings Plans waste report\n")
              handle.write("report time " + report_time + "\n")
              handle.write("utilization threshold " + str(threshold) + " percent\n")
              handle.write("escalation floor " + str(floor) + " USD per month\n")
              handle.write("plans reviewed " + str(reviewed_plans) + "\n")
              handle.write("commitment reviewed " + str(round(reviewed_commitment, 2)) + " USD per month\n")
              handle.write("wasted commitment " + str(waste_usd) + " USD per month\n")
              handle.write("verdict " + verdict + "\n\n")
              if low_util_plans:
                  handle.write("Underutilized plans\n")
                  for plan in low_util_plans:
                      handle.write(
                          "- " + plan["id"] + " at " + str(plan["utilization"])
                          + " percent wastes " + str(plan["waste_usd"]) + " USD per month\n"
                      )
              else:
                  handle.write("No savings plan sits below the utilization threshold, nothing to reassign.\n")

              if low_util_plans:
                  handle.write("\nRecommended moves\n")
                  for plan in low_util_plans:
                      if plan["utilization"] < 50.0:
                          move = "exchange the plan or let it lapse at renewal"
                      else:
                          move = "exchange into a smaller commitment at the next window"
                      handle.write("- " + plan["id"] + ": " + move + "\n")

          print("reviewed=%d low_util=%d waste_usd=%.2f verdict=%s" % (
              reviewed_plans, len(low_util_plans), waste_usd, verdict))
          print("::" + json.dumps({"outputs": {
              "verdict": verdict,
              "waste_usd": waste_usd,
              "low_util_plans": low_util_plans,
              "report_time": report_time,
          }}) + "::")

  # --- 4. verdict gate ---
  - id: check_waste
    type: io.kestra.plugin.core.flow.If
    description: Escalate only when the wasted monthly commitment sits above the
      dollar floor, a healthy book of coverage is recorded as an all-clear.
    condition: "{{ (outputs.analyze_coverage.vars.waste_usd | number) >
      (inputs.min_waste_usd | number) }}"
    then:
      - id: append_advice
        type: io.kestra.plugin.core.flow.Loop
        description: Walk the underutilized plans one at a time and record a recommended
          exchange or reassignment move for each plan.
        values: "{{ outputs.analyze_coverage.vars.low_util_plans }}"
        concurrencyLimit: 1
        tasks:
          - id: append_plan_advice
            type: io.kestra.plugin.scripts.python.Script
            description: Build the advice line for the current plan from its utilization and
              wasted dollars, and emit it through the outputs protocol.
            taskRunner:
              type: io.kestra.plugin.scripts.runner.docker.Docker
            containerImage: python:3.12-slim
            env:
              PLAN_ID: "{{ item.id }}"
              PLAN_UTILIZATION: "{{ item.utilization }}"
              PLAN_WASTE_USD: "{{ item.waste_usd }}"
            script: |
              import json
              import os
              from datetime import datetime, timezone

              plan_id = os.environ.get("PLAN_ID", "unknown-plan")
              utilization = 0.0
              waste = 0.0
              try:
                  utilization = float(os.environ.get("PLAN_UTILIZATION", "0"))
              except (TypeError, ValueError):
                  pass
              try:
                  waste = float(os.environ.get("PLAN_WASTE_USD", "0"))
              except (TypeError, ValueError):
                  pass

              if utilization < 50.0:
                  advice = "commitment is barely used, exchange the plan or let it lapse at renewal"
              else:
                  advice = "coverage is underused, exchange into a smaller commitment at the next window"

              stamp = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
              line = stamp + " " + plan_id + " at " + ("%.1f" % utilization)
              line += " percent wastes " + ("%.2f" % waste) + " USD per month, " + advice

              print(line)
              print("::" + json.dumps({"outputs": {"appended": plan_id, "advice_line": line, "waste_usd": round(waste, 2)}}) + "::")

      - id: open_exchange_issue
        type: io.kestra.plugin.github.issues.Create
        description: Open a labeled GitHub issue carrying the waste total and the plan
          list so the exchange work is tracked where finance already works.
        oauthToken: "{{ secret('GITHUB_TOKEN') }}"
        repository: your-org/cloud-commitments
        title: "Savings Plans wasting ${{ outputs.analyze_coverage.vars.waste_usd }} per
          month"
        body: |
          ## Savings Plan waste audit

          Wasted commitment **${{ outputs.analyze_coverage.vars.waste_usd }}** per month sits above the ${{ inputs.min_waste_usd }} floor, with a utilization threshold of {{ inputs.utilization_threshold }} percent.

          Report time `{{ outputs.analyze_coverage.vars.report_time }}`, execution `{{ execution.id }}`.

          ### Underutilized plans

          ```json
          {{ outputs.analyze_coverage.vars.low_util_plans | toJson }}
          ```

          ### Next steps

          - Exchange the lowest utilized plans into smaller commitments before the renewal window closes.
          - Reassign coverage to the workloads that actually run, then re-run this flow to confirm the waste dropped.
        labels:
          - finops
          - savings-plans
      - id: notify_waste
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Tell the finops channel the wasted commitment, the threshold that
          tripped, and that a GitHub issue is waiting.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "AWS Savings Plan waste found - ${{ outputs.analyze_coverage.vars.waste_usd }} per month of committed coverage is underutilized, above the ${{ inputs.min_waste_usd }} floor (utilization threshold {{ inputs.utilization_threshold }}%). Issue opened in your-org/cloud-commitments. Execution {{ execution.id }}."
          }
    else:
      - id: log_all_clear
        type: io.kestra.plugin.core.log.Log
        description: Record the all-clear so the execution history shows which weeks
          were reviewed.
        message: "Savings plan waste ${{ outputs.analyze_coverage.vars.waste_usd }} per
          month is under the ${{ inputs.min_waste_usd }} floor, no coverage to
          reassign."

errors:
  - id: notify_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Escalate any failure to the ops channel with the execution url.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {"text": "FAILED: {{ flow.id }} {{ execution.id }}"}

outputs:
  - id: verdict
    type: STRING
    description: WASTE_FOUND when the wasted commitment beats the dollar floor,
      otherwise ALL_CLEAR.
    value: "{{ outputs.analyze_coverage.vars.verdict }}"
  - id: waste_usd
    type: FLOAT
    description: Total monthly commitment wasted across every plan below the
      utilization threshold, for example 1920.0.
    value: "{{ outputs.analyze_coverage.vars.waste_usd }}"
  - id: low_util_plans
    type: ARRAY
    description: 'Underutilized plans as objects with id, utilization, and
      waste_usd, for example [{"id": "sp-8c7d6e5f4a3b2c10", "utilization": 52.0,
      "waste_usd": 1920.0}].'
    value: "{{ outputs.analyze_coverage.vars.low_util_plans | toJson }}"
  - id: report_uri
    type: STRING
    description: Internal storage URI of the report.txt waste report written by the
      analysis step.
    value: "{{ outputs.analyze_coverage.outputFiles['report.txt'] }}"