Fix awslogs_stream_prefix pattern

[pyrr]

closes: #43130

---

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

[vincbeck]

Could please add/update unit test?

[pyrr]

Could please add/update unit test?

@vincbeck Should be covered by current log fetch tests because the log stream name is inherited from the function changed in the commit. Let me know if you disagree and I can add a specific test.

[vincbeck]

You're right! It might not be needed

[boring-cyborg]

Awesome work, congrats on your first merged pull request! You are invited to check our Issue Tracker for additional contributions.

[ferruzzi]

The 'run_task` task in the ECS system test [here] has started failing since this change with the following stacktrace:

INFO     airflow.task.operators.airflow.providers.amazon.aws.operators.ecs.EcsRunTaskOperator:task_log_fetcher.py:88 Cannot find log stream yet, it can take a couple of seconds to show up. If this error persists, check that the log group and stream are correct: group: /ecs_test/env00c63b78	stream: ecs/env00c63b78-container/env00c63b78-container/8fdf80d5882e4a4ab9b2523da59cb109
INFO     airflow.task.operators.airflow.providers.amazon.aws.operators.ecs.EcsRunTaskOperator:task_log_fetcher.py:88 Cannot find log stream yet, it can take a couple of seconds to show up. If this error persists, check that the log group and stream are correct: group: /ecs_test/env00c63b78	stream: ecs/env00c63b78-container/env00c63b78-container/8fdf80d5882e4a4ab9b2523da59cb109
INFO     airflow.task.operators.airflow.providers.amazon.aws.operators.ecs.EcsRunTaskOperator:ecs.py:685 ECS Task stopped, check status: {'tasks': [{'attachments': [{'id': 'a002ee5c-48c5-4b51-bc33-42adddc1b806', 'type': 'ElasticNetworkInterface', 'status': 'DELETED', 'details': [{'name': 'subnetId', 'value': 'subnet-043a758ae656cd1a1'}, {'name': 'networkInterfaceId', 'value': 'eni-0876938a93c41615e'}, {'name': 'macAddress', 'value': '02:f9:6f:86:da:cf'}, {'name': 'privateDnsName', 'value': 'ip-10-0-3-89.us-west-2.compute.internal'}, {'name': 'privateIPv4Address', 'value': '10.0.3.89'}]}], 'attributes': [{'name': 'ecs.cpu-architecture', 'value': 'x86_64'}], 'availabilityZone': 'us-west-2b', 'clusterArn': 'arn:aws:ecs:us-west-2:558120655471:cluster/SysTestCluster_example_ecs', 'connectivity': 'CONNECTED', 'connectivityAt': datetime.datetime(2024, 11, 5, 23, 3, 16, 665000, tzinfo=tzlocal()), 'containerInstanceArn': 'arn:aws:ecs:us-west-2:558120655471:container-instance/SysTestCluster_example_ecs/f351c19e980943dab62ea246b00be509', 'containers': [{'containerArn': 'arn:aws:ecs:us-west-2:558120655471:container/SysTestCluster_example_ecs/8fdf80d5882e4a4ab9b2523da59cb109/f0c7840b-630e-4cdb-b5af-6400154f8a01', 'taskArn': 'arn:aws:ecs:us-west-2:558120655471:task/SysTestCluster_example_ecs/8fdf80d5882e4a4ab9b2523da59cb109', 'name': 'env00c63b78-container', 'image': 'ubuntu', 'imageDigest': 'sha256:99c35190e22d294cdace2783ac55effc69d32896daaa265f0bbedbcde4fbe3e5', 'runtimeId': 'e5a9056ce1d68042f96d7fbf2c20bad6b095c51512105c744d7d8fb447ba2b6c', 'lastStatus': 'STOPPED', 'exitCode': 0, 'networkBindings': [], 'networkInterfaces': [{'attachmentId': 'a002ee5c-48c5-4b51-bc33-42adddc1b806', 'privateIpv4Address': '10.0.3.89'}], 'healthStatus': 'UNKNOWN', 'cpu': '0'}], 'cpu': '256', 'createdAt': datetime.datetime(2024, 11, 5, 23, 3, 16, 665000, tzinfo=tzlocal()), 'desiredStatus': 'STOPPED', 'enableExecuteCommand': False, 'executionStoppedAt': datetime.datetime(2024, 11, 5, 23, 3, 29, 662000, tzinfo=tzlocal()), 'group': 'family:env00c63b78-task-definition', 'healthStatus': 'UNKNOWN', 'lastStatus': 'STOPPED', 'launchType': 'EC2', 'memory': '512', 'overrides': {'containerOverrides': [{'name': 'env00c63b78-container', 'command': ['echo hello world']}], 'inferenceAcceleratorOverrides': []}, 'pullStartedAt': datetime.datetime(2024, 11, 5, 23, 3, 29, 257000, tzinfo=tzlocal()), 'pullStoppedAt': datetime.datetime(2024, 11, 5, 23, 3, 29, 513000, tzinfo=tzlocal()), 'startedAt': datetime.datetime(2024, 11, 5, 23, 3, 29, 533000, tzinfo=tzlocal()), 'startedBy': 'airflow', 'stopCode': 'EssentialContainerExited', 'stoppedAt': datetime.datetime(2024, 11, 5, 23, 3, 43, 619000, tzinfo=tzlocal()), 'stoppedReason': 'Essential container in task exited', 'stoppingAt': datetime.datetime(2024, 11, 5, 23, 3, 30, 333000, tzinfo=tzlocal()), 'tags': [], 'taskArn': 'arn:aws:ecs:us-west-2:558120655471:task/SysTestCluster_example_ecs/8fdf80d5882e4a4ab9b2523da59cb109', 'taskDefinitionArn': 'arn:aws:ecs:us-west-2:558120655471:task-definition/env00c63b78-task-definition:1', 'version': 6}], 'failures': [], 'ResponseMetadata': {'RequestId': '7a1d07a1-71be-438d-94da-ff8738f0a8ad', 'HTTPStatusCode': 200, 'HTTPHeaders': {'x-amzn-requestid': '7a1d07a1-71be-438d-94da-ff8738f0a8ad', 'content-type': 'application/x-amz-json-1.1', 'content-length': '2434', 'date': 'Tue, 05 Nov 2024 23:04:16 GMT'}, 'RetryAttempts': 0}}

