I implemented disk monitoring without Lambda using Step Functions + JSONata in Kiro which supports Opus 4.6

# Disk Usage Monitor Step Functions Implementation Plan v4

## Change History

| Version | Changes |
|---|---|
| v3 | Initial version. All disk auto-detection with ListMetrics, double Map structure |
| v4 | Reflecting implementation differences. Empty array guards, improved prediction logic, MetricPeriod parameter added |

## 1. Overview

Monitor HDD usage custom metrics collected by CloudWatch Agent using Step Functions (SDK integration + JSONata) to detect disk depletion risks and send structured JSON notifications to SNS. Implemented without Lambda.

Monitoring logic:
- Use ListMetrics to automatically detect all disks (Dimension combinations) for each instance
- Identify maximum daily increment from the past 14 days of daily data
- Determine if disk will reach full (100%) within 3 days if that increment continues
- Immediate alert if current value exceeds threshold (default 95%)

## 2. Prerequisites

- CloudWatch Agent configured on target EC2 (namespace: `CWAgent`, metric: `disk_used_percent`)
- Identification tags applied to monitored EC2 instances (e.g., `Monitor:DiskCheck`)
- SNS topic created
- CWAgent Dimensions settings (append_dimensions, etc.) irrelevant. Auto-detected by ListMetrics

## 3. Architecture

EventBridge Scheduler (regular execution: rate(1 hour))
    |
    v
Step Functions (default: EXPRESS / can switch to STANDARD via parameter)
    |
    +-- [1] EC2 DescribeInstances (tag filter + running only)
    |
    +-- [2] Pass/JSONata (format instance IDs/names into array)
    |
    +-- [2.5] Choice (if 0 instances, exit normally)
    |
    +-- [3] Map (parallel: per instance)
    |       |
    |       +-- [3-1] CloudWatch ListMetrics (get all disk Dimension combinations)
    |       |
    |       +-- [3-2] Pass/JSONata (format Dimension combinations into array)
    |       |
    |       +-- [3-2.5] Choice (if 0 disks, end with empty result)
    |       |
    |       +-- [3-3] Map (parallel: per disk)
    |       |       +-- CloudWatch GetMetricStatistics (past 14 days, Period as parameter)
    |       |       +-- Pass/JSONata (calculate maximum increment through all pair comparisons, determine depletion risk)
    |       |
    |       +-- Aggregate instanceId/instanceName/results in Output
    |
    +-- [4] Pass/JSONata (categorize into CRITICAL/WARNING/OK, generate report)
    |
    +-- [5] Choice -- CRITICAL or WARNING exists -- SNS Publish
                   -- none -- normal exit

Express mode outputs to CloudWatch Logs (14-day retention)

## 4. State Machine Definition

| State | Type | Process |
|---|---|---|
| GetInstances | Task (SDK: EC2) | DescribeInstances. Get targets using tag filter + instance-state-name=running |
| ExtractInstanceList | Pass (JSONata) | Format instance IDs and Name tags into array |
| CheckHasInstances | Choice | If 0 instances, go to NoAction. Prevents empty array error in Map |
| CheckEachInstance | Map (parallel) | Execute following for each instance (outer Map). ToleratedFailurePercentage: 100 |
| -- ListDiskMetrics | Task (SDK: CloudWatch) | Get all disk Dimensions for that instance using ListMetrics |
| -- ExtractDimensions | Pass (JSONata) | Format Dimension combinations into array. Return empty array if metrics is null/empty |
| -- CheckHasDisks | Choice | If 0 disks, go to NoDisksFound. Prevents empty array error in Map |
| -- NoDisksFound | Pass | Return empty results and end |
| -- CheckEachDisk | Map (parallel) | Execute following for each disk (inner Map). ToleratedFailurePercentage: 100 |
| ---- GetDiskMetrics | Task (SDK: CloudWatch) | Get past LookbackDays data with GetMetricStatistics. Period is MetricPeriod parameter |
| ---- EvaluateRisk | Pass (JSONata) | Calculate max increment via all pair comparisons, determine depletion risk |
| AggregateReport | Pass (JSONata) | Categorize into CRITICAL/WARNING/OK, generate JSON report |
| HasAlerts | Choice | criticalCount > 0 or warningCount > 0 |
| SendReport | Task (SDK: SNS) | Publish JSON report. Subject: "Disk Usage Alert" |
| NoAction | Succeed | Normal exit |

## 5. CloudWatch API Specifications

