如何获取stepfunction执行失败时所在的组件名称

前置准备 #

• 安装AWS SDK for Python：执行pip install boto3
• 提前配置好AWS访问凭证，确保所用身份拥有states:GetExecutionHistory权限，后续代码中替换为你实际使用的AWS区域即可

核心逻辑 #

Step Functions不会直接在执行详情中返回失败节点名称，所有执行过程的节点流转、错误信息都存储在执行历史事件中：

1. 执行失败时会生成ExecutionFailed全局事件，记录执行整体的错误信息
2. 具体组件节点失败时会生成对应类型的失败事件：比如Lambda任务失败对应LambdaFunctionFailed、并行节点失败对应ParallelStateFailed、Map节点失败对应MapStateFailed，这类事件的previousEventId会关联到节点进入时生成的*StateEntered事件，从进入事件的详情中就能拿到失败节点的名称
3. 调用get_execution_history接口时开启倒序拉取，可以优先拿到最新的失败事件，执行历史较长时能减少不必要的API调用量

完整实现代码 #

import boto3

# 初始化Step Functions客户端，替换为你实际使用的区域
sfn_client = boto3.client('stepfunctions', region_name='cn-north-1')

def get_failed_node_detail(execution_arn: str) -> dict:
    """
    输入失败状态机执行的ARN，返回结构化的失败位置、错误信息
    """
    # 用分页器处理长执行历史的拉取
    paginator = sfn_client.get_paginator('get_execution_history')
    page_iterator = paginator.paginate(
        executionArn=execution_arn,
        reverseOrder=True
    )

    event_map = {}
    failed_result = {
        "execution_arn": execution_arn,
        "failed_node_name": None,
        "error_type": None,
        "error_cause": None,
        "is_global_error": False
    }

    for page in page_iterator:
        for event in page['events']:
            event_id = event['id']
            event_map[event_id] = event
            event_type = event['type']

            # 捕获全局执行失败事件
            if event_type == 'ExecutionFailed':
                failed_detail = event['executionFailedEventDetails']
                failed_result['error_type'] = failed_detail.get('error')
                failed_result['error_cause'] = failed_detail.get('cause')
                # 若执行启动后直接失败，属于状态机全局错误（如权限不足、定义语法错误）
                if event.get('previousEventId') == 1:
                    failed_result['is_global_error'] = True
                    return failed_result

            # 匹配各类组件节点的失败事件
            if event_type in [
                'TaskFailed', 'LambdaFunctionFailed', 'ActivityFailed',
                'ParallelStateFailed', 'MapStateFailed', 'ChoiceStateFailed',
                'WaitStateFailed', 'PassStateFailed', 'FailStateEntered'
            ]:
                # 顺着关联事件ID向上查找节点进入事件，获取节点名称
                current_prev_id = event['previousEventId']
                while current_prev_id in event_map:
                    prev_event = event_map[current_prev_id]
                    if prev_event['type'].endswith('StateEntered'):
                        state_detail_key = f"{prev_event['type'][:-7].lower()}StateEnteredEventDetails"
                        failed_result['failed_node_name'] = prev_event[state_detail_key]['name']
                        # 优先取节点级别的错误详情，覆盖全局错误信息
                        node_error_key = f"{event_type[:-6].lower()}EventDetails"
                        if node_error_key in event:
                            failed_result['error_type'] = event[node_error_key].get('error', failed_result['error_type'])
                            failed_result['error_cause'] = event[node_error_key].get('cause', failed_result['error_cause'])
                        return failed_result
                    current_prev_id = prev_event.get('previousEventId')
    return failed_result

if __name__ == "__main__":
    # 替换为你筛选出的所有失败executionArn列表
    failed_exec_arns = [
        "arn:aws-cn:states:cn-north-1:123456789012:execution:your-statemachine-name:exec-id-1",
        "arn:aws-cn:states:cn-north-1:123456789012:execution:your-statemachine-name:exec-id-2"
    ]
    for arn in failed_exec_arns:
        detail = get_failed_node_detail(arn)
        print("="*60)
        print(f"执行ARN: {detail['execution_arn']}")
        if detail['is_global_error']:
            print(f"[全局错误] 类型: {detail['error_type']}, 原因: {detail['error_cause']}")
        else:
            print(f"失败节点: {detail['failed_node_name']}")
            print(f"错误类型: {detail['error_type']}")
            print(f"错误原因: {detail['error_cause']}")

注意事项 #

• 上述代码已经处理了API分页逻辑，即使执行历史有上千条事件也能正常拉取
• 如果你的工作流存在嵌套Step Function调用（即Task节点触发另一个状态机执行），需要从失败Task节点的输出中拿到子执行的ARN，递归调用上述方法才能定位到最内层的失败节点
• 若返回结果标记is_global_error=True，说明错误不是出在具体业务组件节点，需要优先排查状态机角色权限、状态机定义语法、执行输入参数这类全局配置问题

内容的提问来源于stack exchange，提问作者Jay Patel