From ae4cbcd52a2865b7ad16c9cab53fe9f4cc2aab97 Mon Sep 17 00:00:00 2001 From: Steve Zhang Date: Thu, 3 Mar 2022 14:46:25 -0800 Subject: [PATCH 1/4] If uploading task logs to S3 fails, retry once --- airflow/providers/amazon/aws/log/s3_task_handler.py | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/airflow/providers/amazon/aws/log/s3_task_handler.py b/airflow/providers/amazon/aws/log/s3_task_handler.py index 574607e0632d3..9020d4e959eee 100644 --- a/airflow/providers/amazon/aws/log/s3_task_handler.py +++ b/airflow/providers/amazon/aws/log/s3_task_handler.py @@ -190,4 +190,14 @@ def s3_write(self, log: str, remote_log_location: str, append: bool = True): encrypt=conf.getboolean('logging', 'ENCRYPT_S3_LOGS'), ) except Exception: - self.log.exception('Could not write logs to %s', remote_log_location) + self.log.exception('Could not write logs to %s, retrying once', remote_log_location) + + try: + self.hook.load_string( + log, + key=remote_log_location, + replace=True, + encrypt=conf.getboolean('logging', 'ENCRYPT_S3_LOGS'), + ) + except Exception: + self.log.exception('Could not write logs to %s, retry failed', remote_log_location) From aedcdf79157570d1e8924070f61b1d9a464caa69 Mon Sep 17 00:00:00 2001 From: Steve Zhang Date: Thu, 3 Mar 2022 16:12:31 -0800 Subject: [PATCH 2/4] change first failure message to warning --- airflow/providers/amazon/aws/log/s3_task_handler.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/airflow/providers/amazon/aws/log/s3_task_handler.py b/airflow/providers/amazon/aws/log/s3_task_handler.py index 9020d4e959eee..88c8087d4ec14 100644 --- a/airflow/providers/amazon/aws/log/s3_task_handler.py +++ b/airflow/providers/amazon/aws/log/s3_task_handler.py @@ -190,7 +190,7 @@ def s3_write(self, log: str, remote_log_location: str, append: bool = True): encrypt=conf.getboolean('logging', 'ENCRYPT_S3_LOGS'), ) except Exception: - self.log.exception('Could not write logs to %s, retrying once', remote_log_location) + self.log.warn('Failed first attempt to write logs to %s, retrying', remote_log_location) try: self.hook.load_string( @@ -200,4 +200,4 @@ def s3_write(self, log: str, remote_log_location: str, append: bool = True): encrypt=conf.getboolean('logging', 'ENCRYPT_S3_LOGS'), ) except Exception: - self.log.exception('Could not write logs to %s, retry failed', remote_log_location) + self.log.exception('Could not write logs to %s', remote_log_location) From 316562b809938c846ed82673cdc35d155a3df739 Mon Sep 17 00:00:00 2001 From: Steve Zhang Date: Thu, 3 Mar 2022 21:26:33 -0800 Subject: [PATCH 3/4] fix warn->warning --- airflow/providers/amazon/aws/log/s3_task_handler.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/airflow/providers/amazon/aws/log/s3_task_handler.py b/airflow/providers/amazon/aws/log/s3_task_handler.py index 88c8087d4ec14..084b0098096f4 100644 --- a/airflow/providers/amazon/aws/log/s3_task_handler.py +++ b/airflow/providers/amazon/aws/log/s3_task_handler.py @@ -190,7 +190,7 @@ def s3_write(self, log: str, remote_log_location: str, append: bool = True): encrypt=conf.getboolean('logging', 'ENCRYPT_S3_LOGS'), ) except Exception: - self.log.warn('Failed first attempt to write logs to %s, retrying', remote_log_location) + self.log.warning('Failed first attempt to write logs to %s, retrying', remote_log_location) try: self.hook.load_string( From 6348a825cad2fc117f57f03ba8ac36a2c7aebefd Mon Sep 17 00:00:00 2001 From: Steve Zhang Date: Mon, 7 Mar 2022 10:12:14 -0800 Subject: [PATCH 4/4] address review feedback --- .../amazon/aws/log/s3_task_handler.py | 23 +++++++++---------- 1 file changed, 11 insertions(+), 12 deletions(-) diff --git a/airflow/providers/amazon/aws/log/s3_task_handler.py b/airflow/providers/amazon/aws/log/s3_task_handler.py index 084b0098096f4..695c4623d97b2 100644 --- a/airflow/providers/amazon/aws/log/s3_task_handler.py +++ b/airflow/providers/amazon/aws/log/s3_task_handler.py @@ -165,7 +165,7 @@ def s3_read(self, remote_log_location: str, return_error: bool = False) -> str: return msg return '' - def s3_write(self, log: str, remote_log_location: str, append: bool = True): + def s3_write(self, log: str, remote_log_location: str, append: bool = True, max_retry: int = 1): """ Writes the log to the remote_log_location. Fails silently if no hook was created. @@ -174,6 +174,7 @@ def s3_write(self, log: str, remote_log_location: str, append: bool = True): :param remote_log_location: the log's location in remote storage :param append: if False, any existing log file is overwritten. If True, the new log is appended to any existing logs. + :param max_retry: Maximum number of times to retry on upload failure """ try: if append and self.s3_log_exists(remote_log_location): @@ -182,16 +183,10 @@ def s3_write(self, log: str, remote_log_location: str, append: bool = True): except Exception: self.log.exception('Could not verify previous log to append') - try: - self.hook.load_string( - log, - key=remote_log_location, - replace=True, - encrypt=conf.getboolean('logging', 'ENCRYPT_S3_LOGS'), - ) - except Exception: - self.log.warning('Failed first attempt to write logs to %s, retrying', remote_log_location) - + # Default to a single retry attempt because s3 upload failures are + # rare but occasionally occur. Multiple retry attempts are unlikely + # to help as they usually indicate non-empheral errors. + for try_num in range(1 + max_retry): try: self.hook.load_string( log, @@ -199,5 +194,9 @@ def s3_write(self, log: str, remote_log_location: str, append: bool = True): replace=True, encrypt=conf.getboolean('logging', 'ENCRYPT_S3_LOGS'), ) + break except Exception: - self.log.exception('Could not write logs to %s', remote_log_location) + if try_num < max_retry: + self.log.warning('Failed attempt to write logs to %s, will retry', remote_log_location) + else: + self.log.exception('Could not write logs to %s', remote_log_location)