Purge empty log streams with AWS Step Functions

Policy.json

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "logs:DescribeLogGroups",
        "logs:DescribeLogStreams",
        "logs:GetLogEvents",
        "logs:DeleteLogStream"
      ],
      "Resource": "*"
    }
  ]
}

StateMachine.asl.json

{
  "Comment": "Purge empty log streams from log groups.",
  "StartAt": "DescribeLogGroups",
  "States": {
    "DescribeLogGroups": {
      "Type": "Task",
      "Next": "ForEachLogGroup",
      "Parameters": {},
      "Resource": "arn:aws:states:::aws-sdk:cloudwatchlogs:describeLogGroups"
    },
    "ForEachLogGroup": {
      "Type": "Map",
      "ItemProcessor": {
        "ProcessorConfig": {
          "Mode": "INLINE"
        },
        "StartAt": "HasRetentionSetting",
        "States": {
          "HasRetentionSetting": {
            "Type": "Choice",
            "Choices": [
              {
                "Variable": "$.RetentionInDays",
                "IsPresent": true,
                "Next": "DescribeLogStreams"
              }
            ],
            "Default": "NextLogGroup"
          },
          "NextLogGroup": {
            "Type": "Succeed"
          },
          "DescribeLogStreams": {
            "Type": "Task",
            "Parameters": {
              "LogGroupName.$": "$.LogGroupName",
              "OrderBy": "LastEventTime"
            },
            "Resource": "arn:aws:states:::aws-sdk:cloudwatchlogs:describeLogStreams",
            "Next": "ForEachLogStream",
            "ResultPath": "$.LogStreams"
          },
          "ForEachLogStream": {
            "Type": "Map",
            "ItemProcessor": {
              "ProcessorConfig": {
                "Mode": "INLINE"
              },
              "StartAt": "GetLogEventCount",
              "States": {
                "GetLogEventCount": {
                  "Type": "Task",
                  "Parameters": {
                    "LogGroupName.$": "$.LogGroupName",
                    "LogStreamName.$": "$.LogStreamName",
                    "StartFromHead": true,
                    "Limit": 1
                  },
                  "Resource": "arn:aws:states:::aws-sdk:cloudwatchlogs:getLogEvents",
                  "Next": "IfEmptyStream",
                  "ResultPath": "$.Events",
                  "ResultSelector": {
                    "Count.$": "States.ArrayLength($.Events)"
                  }
                },
                "IfEmptyStream": {
                  "Type": "Choice",
                  "Choices": [
                    {
                      "Variable": "$.Events.Count",
                      "NumericEquals": 0,
                      "Next": "DeleteLogStream"
                    }
                  ],
                  "Default": "NextLogStream"
                },
                "NextLogStream": {
                  "Type": "Succeed"
                },
                "DeleteLogStream": {
                  "Type": "Task",
                  "Parameters": {
                    "LogGroupName.$": "$.LogGroupName",
                    "LogStreamName.$": "$.LogStreamName"
                  },
                  "Resource": "arn:aws:states:::aws-sdk:cloudwatchlogs:deleteLogStream",
                  "Next": "NextLogStream"
                }
              }
            },
            "ItemSelector": {
              "LogGroupName.$": "$.LogGroupName",
              "LogStreamName.$": "$$.Map.Item.Value.LogStreamName"
            },
            "ItemsPath": "$.LogStreams.LogStreams",
            "MaxConcurrency": 0,
            "Next": "HasMoreLogStreams",
            "ResultPath": "$.LogStreams",
            "ResultSelector": []
          },
          "HasMoreLogStreams": {
            "Type": "Choice",
            "Choices": [
              {
                "Variable": "$.NextToken",
                "IsPresent": true,
                "Next": "DescribeNextLogStreams"
              }
            ],
            "Default": "NextLogGroup"
          },
          "DescribeNextLogStreams": {
            "Type": "Task",
            "Parameters": {
              "LogGroupName.$": "$.LogGroupName",
              "OrderBy": "LastEventTime",
              "NextToken.$": "$.NextToken"
            },
            "Resource": "arn:aws:states:::aws-sdk:cloudwatchlogs:describeLogStreams",
            "Next": "ForEachLogStream"
          }
        }
      },
      "ItemsPath": "$.LogGroups",
      "MaxConcurrency": 0,
      "ResultPath": "$.LogGroups",
      "Next": "HasMoreLogGroups",
      "ResultSelector": [],
      "Catch": [
        {
          "ErrorEquals": [
            "States.TaskFailed"
          ],
          "Next": "NoMoreLogGroup",
          "Comment": "Ignore error (maybe rate exceeded)"
        }
      ]
    },
    "HasMoreLogGroups": {
      "Type": "Choice",
      "Choices": [
        {
          "Variable": "$.NextToken",
          "IsPresent": true,
          "Next": "DescribeNextLogGroups"
        }
      ],
      "Default": "NoMoreLogGroup"
    },
    "DescribeNextLogGroups": {
      "Type": "Task",
      "Parameters": {
        "NextToken.$": "$.NextToken"
      },
      "Resource": "arn:aws:states:::aws-sdk:cloudwatchlogs:describeLogGroups",
      "Next": "ForEachLogGroup",
      "Catch": [
        {
          "ErrorEquals": [
            "States.TaskFailed"
          ],
          "Comment": "Ignore error (maybe rate exceedeed)",
          "Next": "NoMoreLogGroup"
        }
      ]
    },
    "NoMoreLogGroup": {
      "Type": "Succeed"
    }
  }
}