### 5.1 ListMetrics (Disk Auto-Detection)

Once per instance. Gets all Dimension combinations for all disks on that instance.

| Parameter | Value |
|---|---|
| Namespace | CWAgent |
| MetricName | disk_used_percent |
| Dimensions | [{"Name": "InstanceId", "Value": "<target ID>"}] |
| RecentlyActive | PT3H |

RecentlyActive: PT3H returns only disks that have sent data within the last 3 hours.

### 5.2 GetMetricStatistics (Trend Data Retrieval)

Once per disk. Specifies the complete Dimensions obtained from ListMetrics as-is.

| Parameter | Value |
|---|---|
| Namespace | CWAgent |
| MetricName | disk_used_percent |
| Period | MetricPeriod parameter (default: 86400 seconds = 1 day) |
| Statistics | ["Average"] |
| StartTime | now - LookbackDays |
| EndTime | now |
| Dimensions | Use ListMetrics response as-is |

### 5.3 Cost Estimate

For 10 instances x avg. 2 disks/instance:
- ListMetrics: 10 calls/execution
- GetMetricStatistics: 20 calls/execution
- Total: 30 calls/execution x 24 executions/day x 30 days = 21,600 calls/month -> approx. $0.22/month

## 6. Prediction Logic (JSONata)

### 6.1 Processing Flow

