Copilot commented on code in PR #64745:
URL: https://github.com/apache/airflow/pull/64745#discussion_r3463472619
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -220,14 +220,20 @@ def execute(self, context: Context) -> str | None:
if self.deferrable:
if not self.job_id:
- raise AirflowException("AWS Batch job - job_id was not found")
+ raise ValueError("AWS Batch job - job_id was not found")
Review Comment:
`execute()` now raises `ValueError` when `job_id` is missing in deferrable
mode, but the docstring still claims only `AirflowException` is raised. Update
the docstring so it accurately reflects possible exceptions.
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -368,33 +398,34 @@ def monitor_job(self, context: Context):
job_queue_arn=job_queue_arn,
)
- if self.awslogs_enabled:
- if self.waiters:
- self.waiters.wait_for_job(self.job_id,
get_batch_log_fetcher=self._get_batch_log_fetcher)
- else:
- self.hook.wait_for_job(self.job_id,
get_batch_log_fetcher=self._get_batch_log_fetcher)
- else:
- if self.waiters:
- self.waiters.wait_for_job(self.job_id)
- else:
- self.hook.wait_for_job(self.job_id)
+ return job_desc
+
+ def _persist_cloudwatch_link(self, context: Context) -> None:
+ """
+ Persist CloudWatch logs link if available.
+
+ :param context: Task context
+ """
+ if not context or "ti" not in context:
+ return
+
+ if not self.job_id:
+ return
Review Comment:
`BaseAwsLink.persist()` no-ops when `operator.do_xcom_push` is False (see
`providers/amazon/.../links/base_aws.py:87-89`), but
`_persist_cloudwatch_link()` still calls `get_job_all_awslogs_info()` first.
This does extra AWS API work (and can log warnings) even though the link can
never be persisted. Add an early return when `do_xcom_push` is disabled.
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/batch.py:
##########
@@ -252,14 +258,31 @@ def execute(self, context: Context) -> str | None:
def execute_complete(self, context: Context, event: dict[str, Any] | None
= None) -> str:
validated_event = validate_execute_complete_event(event)
- if validated_event["status"] != "success":
- raise AirflowException(f"Error while running job:
{validated_event}")
-
+ # Set job_id first so CloudWatch link can be persisted even on failure
self.job_id = validated_event["job_id"]
# Fetch logs if awslogs_enabled
if self.awslogs_enabled:
- self.monitor_job(context) # fetch logs, no need to return
+ log_fetcher = self._get_batch_log_fetcher(self.job_id)
+ if log_fetcher:
+ self.log.info("Fetching logs for AWS Batch job: %s",
self.job_id)
+ # Fetch up to 10,000 messages (CloudWatch limit)
+ log_messages = log_fetcher.get_last_log_messages(10000)
+ for message in log_messages:
+ self.log.info(message)
+ if log_messages:
+ self.log.info("Retrieved %d log messages from CloudWatch",
len(log_messages))
+ else:
+ self.log.info("No logs found in CloudWatch for job: %s",
self.job_id)
Review Comment:
`execute_complete()` calls `AwsTaskLogFetcher.get_last_log_messages()`
directly, which can raise `botocore` exceptions (e.g. log stream not found /
access denied). Unlike the background `run()` path (which catches these), this
will fail the task just because log retrieval failed. Wrap the fetch in
try/except and log a warning so deferred completion doesn’t error due to
CloudWatch log issues.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
