Airflow ECS Register Task Operator, error when passing env. variables

[dominikhei]

Hello, when I am trying to pass env. variables to my ECSRegisterTaskOperator, I get the following error:

Unknown parameter in input: "environment", must be one of: family, taskRoleArn, executionRoleArn, networkMode, containerDefinitions, volumes, placementConstraints, requiresCompatibilities, cpu, memory, tags, pidMode, ipcMode

However according to the documentation for boto3 with Ecs, one has to pass env variables via the environment dictionary.
[reference: https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ecs.html#ECS.Client.register_task_definition ]

The part of my task looks like this:

register_task_kwargs={
"cpu": "256",
"memory": "512",
"networkMode": "awsvpc",
"environment": [
{
"name": "REDSHIFT_HOST",
"value": redshift_cluster_test_connection.host
},
{
"name": "REDSHIFT_USER",
"value": "redshift_cluster_test_connection.user"
},
{ "name": "REDSHIFT_PW",
"value": redshift_cluster_test_connection.password}
]

[hussein-awala]

It's a problem with the boto3 client, this parameter is not supported in your old boto3 version.

Can you check which version you are using? same for Airflow Aws provider

[dominikhei]

I have now solved it. If one looks at the documentation of the boto3 api, the environment variables have to passed in the containerDefinitions section.

However I now get the error: AWS Connection (conn_id='aws-fargate', conn_type='aws') credentials retrieved from login and password. INFO - No connection ID provided. Fallback on boto3 credential strategy (region_name=None) botocore.exceptions.NoRegionError: You must specify a region. I tried solving it but so far nothing worked, I have specified a connection_id and also a region inside the connection section in airflows web ui. Can someone maybe explain where this one came from?

[dominikhei]

My Version of the airflow-providers-amazon package is 7.1.0. I have again checked the access keys I have specified and they are correct. (I am currently developing locally, Once I run everything in production, I won't use keys and just assume the role of the underlying ec2 instance).

The full logs are:

*** Reading local file: /opt/airflow/logs/dag_id=testaa/run_id=manual__2023-02-15T10:56:41.741923+00:00/task_id=register_dbt_chart_task/attempt=1.log
[2023-02-15, 10:56:51 UTC] {taskinstance.py:1083} INFO - Dependencies all met for <TaskInstance: testaa.register_dbt_chart_task manual__2023-02-15T10:56:41.741923+00:00 [queued]>
[2023-02-15, 10:56:51 UTC] {taskinstance.py:1083} INFO - Dependencies all met for <TaskInstance: testaa.register_dbt_chart_task manual__2023-02-15T10:56:41.741923+00:00 [queued]>
[2023-02-15, 10:56:51 UTC] {taskinstance.py:1279} INFO - 
--------------------------------------------------------------------------------
[2023-02-15, 10:56:51 UTC] {taskinstance.py:1280} INFO - Starting attempt 1 of 4
[2023-02-15, 10:56:51 UTC] {taskinstance.py:1281} INFO - 
--------------------------------------------------------------------------------
[2023-02-15, 10:56:51 UTC] {taskinstance.py:1300} INFO - Executing <Task(EcsRegisterTaskDefinitionOperator): register_dbt_chart_task> on 2023-02-15 10:56:41.741923+00:00
[2023-02-15, 10:56:51 UTC] {standard_task_runner.py:55} INFO - Started process 547 to run task
[2023-02-15, 10:56:51 UTC] {standard_task_runner.py:82} INFO - Running: ['***', 'tasks', 'run', 'testaa', 'register_dbt_chart_task', 'manual__2023-02-15T10:56:41.741923+00:00', '--job-id', '329', '--raw', '--subdir', 'DAGS_FOLDER/spotify_dag/dag_2.py', '--cfg-path', '/tmp/tmpbpeh8fzx']
[2023-02-15, 10:56:51 UTC] {standard_task_runner.py:83} INFO - Job 329: Subtask register_dbt_chart_task
[2023-02-15, 10:56:52 UTC] {task_command.py:388} INFO - Running <TaskInstance: testaa.register_dbt_chart_task manual__2023-02-15T10:56:41.741923+00:00 [running]> on host *******
[2023-02-15, 10:56:52 UTC] {taskinstance.py:1509} INFO - Exporting the following env vars:
AIRFLOW_CTX_DAG_OWNER=***
AIRFLOW_CTX_DAG_ID=testaa
AIRFLOW_CTX_TASK_ID=register_dbt_chart_task
AIRFLOW_CTX_EXECUTION_DATE=2023-02-15T10:56:41.741923+00:00
AIRFLOW_CTX_TRY_NUMBER=1
AIRFLOW_CTX_DAG_RUN_ID=manual__2023-02-15T10:56:41.741923+00:00
[2023-02-15, 10:56:52 UTC] {ecs.py:244} INFO - Registering task definition dbt using the following values: {'cpu': '256', 'memory': '512', 'networkMode': 'awsvpc'}
[2023-02-15, 10:56:52 UTC] {ecs.py:246} INFO - Using container definition [{'name': '******', 'image': '*********', 'workingDirectory': '/dbt_project', 'entryPoint': ['bash', 'dbt run'], 'environment': [*****************]}]
[2023-02-15, 10:56:52 UTC] {base.py:73} INFO - Using connection ID 'aws-fargate' for task execution.
[2023-02-15, 10:56:52 UTC] {connection_wrapper.py:334} INFO - AWS Connection (conn_id='aws-fargate', conn_type='aws') credentials retrieved from login and password.
[2023-02-15, 10:57:03 UTC] {base_aws.py:133} INFO - No connection ID provided. Fallback on boto3 credential strategy (region_name=None). See: https://boto3.amazonaws.com/v1/documentation/api/latest/guide/configuration.html
[2023-02-15, 10:57:08 UTC] {taskinstance.py:1768} ERROR - Task failed with exception
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/providers/amazon/aws/operators/ecs.py", line 257, in execute
    ).poke(context):
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/providers/amazon/aws/sensors/ecs.py", line 142, in poke
    self.hook.get_task_definition_state(task_definition=self.task_definition)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/providers/amazon/aws/hooks/ecs.py", line 115, in get_task_definition_state
    return self.conn.describe_task_definition(taskDefinition=task_definition)["taskDefinition"]["status"]
  File "/home/airflow/.local/lib/python3.7/site-packages/cached_property.py", line 36, in __get__
    value = obj.__dict__[self.func.__name__] = self.func(obj)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/providers/amazon/aws/hooks/base_aws.py", line 609, in conn
    return self.get_client_type(region_name=self.region_name)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/providers/amazon/aws/hooks/base_aws.py", line 578, in get_client_type
    verify=self.verify,
  File "/home/airflow/.local/lib/python3.7/site-packages/boto3/session.py", line 309, in client
    config=config,
  File "/home/airflow/.local/lib/python3.7/site-packages/botocore/session.py", line 986, in create_client
    auth_token=auth_token,
  File "/home/airflow/.local/lib/python3.7/site-packages/botocore/client.py", line 167, in create_client
    partition_data,
  File "/home/airflow/.local/lib/python3.7/site-packages/botocore/client.py", line 497, in _get_client_args
    partition_data,
  File "/home/airflow/.local/lib/python3.7/site-packages/botocore/args.py", line 99, in get_client_args
    scoped_config,
  File "/home/airflow/.local/lib/python3.7/site-packages/botocore/args.py", line 211, in compute_client_args
    s3_config=s3_config,
  File "/home/airflow/.local/lib/python3.7/site-packages/botocore/args.py", line 313, in _compute_endpoint_config
    return self._resolve_endpoint(**resolve_endpoint_kwargs)
  File "/home/airflow/.local/lib/python3.7/site-packages/botocore/args.py", line 419, in _resolve_endpoint
    service_name, region_name, endpoint_url, is_secure
  File "/home/airflow/.local/lib/python3.7/site-packages/botocore/client.py", line 594, in resolve
    use_fips_endpoint=use_fips_endpoint,
  File "/home/airflow/.local/lib/python3.7/site-packages/botocore/regions.py", line 234, in construct_endpoint
    use_fips_endpoint,
  File "/home/airflow/.local/lib/python3.7/site-packages/botocore/regions.py", line 277, in _endpoint_for_partition
    raise NoRegionError()
botocore.exceptions.NoRegionError: You must specify a region.
[2023-02-15, 10:57:08 UTC] {taskinstance.py:1323} INFO - Marking task as UP_FOR_RETRY. dag_id=testaa, task_id=register_dbt_chart_task, execution_date=20230215T105641, start_date=20230215T105651, end_date=20230215T105708
[2023-02-15, 10:57:08 UTC] {standard_task_runner.py:105} ERROR - Failed to execute job 329 for task register_dbt_chart_task (You must specify a region.; 547)
[2023-02-15, 10:57:08 UTC] {local_task_job.py:208} INFO - Task exited with return code 1
[2023-02-15, 10:57:08 UTC] {taskinstance.py:2578} INFO - 0 downstream tasks scheduled from follow-on schedule check

[Taragolis]

Well this is a 100% a bug. As workaround I would recommend to create aws_default with the same creds if it applicable in your case

@ferruzzi I guess we need to get rid of run sensor inside of operators and use waiter for this case. In this case also aws_conn_id and region_name / region doesn't propagate to Sensor from Operator. I have a plan for create issue, and maybe when you would read this it already created (due to different timezones)

airflow/airflow/providers/amazon/aws/operators/ecs.py

Lines 112 to 118 in 2a34dc9

	while not EcsClusterStateSensor(
	task_id="await_cluster",
	cluster_name=self.cluster_name,
	).poke(context):
	# The sensor has a built-in delay and will try again until
	# the cluster is ready or has reached a failed state.
	pass

[Taragolis]

@dominikhei other workaround for not, until this not fixed and new provider released:
You could set for operator wait_for_completion to False and use EcsTaskDefinitionStateSensor for wait completion:
see ecs sensors docs

[ferruzzi]

I think you may be right., I wonder if it would work to pass aws_conn_id into the sensor, but a waiter may be a smarter solution

[ferruzzi]

Looks like boto doesn't provide a "cluster_active" waiter so we'd need to write a custom waiter using this: https://github.com/apache/airflow/tree/main/airflow/providers/amazon/aws/waiters