AWS Cost Explorer Spend Anomaly Guard

id: aws-cost-explorer-anomaly-guardrail
namespace: company.team
description: |
  Compare AWS spend month-to-date against the identical window one month
  earlier using the Cost Explorer API, and alert Slack when the pace of
  spending has jumped past your threshold — while the month is still
  running and there is still time to react.

triggers:
  - id: daily_cost_check
    type: io.kestra.plugin.core.trigger.Schedule
    description: Check the spend trajectory once a day. Shipped disabled; enable it
      after your first successful manual run.
    cron: "0 7 * * *"
    disabled: true

inputs:
  - id: spike_threshold_pct
    type: FLOAT
    defaults: 25.0
    description: Allowed increase in month-to-date spend versus the same window last
      month, in percent, before Slack is alerted.
  - id: aws_region
    type: STRING
    defaults: us-east-1
    description: Region used for the Cost Explorer API call — the ce endpoint is
      only served from us-east-1 and us-west-2.

tasks:
  - id: compute_windows
    type: io.kestra.plugin.scripts.python.Script
    description: Build both comparison windows — month-to-date (the 1st through
      tomorrow, since Cost Explorer's End is exclusive) and the identical
      day-range one month earlier — so the comparison is like-for-like instead
      of a partial month against a full one.
    taskRunner:
      type: io.kestra.plugin.core.runner.Process
    script: |
      from datetime import date, timedelta
      import json

      today = date.today()
      month_start = today.replace(day=1)
      current_end = today + timedelta(days=1)

      if month_start.month == 1:
          prev_start = date(month_start.year - 1, 12, 1)
      else:
          prev_start = date(month_start.year, month_start.month - 1, 1)
      days_elapsed = (today - month_start).days + 1
      prev_end = prev_start + timedelta(days=days_elapsed)

      windows = {
          "current_start": month_start.isoformat(),
          "current_end": current_end.isoformat(),
          "previous_start": prev_start.isoformat(),
          "previous_end": prev_end.isoformat(),
      }
      print(f"Comparing {windows['current_start']}..{windows['current_end']} against {windows['previous_start']}..{windows['previous_end']}")
      print("::" + json.dumps({"outputs": windows}) + "::")

  - id: fetch_current_by_service
    type: io.kestra.plugin.aws.cli.AwsCLI
    description: Pull UnblendedCost for the current month-to-date window, grouped by
      service, so the alert can name the service driving the change.
    accessKeyId: "{{ secret('AWS_ACCESS_KEY_ID') }}"
    secretKeyId: "{{ secret('AWS_SECRET_ACCESS_KEY') }}"
    region: "{{ inputs.aws_region }}"
    commands:
      - aws ce get-cost-and-usage --time-period Start={{
        outputs.compute_windows.vars.current_start }},End={{
        outputs.compute_windows.vars.current_end }} --granularity DAILY
        --metrics UnblendedCost --group-by Type=DIMENSION,Key=SERVICE --query
        'ResultsByTime[].Groups[].[Key, Metrics.UnblendedCost.Amount]' --output
        json

  - id: fetch_previous_by_service
    type: io.kestra.plugin.aws.cli.AwsCLI
    description: Pull the same per-service numbers for the identical window one
      month earlier, giving the comparison a real baseline instead of a
      full-month guess.
    accessKeyId: "{{ secret('AWS_ACCESS_KEY_ID') }}"
    secretKeyId: "{{ secret('AWS_SECRET_ACCESS_KEY') }}"
    region: "{{ inputs.aws_region }}"
    commands:
      - aws ce get-cost-and-usage --time-period Start={{
        outputs.compute_windows.vars.previous_start }},End={{
        outputs.compute_windows.vars.previous_end }} --granularity DAILY
        --metrics UnblendedCost --group-by Type=DIMENSION,Key=SERVICE --query
        'ResultsByTime[].Groups[].[Key, Metrics.UnblendedCost.Amount]' --output
        json

  - id: compare_spend
    type: io.kestra.plugin.scripts.python.Script
    description: Reduce both per-service payloads to month-to-date totals, compute
      the percent change (with a $0.01 floor so brand-new spend against an empty
      baseline still trips the guard), and keep the leading service from the
      current window.
    taskRunner:
      type: io.kestra.plugin.core.runner.Process
    env:
      CURRENT_DATA: "{{ outputs.fetch_current_by_service.stdOut }}"
      PREVIOUS_DATA: "{{ outputs.fetch_previous_by_service.stdOut }}"
    script: |
      import json
      import os

      def totals_by_service(raw):
          rows = json.loads(raw or "[]")
          services = {}
          for service, amount in rows:
              services[service] = services.get(service, 0.0) + float(amount)
          return services

      current = totals_by_service(os.environ["CURRENT_DATA"])
      previous = totals_by_service(os.environ["PREVIOUS_DATA"])
      current_total = round(sum(current.values()), 2)
      previous_total = round(sum(previous.values()), 2)
      baseline = max(previous_total, 0.01)
      pct_change = round((current_total - previous_total) / baseline * 100, 1)
      top_service = max(current, key=current.get) if current else "n/a"

      print(f"Month-to-date ${current_total} vs ${previous_total} in the same window last month ({pct_change}%), top service: {top_service}")
      print("::" + json.dumps({"outputs": {
          "current_cost": current_total,
          "previous_cost": previous_total,
          "pct_change": pct_change,
          "top_service": top_service,
      }}) + "::")

  - id: check_spike
    type: io.kestra.plugin.core.flow.If
    description: Compare the percent change against the threshold input and only
      wake anyone when the month is tracking over.
    condition: "{{ outputs.compare_spend.vars.pct_change > inputs.spike_threshold_pct }}"
    then:
      - id: alert_spike
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Post the measured spend, last month's equivalent, the percent
          change, and the leading service so the channel knows where to look.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "AWS cost guard: month-to-date spend is ${{ outputs.compare_spend.vars.current_cost }} ({{ outputs.compare_spend.vars.pct_change }}% vs ${{ outputs.compare_spend.vars.previous_cost }} in the same window last month), above the {{ inputs.spike_threshold_pct }}% threshold. Top service: {{ outputs.compare_spend.vars.top_service }}. Execution {{ execution.id }}."
          }
    else:
      - id: log_on_track
        type: io.kestra.plugin.core.log.Log
        description: Quietly record healthy checks so the execution history becomes a
          day-by-day record of the month's trajectory.
        message: "AWS month-to-date spend ${{ outputs.compare_spend.vars.current_cost
          }}, {{ outputs.compare_spend.vars.pct_change }}% vs the same window
          last month — under the {{ inputs.spike_threshold_pct }}% threshold."

errors:
  - id: alert_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert when the guard itself fails, because a broken cost check
      looks exactly like a cheap month until the invoice lands.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "AWS cost guard FAILED in flow {{ flow.id }} (execution {{ execution.id }}). Month-to-date spend was NOT checked — verify the AWS credentials and the ce:GetCostAndUsage permission."
      }

outputs:
  - id: cost_summary
    type: JSON
    description: 'Month-to-date cost, last month''s equivalent window, percent
      change, and the top service — e.g. {"current_cost": 154.2,
      "previous_cost": 121.7, "pct_change": 26.7, "top_service": "Amazon EC2"}.'
    value: "{{ outputs.compare_spend.vars | toJson }}"