You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "ksharlandjiev (via GitHub)" <gi...@apache.org> on 2025/11/03 15:54:04 UTC
[PR] Add fail_on_nonzero_exit parameter to SSM operators for exit code routing [airflow]
ksharlandjiev opened a new pull request, #57753:
URL: https://github.com/apache/airflow/pull/57753
## Problem
SSM operators currently fail when commands return non-zero exit codes, making it impossible to:
- Route workflows based on different exit codes
- Handle commands where non-zero exit codes represent valid business states (e.g., partial success, warnings)
- Implement conditional retry logic based on specific exit codes
- Migrate from traditional schedulers like Autosys that support exit code routing
Users have been forced to implement manual polling workarounds with custom Python tasks to handle these scenarios.
## Proposal
Add a `fail_on_nonzero_exit` parameter (default: `True`) to `SsmRunCommandOperator`, `SsmRunCommandCompletedSensor`, and `SsmRunCommandTrigger`.
When set to `False`:
- Tasks complete successfully regardless of command exit codes
- Exit codes can be retrieved with `SsmGetCommandInvocationOperator` for routing decisions
- AWS-level failures (TimedOut, Cancelled) still raise exceptions
- Command-level failures (non-zero exit codes) are tolerated
The default value of `True` maintains existing behavior for backward compatibility.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add fail_on_nonzero_exit parameter to SSM operators for exit code routing [airflow]
Posted by "ksharlandjiev (via GitHub)" <gi...@apache.org>.
ksharlandjiev commented on PR #57753:
URL: https://github.com/apache/airflow/pull/57753#issuecomment-3482791984
> My general comments:
>
> * I think the idea is great, that is indeed a feature user might want. Thanks for creating a PR for that
>
> * Creating documentation is great, however, I think the document is way too long. This is only my personal opinion so I would wait to see what others think but if for new parameter we are creating documentation that big, this will be impossible to maintain. Using AI to create code and/or documentation is great but we should also keep in mind, the longer is NOT the better. Again, I support the documentation, but this is way too big to me. As a user I wont probably read it all, and as a developer I am scared we need to maintain that
>
> * The system test is a great idea, could you please move these 3 examples in the current system test?
Thanks for your feedback. I was on the fence myself on the extra docs, and I understand the concern. I'm happy to move all documented patterns to an external article.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add fail_on_nonzero_exit parameter to SSM operators for exit code routing [airflow]
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas commented on code in PR #57753:
URL: https://github.com/apache/airflow/pull/57753#discussion_r2492396406
##########
providers/amazon/src/airflow/providers/amazon/aws/triggers/ssm.py:
##########
@@ -89,14 +94,47 @@ async def run(self) -> AsyncIterator[TriggerEvent]:
for instance_id in instance_ids:
self.waiter_args["InstanceId"] = instance_id
- await async_wait(
- waiter,
- self.waiter_delay,
- self.attempts,
- self.waiter_args,
- self.failure_message,
- self.status_message,
- self.status_queries,
- )
+ try:
+ await async_wait(
+ waiter,
+ self.waiter_delay,
+ self.attempts,
+ self.waiter_args,
+ self.failure_message,
+ self.status_message,
+ self.status_queries,
+ )
+ except Exception:
Review Comment:
It looks like you do this pattern three separate times with lots of duplicated log strings, state sets/tuples, etc. Can this logic be extracted out into a helper method that's defined once and used three times?
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/ssm.py:
##########
@@ -132,14 +143,36 @@ def execute(self, context: Context):
instance_ids = response["Command"]["InstanceIds"]
for instance_id in instance_ids:
- waiter.wait(
- CommandId=command_id,
- InstanceId=instance_id,
- WaiterConfig={
- "Delay": self.waiter_delay,
- "MaxAttempts": self.waiter_max_attempts,
- },
- )
+ try:
+ waiter.wait(
+ CommandId=command_id,
+ InstanceId=instance_id,
+ WaiterConfig={
+ "Delay": self.waiter_delay,
+ "MaxAttempts": self.waiter_max_attempts,
+ },
+ )
+ except WaiterError:
+ if not self.fail_on_nonzero_exit:
+ # Enhanced mode: distinguish between AWS-level and command-level failures
+ invocation = self.hook.get_command_invocation(command_id, instance_id)
+ status = invocation.get("Status", "")
+
+ # AWS-level failures should always raise
+ if status in ("Cancelled", "TimedOut"):
+ self.log.error("Command failed with AWS-level error: %s", status)
Review Comment:
We probably don't need this? It fails today without a log when these situations happen.
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/ssm.py:
##########
@@ -132,14 +143,36 @@ def execute(self, context: Context):
instance_ids = response["Command"]["InstanceIds"]
for instance_id in instance_ids:
- waiter.wait(
- CommandId=command_id,
- InstanceId=instance_id,
- WaiterConfig={
- "Delay": self.waiter_delay,
- "MaxAttempts": self.waiter_max_attempts,
- },
- )
+ try:
+ waiter.wait(
+ CommandId=command_id,
+ InstanceId=instance_id,
+ WaiterConfig={
+ "Delay": self.waiter_delay,
+ "MaxAttempts": self.waiter_max_attempts,
+ },
+ )
+ except WaiterError:
+ if not self.fail_on_nonzero_exit:
+ # Enhanced mode: distinguish between AWS-level and command-level failures
+ invocation = self.hook.get_command_invocation(command_id, instance_id)
+ status = invocation.get("Status", "")
+
+ # AWS-level failures should always raise
+ if status in ("Cancelled", "TimedOut"):
Review Comment:
Are these the only "aws-level" failures? Always cautious of static lists like this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add fail_on_nonzero_exit parameter to SSM operators for exit code routing [airflow]
Posted by "ksharlandjiev (via GitHub)" <gi...@apache.org>.
ksharlandjiev commented on PR #57753:
URL: https://github.com/apache/airflow/pull/57753#issuecomment-3639421773
> Have you run the system test to ensue that it's working correctly?
Thanks for the approval! Yes, I have also added few more tests in the main system test following @vincbeck feedback and I can confirm it executes successfully.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add fail_on_nonzero_exit parameter to SSM operators for exit code routing [airflow]
Posted by "ksharlandjiev (via GitHub)" <gi...@apache.org>.
ksharlandjiev commented on code in PR #57753:
URL: https://github.com/apache/airflow/pull/57753#discussion_r2599040923
##########
providers/amazon/src/airflow/providers/amazon/aws/triggers/ssm.py:
##########
@@ -89,14 +94,47 @@ async def run(self) -> AsyncIterator[TriggerEvent]:
for instance_id in instance_ids:
self.waiter_args["InstanceId"] = instance_id
- await async_wait(
- waiter,
- self.waiter_delay,
- self.attempts,
- self.waiter_args,
- self.failure_message,
- self.status_message,
- self.status_queries,
- )
+ try:
+ await async_wait(
+ waiter,
+ self.waiter_delay,
+ self.attempts,
+ self.waiter_args,
+ self.failure_message,
+ self.status_message,
+ self.status_queries,
+ )
+ except Exception:
Review Comment:
done. I've removed the code duplication across operator, sensor, and trigger
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add fail_on_nonzero_exit parameter to SSM operators for exit code routing [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck merged PR #57753:
URL: https://github.com/apache/airflow/pull/57753
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add fail_on_nonzero_exit parameter to SSM operators for exit code routing [airflow]
Posted by "ksharlandjiev (via GitHub)" <gi...@apache.org>.
ksharlandjiev commented on code in PR #57753:
URL: https://github.com/apache/airflow/pull/57753#discussion_r2599037219
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/ssm.py:
##########
@@ -132,14 +143,36 @@ def execute(self, context: Context):
instance_ids = response["Command"]["InstanceIds"]
for instance_id in instance_ids:
- waiter.wait(
- CommandId=command_id,
- InstanceId=instance_id,
- WaiterConfig={
- "Delay": self.waiter_delay,
- "MaxAttempts": self.waiter_max_attempts,
- },
- )
+ try:
+ waiter.wait(
+ CommandId=command_id,
+ InstanceId=instance_id,
+ WaiterConfig={
+ "Delay": self.waiter_delay,
+ "MaxAttempts": self.waiter_max_attempts,
+ },
+ )
+ except WaiterError:
+ if not self.fail_on_nonzero_exit:
+ # Enhanced mode: distinguish between AWS-level and command-level failures
+ invocation = self.hook.get_command_invocation(command_id, instance_id)
+ status = invocation.get("Status", "")
+
+ # AWS-level failures should always raise
+ if status in ("Cancelled", "TimedOut"):
Review Comment:
Yes. I have double checked the documentation. However, to make things easier, I've added `SsmHook.is_aws_level_failure()` helper method with the statuses list.
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/ssm.py:
##########
@@ -132,14 +143,36 @@ def execute(self, context: Context):
instance_ids = response["Command"]["InstanceIds"]
for instance_id in instance_ids:
- waiter.wait(
- CommandId=command_id,
- InstanceId=instance_id,
- WaiterConfig={
- "Delay": self.waiter_delay,
- "MaxAttempts": self.waiter_max_attempts,
- },
- )
+ try:
+ waiter.wait(
+ CommandId=command_id,
+ InstanceId=instance_id,
+ WaiterConfig={
+ "Delay": self.waiter_delay,
+ "MaxAttempts": self.waiter_max_attempts,
+ },
+ )
+ except WaiterError:
+ if not self.fail_on_nonzero_exit:
+ # Enhanced mode: distinguish between AWS-level and command-level failures
+ invocation = self.hook.get_command_invocation(command_id, instance_id)
+ status = invocation.get("Status", "")
+
+ # AWS-level failures should always raise
+ if status in ("Cancelled", "TimedOut"):
+ self.log.error("Command failed with AWS-level error: %s", status)
Review Comment:
removed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add fail_on_nonzero_exit parameter to SSM operators for exit code routing [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on code in PR #57753:
URL: https://github.com/apache/airflow/pull/57753#discussion_r2668809960
##########
providers/amazon/docs/operators/ssm.rst:
##########
@@ -52,12 +52,42 @@ Waiter. Additionally, you can use the following components to track the status o
:class:`~airflow.providers.amazon.aws.sensors.ssm.SsmRunCommandCompletedSensor` Sensor,
or the :class:`~airflow.providers.amazon.aws.triggers.ssm.SsmRunCommandTrigger` Trigger.
-
.. exampleinclude:: /../../amazon/tests/system/amazon/aws/example_ssm.py
:language: python
:dedent: 4
:start-after: [START howto_operator_run_command]
:end-before: [END howto_operator_run_command]
+
+Exit Code Handling
Review Comment:
nit: I am not a big fan of capitalizing everything
```suggestion
Exit code handling
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org