kakatur commented on code in PR #64745:
URL: https://github.com/apache/airflow/pull/64745#discussion_r3463528305
##########
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:
I've added try/except around get_last_log_messages(). Unlike the background
thread's _get_log_events(), it doesn't handle botocore exceptions, so log
retrieval failures would incorrectly fail the task.
--
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]
