Run s3 jobs via step functions

infra/modules/service/events_jobs.tf

@@ -40,30 +40,18 @@ resource "aws_cloudwatch_event_target" "document_upload_jobs" {
 
   target_id = "${local.cluster_name}-${each.key}"
   rule      = aws_cloudwatch_event_rule.file_upload_jobs[each.key].name
-  arn       = aws_ecs_cluster.cluster.arn
+  arn       = aws_sfn_state_machine.file_upload_jobs[each.key].arn
   role_arn  = aws_iam_role.events.arn
 
-  ecs_target {
-    task_definition_arn = aws_ecs_task_definition.app.arn
-    launch_type         = "FARGATE"
-    propagate_tags      = "TASK_DEFINITION"
-
-    # Configuring Network Configuration is required when the task definition uses the awsvpc network mode.
-    network_configuration {
-      subnets         = var.private_subnet_ids
-      security_groups = [aws_security_group.app.id]
-    }
-  }
-
   input_transformer {
     input_paths = {
       bucket_name = "$.detail.bucket.name",
       object_key  = "$.detail.object.key",
     }
 
-    # When triggering the ECS task, override the command to run in the container to the
-    # command specified by the file_upload_job config. To do this define an input_template
-    # that transforms the input S3 event:
+    # When triggering the ECS task (via step functions), override the command to run in
+    # the container to the command specified by the file_upload_job config. To do this
+    # define an input_template that transforms the input S3 event:
     #   {
     #     detail: {
     #       bucket: { name: "mybucket" },
@@ -98,10 +86,85 @@ resource "aws_cloudwatch_event_target" "document_upload_jobs" {
     input_template = replace(replace(jsonencode({
       containerOverrides = [
         {
-          name    = local.container_name,
           command = each.value.task_command
         }
       ]
     }), "\\u003c", "<"), "\\u003e", ">")
   }
 }
+
+resource "aws_sfn_state_machine" "file_upload_jobs" {
+  for_each = var.file_upload_jobs
+
+  name     = "${var.service_name}-${each.key}"
+  role_arn = aws_iam_role.workflow_orchestrator.arn
+
+  definition = jsonencode({
+    "StartAt" : "RunTask",
+    "States" : {
+      "RunTask" : {
+        "Type" : "Task",
+        # docs: https://docs.aws.amazon.com/step-functions/latest/dg/connect-ecs.html
+        "Resource" : "arn:aws:states:::ecs:runTask.sync",
+        "Parameters" : {
+          "Cluster" : aws_ecs_cluster.cluster.arn,
+          "TaskDefinition" : aws_ecs_task_definition.app.arn,
+          "LaunchType" : "FARGATE",
+          "NetworkConfiguration" : {
+            "AwsvpcConfiguration" : {
+              "Subnets" : var.private_subnet_ids,
+              "SecurityGroups" : [aws_security_group.app.id],
+            }
+          },
+          "Overrides" : {
+            "ContainerOverrides" : [
+              {
+                # Pull the task command out of the input data, which is shaped like so:
+                #
+                # {
+                #   "containerOverrides": [
+                #     {
+                #       "command": [
+                #         "<task_command_arg_1>"
+                #         "<task_command_arg_2>"
+                #         ...
+                #       ]
+                #     }
+                #   ]
+                # }
+                #
+                # The syntax for parsing the input data comes from JSONPath.
+                "Name" : local.container_name,
+                "Command.$" : "$.containerOverrides[0].command[*]"
+              }
+            ]
+          }
+        },
+        "End" : true
+      }
+    }
+  })
+
+  logging_configuration {
+    log_destination        = "${aws_cloudwatch_log_group.file_upload_jobs[each.key].arn}:*"
+    include_execution_data = true
+    level                  = "ERROR"
+  }
+
+  tracing_configuration {
+    enabled = true
+  }
+}
+
+resource "aws_cloudwatch_log_group" "file_upload_jobs" {
+  for_each = var.file_upload_jobs
+
+  name_prefix = "/aws/vendedlogs/states/${var.service_name}-${each.key}"
+
+  # Conservatively retain logs for 5 years.
+  # Looser requirements may allow shorter retention periods
+  retention_in_days = 1827
+
+  # TODO(https://github.com/navapbc/template-infra/issues/164) Encrypt with customer managed KMS key
+  # checkov:skip=CKV_AWS_158:Encrypt service logs with customer key in future work
+}

infra/modules/service/events_role.tf

@@ -30,27 +30,35 @@ resource "aws_iam_policy" "run_task" {
 
 data "aws_iam_policy_document" "run_task" {
   statement {
-    effect    = "Allow"
-    actions   = ["ecs:RunTask"]
-    resources = ["${aws_ecs_task_definition.app.arn_without_revision}:*"]
-    condition {
-      test     = "ArnLike"
-      variable = "ecs:cluster"
-      values   = [aws_ecs_cluster.cluster.arn]
+    sid = "StepFunctionsEvents"
+    actions = [
+      "events:PutTargets",
+      "events:PutRule",
+      "events:DescribeRule",
+    ]
+    resources = ["arn:aws:events:${data.aws_region.current.name}:${data.aws_caller_identity.current.account_id}:rule/StepFunctionsGetEventsForStepFunctionsExecutionRule"]
+  }
+
+  dynamic "statement" {
+    for_each = aws_sfn_state_machine.file_upload_jobs
+
+    content {
+      actions = [
+        "states:StartExecution",
+      ]
+      resources = [statement.value.arn]
     }
   }
 
-  statement {
-    effect  = "Allow"
-    actions = ["iam:PassRole"]
-    resources = [
-      aws_iam_role.task_executor.arn,
-      aws_iam_role.app_service.arn,
-    ]
-    condition {
-      test     = "StringLike"
-      variable = "iam:PassedToService"
-      values   = ["ecs-tasks.amazonaws.com"]
+  dynamic "statement" {
+    for_each = aws_sfn_state_machine.file_upload_jobs
+
+    content {
+      actions = [
+        "states:DescribeExecution",
+        "states:StopExecution",
+      ]
+      resources = ["${statement.value.arn}:*"]
     }
   }
 }