Unnecessary dependence between capacity_provider_strategy and volume_configurations

[pyrr]

Apache Airflow Provider(s)

amazon

Versions of Apache Airflow Providers

apache-airflow-providers-amazon==8.29.0

Apache Airflow version

2.9.2

Operating System

Amazon Linux 2023.5.20240805

Deployment

Amazon (AWS) MWAA

Deployment details

Using a Python 3.11 environment with botocore==1.34.106 and apache-airflow-providers-amazon==8.29.0.

This allows us to use the volume_configurations feature so we can deploy DAGs that use the ECSRunTaskOperator with EBS volume attachments configured at launch.

What happened

Omitting capacity_provider_strategy from EcsRunTaskOperator prevents volume_configurations from being passed to self.client.run_task(**run_opts) in EcsRunTaskOperator._start_task().

So, if you aren't using capacity_provider_strategy (in this case we are using launch_type="FARGATE") this prevents an EBS volume configuration from being defined which throws the following error during DAG execution:

Traceback (most recent call last):
  File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 2479, in _run_raw_task
    self._execute_task_with_callbacks(context, test_mode, session=session)
  File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 2676, in _execute_task_with_callbacks
    result = self._execute_task(context, task_orig)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 2701, in _execute_task
    return _execute_task(self, context, task_orig)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 465, in _execute_task
    result = _execute_callable(context=context, **execute_callable_kwargs)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 432, in _execute_callable
    return execute_callable(context=context, **execute_callable_kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/models/baseoperator.py", line 401, in wrapper
    return func(self, *args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/providers/amazon/aws/operators/ecs.py", line 543, in execute
    self._start_task()
  File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/providers/amazon/aws/operators/ecs.py", line 643, in _start_task
    response = self.client.run_task(**run_opts)
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/airflow/.local/lib/python3.11/site-packages/botocore/client.py", line 565, in _api_call
    return self._make_api_call(operation_name, kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/airflow/.local/lib/python3.11/site-packages/botocore/client.py", line 1021, in _make_api_call
    raise error_class(parsed_response, operation_name)
botocore.errorfactory.InvalidParameterException: An error occurred (InvalidParameterException) when calling the RunTask operation: The volume name '<volume_name>' in your request does not have a matching volume in your Task Definition that is configurable at launch.

What you think should happen instead

EcsRunTaskOperator.volume_configurations should be passed to self.client.run_task(**run_opts) regardless of the value of capacity_provider_strategy.

How to reproduce

1. Create an ECS task definition that includes volume and mountPoint definitions for an EBS volume.
2. Run a task using ECSRunTaskOperator with volume_configurations containing a managedEBSVolume defined and capacity_provider_strategy undefined.
3. Try to run the corresponding DAG with an EBS volume attached.

Anything else

• Directly calling ECS.Client.run_task with the boto3 API with the same configuration passed to EcsRunTaskOperator (with camelCase keywords) works.
• Replacing ecs.py in my deployment with the code in Decouple volume_configurations from capacity_provider_strategy #43047 fixes the issue and allows us to run these tasks.

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.