Apache Airflow 集成 Amazon SSM：SsmRunCommandOperator 运维命令执行与状态监控实战指南

Apache Airflow 集成 Amazon SSMSsmRunCommandOperator 运维命令执行与状态监控实战指南【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflowAmazon Simple Systems ManagerSSM是 AWS 提供的节点运维服务可在 AWS 云上、本地数据中心与多云环境中统一查看、管理和操作海量节点。本指南以 Apache Airflow 的 amazon provider 中的 SSM 模块为核心讲解如何用SsmRunCommandOperator在托管节点上远程执行命令、用SsmRunCommandCompletedSensor与SsmRunCommandTrigger跟踪命令状态、用SsmGetCommandInvocationOperator回收执行输出并结合fail_on_nonzero_exit参数实现基于退出码的工作流路由。读完本文你将掌握一套从命令下发、状态等待到输出回收的完整 SSM 运维编排方案。一、理解 SSM 与托管节点Managed Nodes前提SSM 将大量常用节点管理工具整合到一项服务中可跨 AWS 账户与 Region 完成日常节点任务。要使用 SSM节点必须是被托管的机器上需要安装SSM Agent并且 Agent 能与 Systems Manager 服务正常通信。从当前仓库的实现看SSM 相关能力集中在 amazon provider 的如下模块中SSM 操作符SsmRunCommandOperator、SsmGetCommandInvocationOperatorSSM 传感器SsmRunCommandCompletedSensorSSM 触发器SsmRunCommandTriggerSSM Hook底层封装boto3.client(ssm)的SsmHook仓库中的系统测试示例 example_ssm.py 演示了完整的托管前提创建带实例配置文件的 EC2 实例、通过 user-data 脚本安装并启动amazon-ssm-agent然后用describe_instance_information轮询PingStatus Online来确认节点已注册进 SSM。这也验证了官方文档所述节点必须安装 SSM Agent 且 Agent 与 SSM 服务可通信命令才能真正下发执行。二、环境准备安装、连接与 IAM 权限使用这些操作符之前需要完成三件事对应 prerequisite_tasks.rst创建必要的 AWS 资源使用 AWS Console 或 AWS CLI 预先准备 SSM 文档、托管实例、IAM 角色等资源。通过 pip 安装 API 库pip install apache-airflow[amazon]配置 AWS Connection参考 AWS 连接配置文档在 Airflow 中建立指向 AWS 凭证的 Connection默认 ID 为aws_default。IAM 权限清单要让 Airflow 具备下发、检索和监控 SSM Run Command 的能力需要为执行角色授予以下 IAM 权限来自 ssm.rst 原文{ Effect: Allow, Action: [ ssm:SendCommand, ssm:ListCommandInvocations, ssm:GetCommandInvocation ], Resource: * }其中ssm:SendCommand下发 Run CommandSsmRunCommandOperator的核心调用ssm:ListCommandInvocations枚举某个 CommandId 在各实例上的调用记录传感器轮询、触发器查询均依赖它ssm:GetCommandInvocation获取单实例的详细输出SsmGetCommandInvocationOperator与失败诊断依赖它。该策略允许访问所有 SSM 文档与托管实例。生产环境建议遵循最小权限原则将Resource收窄为具体的 SSM 文档 ARN以及在适用情况下收窄为目标资源如 EC2 实例的 ARN。三、通用参数aws_conn_id、region_name、verify、botocore_configSSM 相关的 Operator 与 Sensor 均继承自AwsBaseOperator/AwsBaseSensor共享以下通用参数详见 generic_parameters.rst参数说明默认值aws_conn_id引用 AWS Connection 的 ID设为None时走 boto3 默认凭证查找不查 Connection。分布式部署下若为None每台 Worker 都需自行维护 boto3 配置aws_defaultregion_nameAWS Region 名称为None时使用 Connection Extra 中的region_nameNoneverify是否校验 SSL 证书。False不校验也可传 CA 证书包文件路径None时使用 Connection Extra 中的verifyNonebotocore_config构造botocore.config.Config的字典用于配置重试、超时、签名等可规避节流异常None使用 Connection 配置botocore_config示例同样来自 generic_parameters.rst{ signature_version: unsigned, s3: { us_east_1_regional_endpoint: True, }, retries: { mode: standard, max_attempts: 10, }, connect_timeout: 300, read_timeout: 300, tcp_keepalive: True, }注意传入空字典{}会覆盖Connection 中已有的 botocore 配置。四、SsmRunCommandOperator在一个或多个托管节点上执行命令SsmRunCommandOperator通过 SSM Run Command 在目标节点上执行命令是本文的核心组件。核心参数参数说明默认值document_name必填要运行的 SSM 文档名称如AWS-RunShellScript—run_command_kwargs透传给send_commandAPI 的可选参数如InstanceIds、Parameters、Comment等{}wait_for_completion是否等待命令执行完成Truewaiter_delay状态检查之间的等待秒数120waiter_max_attempts检查完成的最大尝试次数75fail_on_nonzero_exit命令返回非零退出码时是否让任务失败Truedeferrable是否以 deferrable异步/触发器模式等待需要安装 aiobotocore该模式隐含等待完成False可由配置项operators.default_deferrable覆盖document_name与run_command_kwargs都是模板字段见 ssm.py 中的template_fields支持 Jinja 模板渲染例如用 XCom 动态注入实例 ID。最小示例from airflow.providers.amazon.aws.operators.ssm import SsmRunCommandOperator run_command SsmRunCommandOperator( task_idrun_command, document_nameAWS-RunShellScript, run_command_kwargs{ InstanceIds: [i-0123456789abcdef0], Parameters: {commands: [touch /tmp/ssm_test_passed]}, }, )执行链路与三种等待模式从 ssm.py 的execute()源码可以看到该操作符的执行分为两个阶段调用self.hook.conn.send_command(DocumentName..., **run_command_kwargs)下发命令并从响应中取出Command[CommandId]根据配置进入三种等待路径之一deferrableTrue异步模式将任务挂起self.defer(...)把等待交给SsmRunCommandTrigger由 Trigger 在触发器进程中异步等待command_executedWaiter完成后通过execute_complete()恢复任务并返回command_id。适合长耗时命令可显著释放 Worker 资源。wait_for_completionTrue同步模式用self.hook.get_waiter(command_executed)对response[Command][InstanceIds]中的每个实例逐一waiter.wait(...)并按waiter_delay/waiter_max_attempts轮询。wait_for_completionFalse即发即走模式只下发命令立即返回command_id由下游 Sensor 负责等待。无论哪种模式操作符最终都会返回command_iddeferrable 与同步模式下返回值相同可直接作为下游任务如 Sensor、SsmGetCommandInvocationOperator的输入。系统测试中正是通过run_command.output把命令 ID 传递给后续任务。五、退出码处理fail_on_nonzero_exit 的两种行为默认情况下SsmRunCommandOperator与SsmRunCommandCompletedSensor在命令返回非零退出码时会让任务失败。通过fail_on_nonzero_exit参数可以改变这一行为# 默认行为 —— 非零退出码导致任务失败 run_command SsmRunCommandOperator( task_idrun_command, document_nameAWS-RunShellScript, run_command_kwargs{...}, ) # 允许非零退出码 —— 无论退出码如何任务都成功 run_command SsmRunCommandOperator( task_idrun_command, document_nameAWS-RunShellScript, run_command_kwargs{...}, fail_on_nonzero_exitFalse, )设置fail_on_nonzero_exitFalse后可以配合SsmGetCommandInvocationOperator取回退出码用于工作流路由决策例如命令执行失败但需要走补偿分支而不是直接失败。需要特别强调的是无论该参数如何设置AWS 层面的失败如TimedOut、Cancelled都仍会抛出异常。这一规则在三层实现中保持一致SsmHook.is_aws_level_failure() 静态方法把Cancelled、TimedOut、Cancelling判定为 AWS 级失败操作符在捕获WaiterError后若处于增强模式fail_on_nonzero_exitFalse会先调用get_command_invocation获取状态命中 AWS 级失败则raise重抛传感器与触发器同样只在非 AWS 级失败时容忍AWS 级失败一律报错。六、SsmGetCommandInvocationOperator回收命令输出与执行详情SsmGetCommandInvocationOperator用于检索已执行 SSM 命令的输出与执行细节典型使用场景包括回收上游SsmRunCommandOperator已执行命令的输出获取在 Airflow 之外如手动或其它系统发起的 SSM 命令结果为调试或数据处理检查命令结果。获取所有实例的输出不指定instance_id时操作符会先通过list_command_invocations(command_id)枚举执行过该命令的所有实例再逐个拉取输出get_all_output SsmGetCommandInvocationOperator( task_idget_command_output, command_id{{ ti.xcom_pull(task_idsrun_command) }}, # 来自上游任务 )获取指定实例的输出指定instance_id时只查询单个实例系统测试中的写法见 example_ssm.pyget_command_output SsmGetCommandInvocationOperator( task_idget_command_output, command_idrun_command.output, instance_idinstance_id, )返回的数据结构从 ssm.py 的execute()实现看返回的是结构化字典顶层包含command_id与invocations列表每个 invocation 包含instance_id实例 IDstatus命令执行状态如Success、Failedresponse_code命令退出码配合fail_on_nonzero_exitFalse做路由判断的关键字段standard_output/standard_error标准输出与标准错误内容execution_start_time/execution_end_time执行起止时间document_name文档名称comment命令备注。单个实例查询失败时该实例会以{instance_id: ..., error: ...}形式记录到结果中而不会让整个任务中断便于后续统一排查。七、SsmRunCommandCompletedSensor等待 SSM 命令进入终态当SsmRunCommandOperator以wait_for_completionFalse即发即走时可用SsmRunCommandCompletedSensor在独立任务中等待命令到达终态from airflow.providers.amazon.aws.sensors.ssm import SsmRunCommandCompletedSensor await_run_command SsmRunCommandCompletedSensor( task_idawait_run_command, command_idrun_command.output, )从 ssm.py 的源码可以看到其状态机设计中间态Pending、Delayed、InProgress、Cancelling—— 继续轮询返回False失败态Cancelled、TimedOut、Failed—— 依据fail_on_nonzero_exit决定抛错或容忍成功态Success—— 结束轮询。Sensor 关键参数poke_interval轮询间隔默认 120 秒、max_retries最大轮询次数默认 75、fail_on_nonzero_exit、deferrable。同样地deferrableTrue时 Sensor 会委托给SsmRunCommandTrigger异步等待而不是占用 Worker 轮询。八、三种编排模式与基于退出码的路由仓库系统测试 example_ssm.py 完整演示了三种可落地的编排模式可直接作为实战模板模式一传统模式Operator 下发 Sensor 等待run_command_traditional SsmRunCommandOperator( task_idrun_command_traditional, document_nameAWS-RunShellScript, run_command_kwargs{ InstanceIds: [instance_id], Parameters: {commands: [echo Testing traditional pattern, exit 0]}, }, wait_for_completionFalse, ) wait_command_traditional SsmRunCommandCompletedSensor( task_idwait_command_traditional, command_idrun_command_traditional.output, )模式二增强异步模式容忍失败 退出码路由run_command_async SsmRunCommandOperator( task_idrun_command_async, document_nameAWS-RunShellScript, run_command_kwargs{ InstanceIds: [instance_id], Parameters: {commands: [echo Testing async pattern, exit 1]}, }, wait_for_completionFalse, fail_on_nonzero_exitFalse, ) wait_command_async SsmRunCommandCompletedSensor( task_idwait_command_async, command_idrun_command_async.output, fail_on_nonzero_exitFalse, ) get_exit_code_output SsmGetCommandInvocationOperator( task_idget_exit_code_output, command_idrun_command_async.output, instance_idinstance_id, ) task def route_based_on_exit_code(**context): output context[ti].xcom_pull(task_idsget_exit_code_output) exit_code output.get(response_code) if output else None log.info(Command exit code: %s, exit_code) return handle_exit_code这里命令故意exit 1但因为fail_on_nonzero_exitFalseDAG 不会失败下游通过SsmGetCommandInvocationOperator拿到response_code后由普通 Python 任务完成分支路由。模式三增强同步模式等待完成但容忍非零退出码run_command_sync SsmRunCommandOperator( task_idrun_command_sync, document_nameAWS-RunShellScript, run_command_kwargs{ InstanceIds: [instance_id], Parameters: {commands: [echo Testing sync pattern, exit 2]}, }, wait_for_completionTrue, fail_on_nonzero_exitFalse, )同步等待完成但即使退出码为 2 也成功返回适合命令必须执行完、但结果交由业务逻辑处理的场景。三种模式在同一 DAG 中并行运行、最后统一回收资源终止实例、删除实例配置文件依赖关系通过chain(...)与运算符组织展示了与 EC2 创建/销毁操作符组合成完整闭环的编排能力。九、底层原理Hook 与 Trigger 的协作SSM 组件的底层支撑是SsmHookhooks/ssm.py它以client_typessm构造 boto3 客户端并提供三个关键方法get_parameter_value(parameter, default)读取 SSM Parameter Store 参数SecureString类型会自动解密并调用mask_secret打码避免敏感值泄漏到日志参数不存在时可返回默认值get_command_invocation(command_id, instance_id)包装get_command_invocationAPI返回单实例调用详情含输出list_command_invocations(command_id)包装list_command_invocationsAPI枚举命令在各实例上的调用。deferrable 模式的核心是SsmRunCommandTriggertriggers/ssm.py它继承自AwsBaseWaiterTrigger以command_executedWaiter 为基础run()中通过异步客户端list_command_invocations解析出所有目标实例 ID对每个实例调用async_wait(...)等待 Waiter 完成并打印状态日志异常时用get_command_invocation获取状态与ResponseCodeAWS 级失败Cancelled/TimedOut/Cancelling直接抛出命令级失败则视fail_on_nonzero_exit决定容忍continue还是产出status: failed事件全部完成后产出{status: success, command_id: ...}事件由操作符/传感器的execute_complete()恢复执行并记录日志。这一设计意味着deferrable 模式下失败不是简单的异常重抛而是通过结构化 TriggerEvent 携带command_status、exit_code、instance_id等信息返回给操作符操作符据此拼装出包含完整上下文的错误信息后抛出便于排障。十、测试与验证系统测试示例上述所有能力都由仓库中的系统测试 example_ssm.py 覆盖验证测试 DAGdag_idexample_ssm的完整生命周期包括环境准备创建 IAM 实例配置文件、用EC2CreateInstanceOperator拉起带 SSM Agent 的 EC2 实例user-data 自动安装并启动 agent、开启 IMDSv2、加密根卷并通过wait_until_ssm_ready轮询PingStatus Online命令下发与等待依次执行上文介绍的三种编排模式与退出码路由资源回收EC2TerminateInstanceOperatorTriggerRule.ALL_DONE保证无论如何都清理终止实例并删除实例配置文件。这段示例展示了如何在真实 AWS 环境中端到端验证 SSM 编排链路也是理解本文所有组件配合关系的最佳参考实现。十一、总结与最佳实践托管前提目标节点必须安装并运行 SSM Agent且能与 SSM 服务通信否则命令无法下发权限最小化至少需要ssm:SendCommand、ssm:ListCommandInvocations、ssm:GetCommandInvocation三个动作生产环境将Resource收窄到具体文档与实例 ARN等待策略短命令用wait_for_completionTrue一步到位长命令用wait_for_completionFalse Sensor 解耦追求资源效率时启用deferrableTrue需安装aiobotocore失败语义默认非零退出码即失败需要补偿/路由时设fail_on_nonzero_exitFalse并用SsmGetCommandInvocationOperator的response_code做分支决策AWS 级失败TimedOut/Cancelled始终抛错无法被容忍输出回收SsmGetCommandInvocationOperator返回结构化结果输出、错误、退出码、起止时间、文档名、备注可直接写入 XCom 供下游消费。更完整的参数说明可继续查阅 SSM 操作符文档、通用参数说明 与 系统测试示例更详细的 API 行为可参考 AWS boto3 官方文档中 Amazon SSM 服务章节。【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考