AWS Step Functions 和 Fargate 任务：容器运行时错误不会导致管道失败

Question

我在AWS Step Functions 中定义了一个管道。一个步骤被定义为Fargate Task，它拉取一个 docker 镜像并运行一些 python 代码。我惊讶地发现，如果在 Fargate 任务中运行的容器遇到运行时错误，Step Functions 不会捕获失败的任务并照常继续管道（将 Fargate 任务设置为成功），但根据文档，管道一旦发生这种情况，应该会失败。

这是阶跃函数定义：

{
  "Comment": "My state machine",
  "StartAt": "MyFargateTask",
  "States": {
    "MyFargateTask": {
      "Type": "Task",
      "Resource": "arn:aws:states:::ecs:runTask.sync",
      "InputPath": "$",
      "Parameters": {
        "Cluster": "my-cluster",
        "TaskDefinition": "arn:aws:ecs:us-east-1:617090640476:task-definition/my-task:1",
        "LaunchType": "FARGATE",
        "NetworkConfiguration": {
          "AwsvpcConfiguration": {
            "Subnets": [
              "subnet-xxxxxxxxxxxxxxxxx",
              "subnet-yyyyyyyyyyyyyyyyy"
            ],
            "AssignPublicIp": "ENABLED"
          }
        },
      },
      "Next": "Done"
    },
    "Done": {
      "Type": "Succeed"
    }
  }
}

我已经为 Fargate 容器尝试了以下简单的 python 代码：

def main():
    raise Exception("foobar")

if __name__ == '__main__':
    main()

在 CloudWatch 上的容器日志中，我可以看到程序按预期失败，但 Step Function 中的管道成功（全部为绿色）。我错过了什么？这是一个错误吗？

Answer 1

AWS Step Functions 不知道 ECS 作业是成功还是失败。Step Functions 需要查看 ECS 作业的容器日志，并尝试确定在 Docker 容器内运行的进程是否以失败代码退出。这不是 Step Functions 所做的。当您配置它时，Step Functions 只是假设只要容器存在，任务就成功了。

如果您更改arn:aws:states:::ecs:runTask.sync为arn:aws:states:::ecs:runTask.waitForTaskTokenthen 而不是仅仅等待 ECS 容器退出，Step Fuctions 将等待 ECS 容器将成功或失败代码发送回 Step Functions API。您还需要将任务令牌传递到 ECS 容器中，这可以通过ContainerOverrides设置完成，如下所示：

{
  "Comment": "My state machine",
  "StartAt": "MyFargateTask",
  "States": {
    "MyFargateTask": {
      "Type": "Task",
      "Resource": "arn:aws:states:::ecs:runTask.waitForTaskToken",
      "InputPath": "$",
      "Parameters": {
        "Cluster": "my-cluster",
        "TaskDefinition": "arn:aws:ecs:us-east-1:617090640476:task-definition/my-task:1",
        "LaunchType": "FARGATE",
        "NetworkConfiguration": {
          "AwsvpcConfiguration": {
            "Subnets": [
              "subnet-xxxxxxxxxxxxxxxxx",
              "subnet-yyyyyyyyyyyyyyyyy"
            ],
            "AssignPublicIp": "ENABLED"
          }
        },
        "Overrides": {
          "ContainerOverrides": [{
            "Environment": [{
              "Name": "TASK_TOKEN",
              "Value.$": "$$.Task.Token"
              }]
          }]
        }
      },
      "Next": "Done"
    },
    "Done": {
      "Type": "Succeed"
    }
  }
}

现在在您的 Python 脚本中，您可以获取TASK_TOKEN环境变量，并向 Step Functions 发送成功或失败消息，如下所示：

token = os.environ['TASK_TOKEN']

def step_success():
    if token is not None:
        stfn = boto3.client('stepfunctions')
        stfn.send_task_success(taskToken=token, output='{"Status": "Success"}')


def step_fail():
    if token is not None:
        stfn = boto3.client('stepfunctions')
        stfn.send_task_failure(taskToken=token, error="An error occurred")

有关此方法的更多详细信息

我还建议在状态机中配置超时，以防您的 Python 脚本无法在容器内执行。此外，您需要向 Fargate 任务的 IAM 角色添加适当的 IAM 权限，以允许它向 Step Functions API 发出这些状态调用。