AWS EFS Burst Credit Balance Depletion Monitor

id: aws-efs-burst-credit-balance-depletion-monitor
namespace: company.cloud

description: |
  Audits AWS Elastic File System (EFS) instances via CloudWatch to detect burst credit balance depletion
  and I/O throughput throttling before stateful workloads and container volumes lock up.

concurrency:
  limit: 1

inputs:
  - id: aws_region
    type: STRING
    displayName: AWS Region
    defaults: us-east-1
    description: Target AWS region where the EFS file systems and CloudWatch metrics reside.

  - id: file_system_id
    type: STRING
    displayName: EFS File System ID
    defaults: "fs-0123456789abcdef0"
    description: Unique identifier of the target AWS EFS file system to monitor.

  - id: min_burst_credit_bytes_threshold
    type: FLOAT
    displayName: Minimum Burst Credit Threshold (Bytes)
    defaults: 107374182400.0
    description: Minimum acceptable BurstCreditBalance in bytes (default 100 GiB)
      before triggering depletion alerts.

  - id: alert_on_depletion
    type: BOOL
    displayName: Alert on Depletion
    defaults: true
    description: Whether to dispatch automated Slack incident notifications when
      burst credits are exhausted.

tasks:
  - id: scan_efs_credits
    type: io.kestra.plugin.scripts.python.Script
    description: Queries AWS CloudWatch for EFS BurstCreditBalance and
      PermittedThroughput metrics.
    containerImage: python:3.11-slim
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    beforeCommands:
      - pip install boto3 kestra
    env:
      AWS_ACCESS_KEY_ID: "{{ secret('AWS_ACCESS_KEY_ID') }}"
      AWS_SECRET_ACCESS_KEY: "{{ secret('AWS_SECRET_ACCESS_KEY') }}"
      AWS_DEFAULT_REGION: "{{ inputs.aws_region }}"
    script: |
      import os
      import json
      import boto3
      from datetime import datetime, timedelta, timezone
      from kestra import Kestra

      region = os.environ.get("AWS_DEFAULT_REGION", "us-east-1")
      fs_id = "{{ inputs.file_system_id }}".strip()
      min_credits_threshold = float("{{ inputs.min_burst_credit_bytes_threshold }}")

      cw_client = boto3.client("cloudwatch", region_name=region)
      efs_client = boto3.client("efs", region_name=region)

      end_time = datetime.now(timezone.utc)
      start_time = end_time - timedelta(minutes=30)

      fs_name = fs_id
      throughput_mode = "bursting"
      try:
          desc = efs_client.describe_file_systems(FileSystemId=fs_id)
          items = desc.get("FileSystems", [])
          if items:
              fs_name = items[0].get("Name", fs_id)
              throughput_mode = items[0].get("ThroughputMode", "bursting")
      except Exception as e:
          print(f"Notice: Could not fetch EFS metadata: {e}")

      # Query BurstCreditBalance metric
      resp = cw_client.get_metric_data(
          MetricDataQueries=[
              {
                  "Id": "burst_credits",
                  "MetricStat": {
                      "Metric": {
                          "Namespace": "AWS/EFS",
                          "MetricName": "BurstCreditBalance",
                          "Dimensions": [{"Name": "FileSystemId", "Value": fs_id}]
                      },
                      "Period": 300,
                      "Stat": "Minimum"
                  },
                  "ReturnData": True
              },
              {
                  "Id": "permitted_throughput",
                  "MetricStat": {
                      "Metric": {
                          "Namespace": "AWS/EFS",
                          "MetricName": "PermittedThroughput",
                          "Dimensions": [{"Name": "FileSystemId", "Value": fs_id}]
                      },
                      "Period": 300,
                      "Stat": "Average"
                  },
                  "ReturnData": True
              }
          ],
          StartTime=start_time,
          EndTime=end_time
      )

      results = {r["Id"]: (r["Values"][0] if r["Values"] else 0.0) for r in resp.get("MetricDataResults", [])}
      current_credits = results.get("burst_credits", 0.0)
      permitted_throughput = results.get("permitted_throughput", 0.0)

      is_depleted = (current_credits < min_credits_threshold) and (throughput_mode == "bursting")
      current_gib = round(current_credits / (1024 ** 3), 2)
      threshold_gib = round(min_credits_threshold / (1024 ** 3), 2)

      report = {
          "file_system_id": fs_id,
          "file_system_name": fs_name,
          "throughput_mode": throughput_mode,
          "burst_credits_gib": current_gib,
          "threshold_gib": threshold_gib,
          "permitted_throughput_bytes": permitted_throughput,
          "is_depleted": is_depleted,
          "remediation_command": f"aws efs update-file-system --file-system-id {fs_id} --throughput-mode elastic"
      }

      with open("efs_report.json", "w") as f:
          json.dump(report, f)

      Kestra.outputs({
          "file_system_id": fs_id,
          "file_system_name": fs_name,
          "throughput_mode": throughput_mode,
          "burst_credits_gib": current_gib,
          "is_depleted": is_depleted
      })

  - id: check_burst_credit_depletion
    type: io.kestra.plugin.core.flow.If
    description: Evaluates whether the EFS burst credit balance has dropped below
      safety limits.
    condition: "{{ fromJson(read('efs_report.json')).is_depleted and
      inputs.alert_on_depletion }}"
    then:
      - id: notify_slack_credit_exhaustion
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Dispatches high-priority EFS I/O throttling alert to platform
          engineering on Slack.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "channel": "#cloud-storage-alerts",
            "username": "EFS-Sentinel",
            "text": "CRITICAL: AWS EFS File System `{{ fromJson(read('efs_report.json')).file_system_name }}` (`{{ fromJson(read('efs_report.json')).file_system_id }}`) burst credit balance is severely depleted! Remaining credits: `{{ fromJson(read('efs_report.json')).burst_credits_gib }} GiB` (Threshold: {{ fromJson(read('efs_report.json')).threshold_gib }} GiB). File system is at risk of baseline I/O throttling. Remediation: `{{ fromJson(read('efs_report.json')).remediation_command }}`"
          }

  - id: log_storage_health_state
    type: io.kestra.plugin.core.log.Log
    description: Records an operational audit trail of EFS storage burst credit health.
    message: "AWS EFS burst credit audit completed for '{{
      fromJson(read('efs_report.json')).file_system_id }}'. Balance: {{
      fromJson(read('efs_report.json')).burst_credits_gib }} GiB, Depleted: {{
      fromJson(read('efs_report.json')).is_depleted }}."

  - id: export_efs_manifest
    type: io.kestra.plugin.core.debug.Return
    description: Emits structured manifest of EFS burst credit diagnostics for
      infrastructure observability.
    format: "{{ fromJson(read('efs_report.json')) }}"

triggers:
  - id: hourly_efs_credit_audit
    type: io.kestra.plugin.core.trigger.Schedule
    description: Automated schedule trigger auditing EFS burst credit metrics every hour.
    cron: "0 * * * *"