1. Get latest datapoint using $reduce ($sort is unstable in Step Functions' JSONata, so not used)
2. If fewer than 2 datapoints -> unpredictable (INSUFFICIENT_DATA)
3. Calculate maximum increment and its date through all pair comparisons ($map + $reduce)
4. If maximum increment ≤ 0 -> no increasing trend (daysUntilFull: null)
5. daysUntilFull = (100 - currentUsage) / maxDailyIncrease

### 6.2 Judgment Criteria

| Condition | Status | Priority |
|---|---|---|
| currentUsage >= thresholdPercent | CRITICAL | 1 (highest) |
| datapoints < 2 | INSUFFICIENT_DATA | 2 |
| daysUntilFull <= predictionDays | WARNING | 3 |
| others | OK | 4 |

### 6.3 Corner Case Handling

| Case | Behavior |
|---|---|
| New instance (1 or fewer datapoints) | INSUFFICIENT_DATA. Not alert target |
| Disk cleanup days with usage decrease | Consider only positive increments in all pair comparisons |
| Usage flat or decreasing across all days | No positive increments -> daysUntilFull: null -> OK |
| Current value already 100% | CRITICAL (immediate judgment on threshold exceeded, even with insufficient data) |
| No disk metrics for instance | Return empty results (NoDisksFound state) |
| 0 monitored instances | Normal exit at CheckHasInstances |

### 6.4 EvaluateRisk Output Fields

| Field | Type | Description |
|---|---|---|
| path | string | Mount path (e.g., /, /data, C:) |
| device | string | Device name (e.g., nvme0n1p1, C:) |
| fstype | string | File system (e.g., xfs, NTFS) |
| currentUsage | number | Current usage (1 decimal place) |
| maxDailyIncrease | number/null | Maximum daily increment (2 decimal places). Null if insufficient data |
| daysUntilFull | number/null | Days until 100% (1 decimal place). Null if no increasing trend |
| worstDay | string/null | Date of maximum increment (YYYY-MM-DD). Null if no increasing trend |
| status | string | CRITICAL / WARNING / INSUFFICIENT_DATA / OK |

## 7. SNS Notification Message

Subject: Disk Usage Alert

json
{
  "subject": "Disk Usage Alert",
  "timestamp": "2026-02-07T08:47:49.163Z",
  "summary": {
    "totalInstances": 10,
    "totalDisks": 18,
    "criticalCount": 1,
    "warningCount": 2,
    "okCount": 15
  },
  "critical": [
    {
      "instanceId": "i-0123456789abcdef0",
      "instanceName": "db-server",
      "path": "/data",
      "device": "nvme1n1",
      "fstype": "xfs",
      "currentUsage": 96.2,
      "maxDailyIncrease": 2.8,
      "daysUntilFull": 1.4,
      "worstDay": "2026-02-03"
    }
  ],
  "warning": [
    {
      "instanceId": "i-0abcdef1234567890",
      "instanceName": "app-server",
      "path": "/",
      "device": "nvme0n1p1",
      "fstype": "xfs",
      "currentUsage": 87.3,
      "maxDailyIncrease": 5.1,
      "daysUntilFull": 2.5,
      "worstDay": "2026-02-05"
    }
  ]
}

## 8. CloudFormation Parameters

| Parameter | Default | Description |
|---|---|---|
| StateMachineType | EXPRESS | EXPRESS or STANDARD |
| TagKey | Monitor | Monitoring target tag key |
| TagValue | DiskCheck | Monitoring target tag value |
| ThresholdPercent | 95 | CRITICAL judgment threshold (%) |
| PredictionDays | 3 | WARNING judgment days |
| LookbackDays | 14 | Historical days for trend analysis |
| MetricPeriod | 86400 | Metric aggregation period (seconds). Set to 300 during testing |
| SnsTopicArn | (required) | Notification SNS topic ARN |
| ScheduleExpression | rate(1 hour) | Execution schedule |

AWSTemplateFormatVersion: '2010-09-09'
Description: Disk Usage Monitor - Step Functions (Lambdaless / JSONata)

Parameters:
  StateMachineType:
    Type: String
    Default: EXPRESS
    AllowedValues: [EXPRESS, STANDARD]
  TagKey:
    Type: String
    Default: Monitor
  TagValue:
    Type: String
    Default: DiskCheck
  ThresholdPercent:
    Type: Number
    Default: 95
  PredictionDays:
    Type: Number
    Default: 3
  LookbackDays:
    Type: Number
    Default: 14
  SnsTopicArn:
    Type: String
    AllowedPattern: arn:aws:sns:.+
  MetricPeriod:
    Type: Number
    Default: 86400
    Description: Metric aggregation period in seconds (86400=daily, 300=5min for testing)
  ScheduleExpression:
    Type: String
    Default: rate(1 hour)

Conditions:
  IsExpress: !Equals [!Ref StateMachineType, EXPRESS]

Resources:
  StateMachineLogGroup:
    Type: AWS::Logs::LogGroup
    Condition: IsExpress
    Properties:
      LogGroupName: !Sub /aws/vendedlogs/states/${AWS::StackName}
      RetentionInDays: 14

  StateMachineRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service: states.amazonaws.com
            Action: sts:AssumeRole
      Policies:
        - PolicyName: DiskMonitorPolicy
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Effect: Allow
                Action:
                  - ec2:DescribeInstances
                  - cloudwatch:ListMetrics
                  - cloudwatch:GetMetricStatistics
                Resource: '*'
              - Effect: Allow
                Action: sns:Publish
                Resource: !Ref SnsTopicArn
              - !If
                - IsExpress
                - Effect: Allow
                  Action:
                    - logs:CreateLogDelivery
                    - logs:GetLogDelivery
                    - logs:UpdateLogDelivery
                    - logs:DeleteLogDelivery
                    - logs:ListLogDeliveries
                    - logs:PutResourcePolicy
                    - logs:DescribeResourcePolicies
                    - logs:DescribeLogGroups
                  Resource: '*'
                - !Ref AWS::NoValue

  SchedulerRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service: scheduler.amazonaws.com
            Action: sts:AssumeRole
      Policies:
        - PolicyName: InvokeStateMachine
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Effect: Allow
                Action:
                  - states:StartExecution
                  - states:StartSyncExecution
                Resource: !GetAtt StateMachine.Arn

  StateMachine:
    Type: AWS::StepFunctions::StateMachine
    Properties:
      StateMachineName: !Sub ${AWS::StackName}
      StateMachineType: !Ref StateMachineType
      RoleArn: !GetAtt StateMachineRole.Arn
      LoggingConfiguration: !If
        - IsExpress
        - Level: ALL
          IncludeExecutionData: true
          Destinations:
            - CloudWatchLogsLogGroup:
                LogGroupArn: !GetAtt StateMachineLogGroup.Arn
        - !Ref AWS::NoValue
      DefinitionSubstitutions:
        TagKey: !Ref TagKey
        TagValue: !Ref TagValue
        ThresholdPercent: !Ref ThresholdPercent
        PredictionDays: !Ref PredictionDays
        LookbackDays: !Ref LookbackDays
        MetricPeriod: !Ref MetricPeriod
        SnsTopicArn: !Ref SnsTopicArn
      Definition:
        QueryLanguage: JSONata
        Comment: Disk Usage Monitor - Lambdaless
        StartAt: GetInstances
        States:
          GetInstances:
            Type: Task
            Resource: arn:aws:states:::aws-sdk:ec2:describeInstances
            Arguments:
              Filters:
                - Name: "tag:${TagKey}"
                  Values:
                    - "${TagValue}"
                - Name: instance-state-name
                  Values:
                    - running
            Next: ExtractInstanceList

          ExtractInstanceList:
            Type: Pass
            Output: >-
              {% [$states.input.Reservations[].Instances[].{
                'instanceId': InstanceId,
                'instanceName': Tags[Key='Name'].Value
              }] %}
            Next: CheckHasInstances

          CheckHasInstances:
            Type: Choice
            Choices:
              - Condition: "{% $count($states.input) > 0 %}"
                Next: CheckEachInstance
            Default: NoAction

          CheckEachInstance:
            Type: Map
            Items: "{% $states.input %}"
            MaxConcurrency: 10
            ToleratedFailurePercentage: 100
            ItemProcessor:
              ProcessorConfig:
                Mode: INLINE
              StartAt: ListDiskMetrics
              States:
                ListDiskMetrics:
                  Type: Task
                  Resource: arn:aws:states:::aws-sdk:cloudwatch:listMetrics
                  Arguments:
                    Namespace: CWAgent
                    MetricName: disk_used_percent
                    Dimensions:
                      - Name: InstanceId
                        Value: "{% $states.input.instanceId %}"
                    RecentlyActive: PT3H
                  Output:
                    instanceId: "{% $states.input.instanceId %}"
                    instanceName: "{% $states.input.instanceName %}"
                    metrics: "{% $states.result.Metrics %}"
                  Next: ExtractDimensions

                ExtractDimensions:
                  Type: Pass
                  Output:
                    instanceId: "{% $states.input.instanceId %}"
                    instanceName: "{% $states.input.instanceName %}"
                    disks: >-
                      {% $states.input.metrics
                        ? [$states.input.metrics.{ 'dimensions': Dimensions }]
                        : [] %}
                  Next: CheckHasDisks

                CheckHasDisks:
                  Type: Choice
                  Choices:
                    - Condition: "{% $count($states.input.disks) > 0 %}"
                      Next: CheckEachDisk
                  Default: NoDisksFound

                NoDisksFound:
                  Type: Pass
                  Output:
                    instanceId: "{% $states.input.instanceId %}"
                    instanceName: "{% $states.input.instanceName %}"
                    results: []
                  End: true

                CheckEachDisk:
                  Type: Map
                  Items: >-
                    {% [$states.input.disks.{
                      'dimensions': dimensions,
                      'instanceId': $states.input.instanceId,
                      'instanceName': $states.input.instanceName
                    }] %}
                  MaxConcurrency: 5
                  ToleratedFailurePercentage: 100
                  ItemProcessor:
                    ProcessorConfig:
                      Mode: INLINE
                    StartAt: GetDiskMetrics
                    States:
                      GetDiskMetrics:
                        Type: Task
                        Resource: arn:aws:states:::aws-sdk:cloudwatch:getMetricStatistics
                        Arguments:
                          Namespace: CWAgent
                          MetricName: disk_used_percent
                          Dimensions: "{% $states.input.dimensions %}"
                          StartTime: >-
                            {% $fromMillis($millis() - ${LookbackDays} * 86400000) %}
                          EndTime: "{% $fromMillis($millis()) %}"
                          Period: "{% $number('${MetricPeriod}') %}"
                          Statistics:
                            - Average
                        Output:
                          dimensions: "{% $states.input.dimensions %}"
                          datapoints: "{% $states.result.Datapoints %}"
                        Next: EvaluateRisk

                      EvaluateRisk:
                        Type: Pass
                        Output: >-
                          {% (
                            $dims := $states.input.dimensions;
                            $path := ($dims[Name='path']).Value;
                            $device := ($dims[Name='device']).Value;
                            $fstype := ($dims[Name='fstype']).Value;
                            $dp := $states.input.datapoints;
                            $cnt := $count($dp);
                            $latest := $cnt > 0
                              ? $reduce($dp, function($acc, $v) {
                                  $v.Timestamp > $acc.Timestamp ? $v : $acc
                                })
                              : {'Average': 0};
                            $current := $latest.Average;
                            $prediction := $cnt < 2
                              ? { 'daysUntilFull': null, 'maxDailyIncrease': null, 'worstDay': null }
                              : (
                                $maxes := $map($dp, function($a) {
                                  $reduce($dp, function($best, $b) {
                                    ($a.Timestamp > $b.Timestamp)
                                      ? ( $d := $a.Average - $b.Average;
                                          $d > $best.delta
                                            ? {'delta': $d, 'date': $a.Timestamp}
                                            : $best )
                                      : $best
                                  }, {'delta': 0, 'date': ''})
                                });
                                $worst := $reduce($maxes, function($acc, $m) {
                                  $m.delta > $acc.delta ? $m : $acc
                                });
                                $worst.delta > 0
                                  ? {
                                    'daysUntilFull': $round((100 - $current) / $worst.delta, 1),
                                    'maxDailyIncrease': $round($worst.delta, 2),
                                    'worstDay': $substringBefore($worst.date, 'T')
                                  }
                                  : { 'daysUntilFull': null, 'maxDailyIncrease': 0, 'worstDay': null }
                              );
                            $status := $current >= ${ThresholdPercent}
                              ? 'CRITICAL'
                              : $cnt < 2
                                ? 'INSUFFICIENT_DATA'
                                : $prediction.daysUntilFull != null
                                  ? ($prediction.daysUntilFull <= ${PredictionDays}
                                      ? 'WARNING' : 'OK')
                                  : 'OK';
                            {
                              'path': $path,
                              'device': $device,
                              'fstype': $fstype,
                              'currentUsage': $round($current, 1),
                              'maxDailyIncrease': $prediction.maxDailyIncrease,
                              'daysUntilFull': $prediction.daysUntilFull,
                              'worstDay': $prediction.worstDay,
                              'status': $status
                            }
                          ) %}
                        End: true
                  Output:
                    instanceId: "{% $states.input.instanceId %}"
                    instanceName: "{% $states.input.instanceName %}"
                    results: "{% $states.result %}"
                  End: true
            Next: AggregateReport

          AggregateReport:
            Type: Pass
            Output: >-
              {% (
                $all := $states.input;
                $allDisks := $all.results[];
                {
                  'subject': 'Disk Usage Alert',
                  'timestamp': $now(),
                  'summary': {
                    'totalInstances': $count($all),
                    'totalDisks': $count($allDisks),
                    'criticalCount': $count($filter($allDisks, function($d) { $d.status = 'CRITICAL' })),
                    'warningCount': $count($filter($allDisks, function($d) { $d.status = 'WARNING' })),
                    'okCount': $count($filter($allDisks, function($d) { $d.status = 'OK' }))
                  },
                  'critical': $all.(
                    $inst := $;
                    results[status='CRITICAL'].{
                      'instanceId': $inst.instanceId,
                      'instanceName': $inst.instanceName,
                      'path': path, 'device': device, 'fstype': fstype,
                      'currentUsage': currentUsage,
                      'maxDailyIncrease': maxDailyIncrease,
                      'daysUntilFull': daysUntilFull,
                      'worstDay': worstDay
                    }
                  )[],
                  'warning': $all.(
                    $inst := $;
                    results[status='WARNING'].{
                      'instanceId': $inst.instanceId,
                      'instanceName': $inst.instanceName,
                      'path': path, 'device': device, 'fstype': fstype,
                      'currentUsage': currentUsage,
                      'maxDailyIncrease': maxDailyIncrease,
                      'daysUntilFull': daysUntilFull,
                      'worstDay': worstDay
                    }
                  )[]
                }
              ) %}
            Next: HasAlerts

          HasAlerts:
            Type: Choice
            Choices:
              - Condition: >-
                  {% $states.input.summary.criticalCount > 0
                    or $states.input.summary.warningCount > 0 %}
                Next: SendReport
            Default: NoAction

          SendReport:
            Type: Task
            Resource: arn:aws:states:::aws-sdk:sns:publish
            Arguments:
              TopicArn: "${SnsTopicArn}"
              Subject: "{% $states.input.subject %}"
              Message: "{% $string($states.input) %}"
            End: true

          NoAction:
            Type: Succeed

  Scheduler:
    Type: AWS::Scheduler::Schedule
    Properties:
      Name: !Sub ${AWS::StackName}-scheduler
      ScheduleExpression: !Ref ScheduleExpression
      FlexibleTimeWindow:
        Mode: 'OFF'
      Target:
        Arn: !GetAtt StateMachine.Arn
        RoleArn: !GetAtt SchedulerRole.Arn
        Input: '{}'

Outputs:
  StateMachineArn:
    Value: !GetAtt StateMachine.Arn
  StateMachineType:
    Value: !Ref StateMachineType
  LogGroupName:
    Condition: IsExpress
    Value: !Ref StateMachineLogGroup