Added retry to ECS Operator

[markhopson]

This PR allows users to enable retry-behaviour for ECSOperator._start_task.

This feature was discussed briefly: here #13725

The design was lifted from hooks/base_google.py. There's 2 main differences with my AWS implementation ...

1. the user can configure the retry with a param quota_retry (Dict) that is passed in to the ECSOperator

2. the user can configure the retry without tenacity (e.g. passing in max=300 and multiplier=1 instead of tenacity.wait_exponential(multiplier=1, max=300)

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contribution Guide (https://github.com/apache/airflow/blob/master/CONTRIBUTING.rst)
Here are some useful points:

• Pay attention to the quality of your code (flake8, pylint and type annotations). Our pre-commits will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example DAG that shows how users should use it.
• Consider using Breeze environment for testing locally, it’s a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[markhopson]

@turbaszek Thanks for the feedback. I also saw some checks fail (i.e. CI Build / Postgres9.6,Py3.6, CI Build / MySQL5.7, Py3.6, CI Build / Sqlite Py3.6) but I'm not sure what to do to fix them. Can you please advise me? Thanks!

[markhopson]

@turbaszek just a friendly bump to see what the next steps are here

[ashb]

Since these are ECS specific they shouldn't really be in the base_aws module.

[markhopson]

Not sure what to do here. It feels weird to move this to ecs.py if I have code in base_aws.py that catches ECSOperatorError. For now, I've moved this variable into the is_permissible_error() scope to clarify intent. Any suggestions are welcomed.

[ashb]

Not sure about this parameter name -- it seems very specific to EcsOperator only.

[markhopson]

Good point. Hm I've changed it to retry_args to be more generic, but I'm open to any suggestions.

[markhopson]

@ashb Friendly bump here

[ashb]

Thanks, I'll take a look on Monday

[ashb]

If we change this to

defretry(should_retry: Callable[[Exception], bool], fun: Callable)

Then the retry_if_permissible_error can move out of BaseAWS in to ECSHook, used like this:

defshould_retry(exception: Exception):
"""Check if exception is related to ECS resource quota (CPU, MEM)."""returnisinstance(exception, ECSOperatorError) andany(
quota_reasoninfailure['reason']
forquota_reasonin ['RESOURCE:MEMORY', 'RESOURCE:CPU']
forfailureinexception.failures
)
...
@AwsBaseHook.retry(should_retry)def_start_task(self):

[markhopson]

I wish I thought of this. I've made this change.

[ashb]

Suggested change

	defdecorator_f(self):
	@functools.wraps
	defdecorator_f(self, *args, **kwargs):

[markhopson]

Done.

[ashb]

@markhopson See the review I just left -- I think that gives us a way to have the base retry functionality in the base hook, without having to put any service specific login in there. WDYT?

[markhopson]

@markhopson See the review I just left -- I think that gives us a way to have the base retry functionality in the base hook, without having to put any service specific login in there. WDYT?

@ashb Good tip. I like it. I've made the change.

[github-actions]

The Workflow run is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.

[github-actions]

The Workflow run is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.

[markhopson]

@ashb Friendly bump.

[ashb]

Code looks good @markhopson

Could you fix up the static checks please?

[github-actions]

The PR is likely OK to be merged with just subset of tests for default Python and Database versions without running the full matrix of tests, because it does not modify the core of Airflow. If the committers decide that the full tests matrix is needed, they will add the label 'full tests needed'. Then you should rebase to the latest master or amend the last commit of the PR, and push it with --force-with-lease.

[ashb]

TestECSOperator.test_execute_with_failures is now failing becuase the type of the exception it throws has changed -- I think that's fine, and the tests just need updating.

I haven't look in detail at the code/test though.

[boring-cyborg]

Awesome work, congrats on your first merged pull request!

[markhopson]

This PR allows users to enable retry-behaviour for ECSOperator._start_task.

This feature was discussed briefly: here #13725

The design was lifted from hooks/base_google.py. There's 2 main differences with my AWS implementation ...

1. the user can configure the retry with a param quota_retry (Dict) that is passed in to the ECSOperator

2. the user can configure the retry without tenacity (e.g. passing in max=300 and multiplier=1 instead of tenacity.wait_exponential(multiplier=1, max=300)

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contribution Guide (https://github.com/apache/airflow/blob/master/CONTRIBUTING.rst)
Here are some useful points:

• Pay attention to the quality of your code (flake8, pylint and type annotations). Our pre-commits will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example DAG that shows how users should use it.
• Consider using Breeze environment for testing locally, it’s a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[markhopson]

@turbaszek Thanks for the feedback. I also saw some checks fail (i.e. CI Build / Postgres9.6,Py3.6, CI Build / MySQL5.7, Py3.6, CI Build / Sqlite Py3.6) but I'm not sure what to do to fix them. Can you please advise me? Thanks!

[markhopson]

@turbaszek just a friendly bump to see what the next steps are here

[ashb]

Since these are ECS specific they shouldn't really be in the base_aws module.

[markhopson]

Not sure what to do here. It feels weird to move this to ecs.py if I have code in base_aws.py that catches ECSOperatorError. For now, I've moved this variable into the is_permissible_error() scope to clarify intent. Any suggestions are welcomed.

[ashb]

Not sure about this parameter name -- it seems very specific to EcsOperator only.

[markhopson]

Good point. Hm I've changed it to retry_args to be more generic, but I'm open to any suggestions.

[markhopson]

@ashb Friendly bump here

[ashb]

Thanks, I'll take a look on Monday

[ashb]

If we change this to

defretry(should_retry: Callable[[Exception], bool], fun: Callable)

Then the retry_if_permissible_error can move out of BaseAWS in to ECSHook, used like this:

defshould_retry(exception: Exception):
"""Check if exception is related to ECS resource quota (CPU, MEM)."""returnisinstance(exception, ECSOperatorError) andany(
quota_reasoninfailure['reason']
forquota_reasonin ['RESOURCE:MEMORY', 'RESOURCE:CPU']
forfailureinexception.failures
)
...
@AwsBaseHook.retry(should_retry)def_start_task(self):

[markhopson]

I wish I thought of this. I've made this change.

[ashb]

Suggested change

	defdecorator_f(self):
	@functools.wraps
	defdecorator_f(self, *args, **kwargs):

[markhopson]

Done.

[ashb]

@markhopson See the review I just left -- I think that gives us a way to have the base retry functionality in the base hook, without having to put any service specific login in there. WDYT?

[markhopson]

@markhopson See the review I just left -- I think that gives us a way to have the base retry functionality in the base hook, without having to put any service specific login in there. WDYT?

@ashb Good tip. I like it. I've made the change.

[github-actions]

The Workflow run is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.

[github-actions]

The Workflow run is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.

[markhopson]

@ashb Friendly bump.

[ashb]

Code looks good @markhopson

Could you fix up the static checks please?

[github-actions]

The PR is likely OK to be merged with just subset of tests for default Python and Database versions without running the full matrix of tests, because it does not modify the core of Airflow. If the committers decide that the full tests matrix is needed, they will add the label 'full tests needed'. Then you should rebase to the latest master or amend the last commit of the PR, and push it with --force-with-lease.

[ashb]

TestECSOperator.test_execute_with_failures is now failing becuase the type of the exception it throws has changed -- I think that's fine, and the tests just need updating.

I haven't look in detail at the code/test though.

[boring-cyborg]

Awesome work, congrats on your first merged pull request!

[markhopson]

This PR allows users to enable retry-behaviour for ECSOperator._start_task.

This feature was discussed briefly: here #13725

The design was lifted from hooks/base_google.py. There's 2 main differences with my AWS implementation ...

1. the user can configure the retry with a param quota_retry (Dict) that is passed in to the ECSOperator

2. the user can configure the retry without tenacity (e.g. passing in max=300 and multiplier=1 instead of tenacity.wait_exponential(multiplier=1, max=300)

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contribution Guide (https://github.com/apache/airflow/blob/master/CONTRIBUTING.rst)
Here are some useful points:

• Pay attention to the quality of your code (flake8, pylint and type annotations). Our pre-commits will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example DAG that shows how users should use it.
• Consider using Breeze environment for testing locally, it’s a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[markhopson]

@turbaszek Thanks for the feedback. I also saw some checks fail (i.e. CI Build / Postgres9.6,Py3.6, CI Build / MySQL5.7, Py3.6, CI Build / Sqlite Py3.6) but I'm not sure what to do to fix them. Can you please advise me? Thanks!

[markhopson]

@turbaszek just a friendly bump to see what the next steps are here

[ashb]

Since these are ECS specific they shouldn't really be in the base_aws module.

[markhopson]

Not sure what to do here. It feels weird to move this to ecs.py if I have code in base_aws.py that catches ECSOperatorError. For now, I've moved this variable into the is_permissible_error() scope to clarify intent. Any suggestions are welcomed.

[ashb]

Not sure about this parameter name -- it seems very specific to EcsOperator only.

[markhopson]

Good point. Hm I've changed it to retry_args to be more generic, but I'm open to any suggestions.

[markhopson]

@ashb Friendly bump here

[ashb]

Thanks, I'll take a look on Monday

[ashb]

If we change this to

defretry(should_retry: Callable[[Exception], bool], fun: Callable)

Then the retry_if_permissible_error can move out of BaseAWS in to ECSHook, used like this:

defshould_retry(exception: Exception):
"""Check if exception is related to ECS resource quota (CPU, MEM)."""returnisinstance(exception, ECSOperatorError) andany(
quota_reasoninfailure['reason']
forquota_reasonin ['RESOURCE:MEMORY', 'RESOURCE:CPU']
forfailureinexception.failures
)
...
@AwsBaseHook.retry(should_retry)def_start_task(self):

[markhopson]

I wish I thought of this. I've made this change.

[ashb]

Suggested change

	defdecorator_f(self):
	@functools.wraps
	defdecorator_f(self, *args, **kwargs):

[markhopson]

Done.

[ashb]

@markhopson See the review I just left -- I think that gives us a way to have the base retry functionality in the base hook, without having to put any service specific login in there. WDYT?

[markhopson]

@markhopson See the review I just left -- I think that gives us a way to have the base retry functionality in the base hook, without having to put any service specific login in there. WDYT?

@ashb Good tip. I like it. I've made the change.

[github-actions]

The Workflow run is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.

[github-actions]

The Workflow run is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.

[markhopson]

@ashb Friendly bump.

[ashb]

Code looks good @markhopson

Could you fix up the static checks please?

[github-actions]

The PR is likely OK to be merged with just subset of tests for default Python and Database versions without running the full matrix of tests, because it does not modify the core of Airflow. If the committers decide that the full tests matrix is needed, they will add the label 'full tests needed'. Then you should rebase to the latest master or amend the last commit of the PR, and push it with --force-with-lease.

[ashb]

TestECSOperator.test_execute_with_failures is now failing becuase the type of the exception it throws has changed -- I think that's fine, and the tests just need updating.

I haven't look in detail at the code/test though.

[boring-cyborg]

Awesome work, congrats on your first merged pull request!

[markhopson]

This PR allows users to enable retry-behaviour for ECSOperator._start_task.

This feature was discussed briefly: here #13725

The design was lifted from hooks/base_google.py. There's 2 main differences with my AWS implementation ...

1. the user can configure the retry with a param quota_retry (Dict) that is passed in to the ECSOperator

2. the user can configure the retry without tenacity (e.g. passing in max=300 and multiplier=1 instead of tenacity.wait_exponential(multiplier=1, max=300)

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contribution Guide (https://github.com/apache/airflow/blob/master/CONTRIBUTING.rst)
Here are some useful points:

• Pay attention to the quality of your code (flake8, pylint and type annotations). Our pre-commits will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example DAG that shows how users should use it.
• Consider using Breeze environment for testing locally, it’s a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[markhopson]

@turbaszek Thanks for the feedback. I also saw some checks fail (i.e. CI Build / Postgres9.6,Py3.6, CI Build / MySQL5.7, Py3.6, CI Build / Sqlite Py3.6) but I'm not sure what to do to fix them. Can you please advise me? Thanks!

[markhopson]

@turbaszek just a friendly bump to see what the next steps are here

[ashb]

Since these are ECS specific they shouldn't really be in the base_aws module.

[markhopson]

Not sure what to do here. It feels weird to move this to ecs.py if I have code in base_aws.py that catches ECSOperatorError. For now, I've moved this variable into the is_permissible_error() scope to clarify intent. Any suggestions are welcomed.

[ashb]

Not sure about this parameter name -- it seems very specific to EcsOperator only.

[markhopson]

Good point. Hm I've changed it to retry_args to be more generic, but I'm open to any suggestions.

[markhopson]

@ashb Friendly bump here

[ashb]

Thanks, I'll take a look on Monday

[ashb]

If we change this to

defretry(should_retry: Callable[[Exception], bool], fun: Callable)

Then the retry_if_permissible_error can move out of BaseAWS in to ECSHook, used like this:

defshould_retry(exception: Exception):
"""Check if exception is related to ECS resource quota (CPU, MEM)."""returnisinstance(exception, ECSOperatorError) andany(
quota_reasoninfailure['reason']
forquota_reasonin ['RESOURCE:MEMORY', 'RESOURCE:CPU']
forfailureinexception.failures
)
...
@AwsBaseHook.retry(should_retry)def_start_task(self):

[markhopson]

I wish I thought of this. I've made this change.

[ashb]

Suggested change

	defdecorator_f(self):
	@functools.wraps
	defdecorator_f(self, *args, **kwargs):

[markhopson]

Done.

[ashb]

@markhopson See the review I just left -- I think that gives us a way to have the base retry functionality in the base hook, without having to put any service specific login in there. WDYT?

[markhopson]

@markhopson See the review I just left -- I think that gives us a way to have the base retry functionality in the base hook, without having to put any service specific login in there. WDYT?

@ashb Good tip. I like it. I've made the change.

[github-actions]

The Workflow run is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.

[github-actions]

The Workflow run is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.

[markhopson]

@ashb Friendly bump.

[ashb]

Code looks good @markhopson

Could you fix up the static checks please?

[github-actions]

The PR is likely OK to be merged with just subset of tests for default Python and Database versions without running the full matrix of tests, because it does not modify the core of Airflow. If the committers decide that the full tests matrix is needed, they will add the label 'full tests needed'. Then you should rebase to the latest master or amend the last commit of the PR, and push it with --force-with-lease.

[ashb]

TestECSOperator.test_execute_with_failures is now failing becuase the type of the exception it throws has changed -- I think that's fine, and the tests just need updating.

I haven't look in detail at the code/test though.

[boring-cyborg]

Awesome work, congrats on your first merged pull request!

[markhopson]

This PR allows users to enable retry-behaviour for ECSOperator._start_task.

This feature was discussed briefly: here #13725

The design was lifted from hooks/base_google.py. There's 2 main differences with my AWS implementation ...

1. the user can configure the retry with a param quota_retry (Dict) that is passed in to the ECSOperator

2. the user can configure the retry without tenacity (e.g. passing in max=300 and multiplier=1 instead of tenacity.wait_exponential(multiplier=1, max=300)

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contribution Guide (https://github.com/apache/airflow/blob/master/CONTRIBUTING.rst)
Here are some useful points:

• Pay attention to the quality of your code (flake8, pylint and type annotations). Our pre-commits will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example DAG that shows how users should use it.
• Consider using Breeze environment for testing locally, it’s a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[markhopson]

@turbaszek Thanks for the feedback. I also saw some checks fail (i.e. CI Build / Postgres9.6,Py3.6, CI Build / MySQL5.7, Py3.6, CI Build / Sqlite Py3.6) but I'm not sure what to do to fix them. Can you please advise me? Thanks!

[markhopson]

@turbaszek just a friendly bump to see what the next steps are here

[ashb]

Since these are ECS specific they shouldn't really be in the base_aws module.

[markhopson]

Not sure what to do here. It feels weird to move this to ecs.py if I have code in base_aws.py that catches ECSOperatorError. For now, I've moved this variable into the is_permissible_error() scope to clarify intent. Any suggestions are welcomed.

[ashb]

Not sure about this parameter name -- it seems very specific to EcsOperator only.

[markhopson]

Good point. Hm I've changed it to retry_args to be more generic, but I'm open to any suggestions.

[markhopson]

@ashb Friendly bump here

[ashb]

Thanks, I'll take a look on Monday

[ashb]

If we change this to

defretry(should_retry: Callable[[Exception], bool], fun: Callable)

Then the retry_if_permissible_error can move out of BaseAWS in to ECSHook, used like this:

defshould_retry(exception: Exception):
"""Check if exception is related to ECS resource quota (CPU, MEM)."""returnisinstance(exception, ECSOperatorError) andany(
quota_reasoninfailure['reason']
forquota_reasonin ['RESOURCE:MEMORY', 'RESOURCE:CPU']
forfailureinexception.failures
)
...
@AwsBaseHook.retry(should_retry)def_start_task(self):

[markhopson]

I wish I thought of this. I've made this change.

[ashb]

Suggested change

	defdecorator_f(self):
	@functools.wraps
	defdecorator_f(self, *args, **kwargs):

[markhopson]

Done.

[ashb]

@markhopson See the review I just left -- I think that gives us a way to have the base retry functionality in the base hook, without having to put any service specific login in there. WDYT?

[markhopson]

@markhopson See the review I just left -- I think that gives us a way to have the base retry functionality in the base hook, without having to put any service specific login in there. WDYT?

@ashb Good tip. I like it. I've made the change.

[github-actions]

The Workflow run is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.

[github-actions]

The Workflow run is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.

[markhopson]

@ashb Friendly bump.

[ashb]

Code looks good @markhopson

Could you fix up the static checks please?

[github-actions]

The PR is likely OK to be merged with just subset of tests for default Python and Database versions without running the full matrix of tests, because it does not modify the core of Airflow. If the committers decide that the full tests matrix is needed, they will add the label 'full tests needed'. Then you should rebase to the latest master or amend the last commit of the PR, and push it with --force-with-lease.

[ashb]

TestECSOperator.test_execute_with_failures is now failing becuase the type of the exception it throws has changed -- I think that's fine, and the tests just need updating.

I haven't look in detail at the code/test though.

[boring-cyborg]

Awesome work, congrats on your first merged pull request!

[markhopson]

This PR allows users to enable retry-behaviour for ECSOperator._start_task.

This feature was discussed briefly: here #13725

The design was lifted from hooks/base_google.py. There's 2 main differences with my AWS implementation ...

1. the user can configure the retry with a param quota_retry (Dict) that is passed in to the ECSOperator

2. the user can configure the retry without tenacity (e.g. passing in max=300 and multiplier=1 instead of tenacity.wait_exponential(multiplier=1, max=300)

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contribution Guide (https://github.com/apache/airflow/blob/master/CONTRIBUTING.rst)
Here are some useful points:

• Pay attention to the quality of your code (flake8, pylint and type annotations). Our pre-commits will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example DAG that shows how users should use it.
• Consider using Breeze environment for testing locally, it’s a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[markhopson]

@turbaszek Thanks for the feedback. I also saw some checks fail (i.e. CI Build / Postgres9.6,Py3.6, CI Build / MySQL5.7, Py3.6, CI Build / Sqlite Py3.6) but I'm not sure what to do to fix them. Can you please advise me? Thanks!

[markhopson]

@turbaszek just a friendly bump to see what the next steps are here

[ashb]

Since these are ECS specific they shouldn't really be in the base_aws module.

[markhopson]

Not sure what to do here. It feels weird to move this to ecs.py if I have code in base_aws.py that catches ECSOperatorError. For now, I've moved this variable into the is_permissible_error() scope to clarify intent. Any suggestions are welcomed.

[ashb]

Not sure about this parameter name -- it seems very specific to EcsOperator only.

[markhopson]

Good point. Hm I've changed it to retry_args to be more generic, but I'm open to any suggestions.

[markhopson]

@ashb Friendly bump here

[ashb]

Thanks, I'll take a look on Monday

[ashb]

If we change this to

defretry(should_retry: Callable[[Exception], bool], fun: Callable)

Then the retry_if_permissible_error can move out of BaseAWS in to ECSHook, used like this:

defshould_retry(exception: Exception):
"""Check if exception is related to ECS resource quota (CPU, MEM)."""returnisinstance(exception, ECSOperatorError) andany(
quota_reasoninfailure['reason']
forquota_reasonin ['RESOURCE:MEMORY', 'RESOURCE:CPU']
forfailureinexception.failures
)
...
@AwsBaseHook.retry(should_retry)def_start_task(self):

[markhopson]

I wish I thought of this. I've made this change.

[ashb]

Suggested change

	defdecorator_f(self):
	@functools.wraps
	defdecorator_f(self, *args, **kwargs):

[markhopson]

Done.

[ashb]

@markhopson See the review I just left -- I think that gives us a way to have the base retry functionality in the base hook, without having to put any service specific login in there. WDYT?

[markhopson]

@markhopson See the review I just left -- I think that gives us a way to have the base retry functionality in the base hook, without having to put any service specific login in there. WDYT?

@ashb Good tip. I like it. I've made the change.

[github-actions]

The Workflow run is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.

[github-actions]

The Workflow run is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.

[markhopson]

@ashb Friendly bump.

[ashb]

Code looks good @markhopson

Could you fix up the static checks please?

[github-actions]

The PR is likely OK to be merged with just subset of tests for default Python and Database versions without running the full matrix of tests, because it does not modify the core of Airflow. If the committers decide that the full tests matrix is needed, they will add the label 'full tests needed'. Then you should rebase to the latest master or amend the last commit of the PR, and push it with --force-with-lease.

[ashb]

TestECSOperator.test_execute_with_failures is now failing becuase the type of the exception it throws has changed -- I think that's fine, and the tests just need updating.

I haven't look in detail at the code/test though.

[boring-cyborg]

Awesome work, congrats on your first merged pull request!

[markhopson]

This PR allows users to enable retry-behaviour for ECSOperator._start_task.

This feature was discussed briefly: here #13725

The design was lifted from hooks/base_google.py. There's 2 main differences with my AWS implementation ...

1. the user can configure the retry with a param quota_retry (Dict) that is passed in to the ECSOperator

2. the user can configure the retry without tenacity (e.g. passing in max=300 and multiplier=1 instead of tenacity.wait_exponential(multiplier=1, max=300)

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contribution Guide (https://github.com/apache/airflow/blob/master/CONTRIBUTING.rst)
Here are some useful points:

• Pay attention to the quality of your code (flake8, pylint and type annotations). Our pre-commits will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example DAG that shows how users should use it.
• Consider using Breeze environment for testing locally, it’s a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[markhopson]

@turbaszek Thanks for the feedback. I also saw some checks fail (i.e. CI Build / Postgres9.6,Py3.6, CI Build / MySQL5.7, Py3.6, CI Build / Sqlite Py3.6) but I'm not sure what to do to fix them. Can you please advise me? Thanks!

[markhopson]

@turbaszek just a friendly bump to see what the next steps are here

[ashb]

Since these are ECS specific they shouldn't really be in the base_aws module.

[markhopson]

Not sure what to do here. It feels weird to move this to ecs.py if I have code in base_aws.py that catches ECSOperatorError. For now, I've moved this variable into the is_permissible_error() scope to clarify intent. Any suggestions are welcomed.

[ashb]

Not sure about this parameter name -- it seems very specific to EcsOperator only.

[markhopson]

Good point. Hm I've changed it to retry_args to be more generic, but I'm open to any suggestions.

[markhopson]

@ashb Friendly bump here

[ashb]

Thanks, I'll take a look on Monday

[ashb]

If we change this to

defretry(should_retry: Callable[[Exception], bool], fun: Callable)

Then the retry_if_permissible_error can move out of BaseAWS in to ECSHook, used like this:

defshould_retry(exception: Exception):
"""Check if exception is related to ECS resource quota (CPU, MEM)."""returnisinstance(exception, ECSOperatorError) andany(
quota_reasoninfailure['reason']
forquota_reasonin ['RESOURCE:MEMORY', 'RESOURCE:CPU']
forfailureinexception.failures
)
...
@AwsBaseHook.retry(should_retry)def_start_task(self):

[markhopson]

I wish I thought of this. I've made this change.

[ashb]

Suggested change

	defdecorator_f(self):
	@functools.wraps
	defdecorator_f(self, *args, **kwargs):

[markhopson]

Done.

[ashb]

@markhopson See the review I just left -- I think that gives us a way to have the base retry functionality in the base hook, without having to put any service specific login in there. WDYT?

[markhopson]

@markhopson See the review I just left -- I think that gives us a way to have the base retry functionality in the base hook, without having to put any service specific login in there. WDYT?

@ashb Good tip. I like it. I've made the change.

[github-actions]

The Workflow run is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.

[github-actions]

The Workflow run is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.

[markhopson]

@ashb Friendly bump.

[ashb]

Code looks good @markhopson

Could you fix up the static checks please?

[github-actions]

The PR is likely OK to be merged with just subset of tests for default Python and Database versions without running the full matrix of tests, because it does not modify the core of Airflow. If the committers decide that the full tests matrix is needed, they will add the label 'full tests needed'. Then you should rebase to the latest master or amend the last commit of the PR, and push it with --force-with-lease.

[ashb]

TestECSOperator.test_execute_with_failures is now failing becuase the type of the exception it throws has changed -- I think that's fine, and the tests just need updating.

I haven't look in detail at the code/test though.

[boring-cyborg]

Awesome work, congrats on your first merged pull request!

[markhopson]

This PR allows users to enable retry-behaviour for ECSOperator._start_task.

This feature was discussed briefly: here #13725

The design was lifted from hooks/base_google.py. There's 2 main differences with my AWS implementation ...

1. the user can configure the retry with a param quota_retry (Dict) that is passed in to the ECSOperator

2. the user can configure the retry without tenacity (e.g. passing in max=300 and multiplier=1 instead of tenacity.wait_exponential(multiplier=1, max=300)

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contribution Guide (https://github.com/apache/airflow/blob/master/CONTRIBUTING.rst)
Here are some useful points:

• Pay attention to the quality of your code (flake8, pylint and type annotations). Our pre-commits will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example DAG that shows how users should use it.
• Consider using Breeze environment for testing locally, it’s a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[markhopson]

@turbaszek Thanks for the feedback. I also saw some checks fail (i.e. CI Build / Postgres9.6,Py3.6, CI Build / MySQL5.7, Py3.6, CI Build / Sqlite Py3.6) but I'm not sure what to do to fix them. Can you please advise me? Thanks!

[markhopson]

@turbaszek just a friendly bump to see what the next steps are here

[ashb]

Since these are ECS specific they shouldn't really be in the base_aws module.

[markhopson]

Not sure what to do here. It feels weird to move this to ecs.py if I have code in base_aws.py that catches ECSOperatorError. For now, I've moved this variable into the is_permissible_error() scope to clarify intent. Any suggestions are welcomed.

[ashb]

Not sure about this parameter name -- it seems very specific to EcsOperator only.

[markhopson]

Good point. Hm I've changed it to retry_args to be more generic, but I'm open to any suggestions.

[markhopson]

@ashb Friendly bump here

[ashb]

Thanks, I'll take a look on Monday

[ashb]

If we change this to

defretry(should_retry: Callable[[Exception], bool], fun: Callable)

Then the retry_if_permissible_error can move out of BaseAWS in to ECSHook, used like this:

defshould_retry(exception: Exception):
"""Check if exception is related to ECS resource quota (CPU, MEM)."""returnisinstance(exception, ECSOperatorError) andany(
quota_reasoninfailure['reason']
forquota_reasonin ['RESOURCE:MEMORY', 'RESOURCE:CPU']
forfailureinexception.failures
)
...
@AwsBaseHook.retry(should_retry)def_start_task(self):

[markhopson]

I wish I thought of this. I've made this change.

[ashb]

Suggested change

	defdecorator_f(self):
	@functools.wraps
	defdecorator_f(self, *args, **kwargs):

[markhopson]

Done.

[ashb]

@markhopson See the review I just left -- I think that gives us a way to have the base retry functionality in the base hook, without having to put any service specific login in there. WDYT?

[markhopson]

@markhopson See the review I just left -- I think that gives us a way to have the base retry functionality in the base hook, without having to put any service specific login in there. WDYT?

@ashb Good tip. I like it. I've made the change.

[github-actions]

The Workflow run is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.

[github-actions]

The Workflow run is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.

[markhopson]

@ashb Friendly bump.

[ashb]

Code looks good @markhopson

Could you fix up the static checks please?

[github-actions]

The PR is likely OK to be merged with just subset of tests for default Python and Database versions without running the full matrix of tests, because it does not modify the core of Airflow. If the committers decide that the full tests matrix is needed, they will add the label 'full tests needed'. Then you should rebase to the latest master or amend the last commit of the PR, and push it with --force-with-lease.

[ashb]

TestECSOperator.test_execute_with_failures is now failing becuase the type of the exception it throws has changed -- I think that's fine, and the tests just need updating.

I haven't look in detail at the code/test though.

[boring-cyborg]

Awesome work, congrats on your first merged pull request!