ERROR    airflow.task:taskinstance.py:3189 Task failed with exception
Traceback (most recent call last):
  File "/opt/airflow/airflow/models/taskinstance.py", line 764, in _execute_task
    result = _execute_callable(context=context, **execute_callable_kwargs)
  File "/opt/airflow/airflow/models/taskinstance.py", line 730, in _execute_callable
    return ExecutionCallableRunner(
  File "/opt/airflow/airflow/utils/operator_helpers.py", line 258, in run
    return self.func(*args, **kwargs)
  File "/opt/airflow/airflow/models/baseoperator.py", line 375, in wrapper
    return func(self, *args, **kwargs)
  File "/opt/airflow/providers/src/airflow/providers/amazon/aws/operators/ecs.py", line 563, in execute
    return self.task_log_fetcher.get_last_log_message()
  File "/opt/airflow/providers/src/airflow/providers/amazon/aws/utils/task_log_fetcher.py", line 125, in get_last_log_message
    return self.get_last_log_messages(1)[0]
  File "/opt/airflow/providers/src/airflow/providers/amazon/aws/utils/task_log_fetcher.py", line 115, in get_last_log_messages
    response = self.hook.conn.get_log_events(
  File "/usr/local/lib/python3.9/site-packages/botocore/client.py", line 569, in _api_call
    return self._make_api_call(operation_name, kwargs)
  File "/usr/local/lib/python3.9/site-packages/botocore/client.py", line 1023, in _make_api_call
    raise error_class(parsed_response, operation_name)
botocore.errorfactory.ResourceNotFoundException: An error occurred (ResourceNotFoundException) when calling the GetLogEvents operation: The specified log stream does not exist.

The test is already embedding the container name in the awslogs_stream_prefix. Should we change the test, or add some logic to check if the container name is already in the prefix?

[vincbeck]

add some logic to check if the container name is already in the prefix?

I think adding some logic to check if the container name is already in the prefix would make sense

[ferruzzi]

I agree, I think that is the safest way to handle it. @pyrr , are you able to add that?

[pyrr]

Yes, thanks for finding out the problem @ferruzzi I'll proceed with your recommendation @vincbeck. Could you re-open the issue? Alternatively I can open a new issue if preferred.

[ferruzzi]

Thanks for taking it on. I don't think you need to bother re-opening the Issue. Just cut a new PR with the fix, put "fixes PR: {link to this one}" in the description (preferably with a little context), and tag me/us in it for reviewers. We'll get a notification and get it merged.