You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by da...@apache.org on 2017/08/07 22:53:42 UTC
incubator-airflow git commit: [AIRFLOW-1486] Unexpected S3 writing
log error
Repository: incubator-airflow
Updated Branches:
refs/heads/master f5c845739 -> d9109d645
[AIRFLOW-1486] Unexpected S3 writing log error
Removed unexpected S3 writing log error and added tests for s3 logging.
Closes #2499 from skudriashev/airflow-1486
Project: http://git-wip-us.apache.org/repos/asf/incubator-airflow/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-airflow/commit/d9109d64
Tree: http://git-wip-us.apache.org/repos/asf/incubator-airflow/tree/d9109d64
Diff: http://git-wip-us.apache.org/repos/asf/incubator-airflow/diff/d9109d64
Branch: refs/heads/master
Commit: d9109d6458d136cd2b76ef7180be498ae09b3ea3
Parents: f5c8457
Author: Stanislav Kudriashev <st...@gmail.com>
Authored: Mon Aug 7 15:52:51 2017 -0700
Committer: Dan Davydov <da...@airbnb.com>
Committed: Mon Aug 7 15:52:54 2017 -0700
----------------------------------------------------------------------
airflow/utils/logging.py | 40 +++++++--------
tests/utils/test_logging.py | 107 +++++++++++++++++++++++++++++++++++----
2 files changed, 117 insertions(+), 30 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/d9109d64/airflow/utils/logging.py
----------------------------------------------------------------------
diff --git a/airflow/utils/logging.py b/airflow/utils/logging.py
index b86d839..6e18e52 100644
--- a/airflow/utils/logging.py
+++ b/airflow/utils/logging.py
@@ -91,10 +91,13 @@ class S3Log(object):
except:
pass
- # raise/return error if we get here
- err = 'Could not read logs from {}'.format(remote_log_location)
- logging.error(err)
- return err if return_error else ''
+ # return error if needed
+ if return_error:
+ msg = 'Could not read logs from {}'.format(remote_log_location)
+ logging.error(msg)
+ return msg
+
+ return ''
def write(self, log, remote_log_location, append=True):
"""
@@ -108,25 +111,21 @@ class S3Log(object):
:param append: if False, any existing log file is overwritten. If True,
the new log is appended to any existing logs.
:type append: bool
-
"""
if self.hook:
-
if append:
old_log = self.read(remote_log_location)
- log = old_log + '\n' + log
+ log = '\n'.join([old_log, log])
+
try:
self.hook.load_string(
log,
key=remote_log_location,
replace=True,
- encrypt=configuration.getboolean('core', 'ENCRYPT_S3_LOGS'))
- return
+ encrypt=configuration.getboolean('core', 'ENCRYPT_S3_LOGS'),
+ )
except:
- pass
-
- # raise/return error if we get here
- logging.error('Could not write logs to {}'.format(remote_log_location))
+ logging.error('Could not write logs to {}'.format(remote_log_location))
class GCSLog(object):
@@ -183,10 +182,13 @@ class GCSLog(object):
except:
pass
- # raise/return error if we get here
- err = 'Could not read logs from {}'.format(remote_log_location)
- logging.error(err)
- return err if return_error else ''
+ # return error if needed
+ if return_error:
+ msg = 'Could not read logs from {}'.format(remote_log_location)
+ logging.error(msg)
+ return msg
+
+ return ''
def write(self, log, remote_log_location, append=True):
"""
@@ -200,12 +202,11 @@ class GCSLog(object):
:param append: if False, any existing log file is overwritten. If True,
the new log is appended to any existing logs.
:type append: bool
-
"""
if self.hook:
if append:
old_log = self.read(remote_log_location)
- log = old_log + '\n' + log
+ log = '\n'.join([old_log, log])
try:
bkt, blob = self.parse_gcs_url(remote_log_location)
@@ -218,7 +219,6 @@ class GCSLog(object):
tmpfile.flush()
self.hook.upload(bkt, blob, tmpfile.name)
except:
- # raise/return error if we get here
logging.error('Could not write logs to {}'.format(remote_log_location))
def parse_gcs_url(self, gsurl):
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/d9109d64/tests/utils/test_logging.py
----------------------------------------------------------------------
diff --git a/tests/utils/test_logging.py b/tests/utils/test_logging.py
index 474430f..3f9f5d6 100644
--- a/tests/utils/test_logging.py
+++ b/tests/utils/test_logging.py
@@ -14,16 +14,103 @@
import unittest
-from airflow.exceptions import AirflowException
-from airflow.utils import logging as logging_utils
-from datetime import datetime, timedelta
+import mock
-class Logging(unittest.TestCase):
+from airflow.utils import logging
+from datetime import datetime
+
+
+class TestLogging(unittest.TestCase):
def test_get_log_filename(self):
- dag_id = 'dag_id'
- task_id = 'task_id'
- execution_date = datetime(2017, 1, 1, 0, 0, 0)
- try_number = 0
- filename = logging_utils.get_log_filename(dag_id, task_id, execution_date, try_number)
- self.assertEqual(filename, 'dag_id/task_id/2017-01-01T00:00:00/1.log')
+ self.assertEqual(
+ logging.get_log_filename(
+ dag_id='dag_id',
+ task_id='task_id',
+ execution_date=datetime(2017, 1, 1, 0, 0, 0),
+ try_number=0,
+ ),
+ 'dag_id/task_id/2017-01-01T00:00:00/1.log',
+ )
+
+
+class TestS3Log(unittest.TestCase):
+
+ def setUp(self):
+ super(TestS3Log, self).setUp()
+ self.remote_log_location = 'remote/log/location'
+ self.hook_patcher = mock.patch("airflow.hooks.S3_hook.S3Hook")
+ self.hook_mock = self.hook_patcher.start()
+ self.hook_inst_mock = self.hook_mock.return_value
+ self.hook_key_mock = self.hook_inst_mock.get_key.return_value
+ self.hook_key_mock.get_contents_as_string.return_value.decode.\
+ return_value = 'content'
+ self.logging_patcher = mock.patch("airflow.utils.logging.logging")
+ self.logging_mock = self.logging_patcher.start()
+
+ def tearDown(self):
+ self.logging_patcher.stop()
+ self.hook_patcher.stop()
+ super(TestS3Log, self).tearDown()
+
+ def test_init(self):
+ logging.S3Log()
+ self.hook_mock.assert_called_once_with('')
+
+ def test_init_raises(self):
+ self.hook_mock.side_effect = Exception('Failed to connect')
+ logging.S3Log()
+ self.logging_mock.error.assert_called_once_with(
+ 'Could not create an S3Hook with connection id "". Please make '
+ 'sure that airflow[s3] is installed and the S3 connection exists.'
+ )
+
+ def test_log_exists(self):
+ self.assertTrue(logging.S3Log().log_exists(self.remote_log_location))
+
+ def test_log_exists_none(self):
+ self.hook_inst_mock.get_key.return_value = None
+ self.assertFalse(logging.S3Log().log_exists(self.remote_log_location))
+
+ def test_log_exists_raises(self):
+ self.hook_inst_mock.get_key.side_effect = Exception('error')
+ self.assertFalse(logging.S3Log().log_exists(self.remote_log_location))
+
+ def test_log_exists_no_hook(self):
+ self.hook_mock.side_effect = Exception('Failed to connect')
+ self.assertFalse(logging.S3Log().log_exists(self.remote_log_location))
+
+ def test_read(self):
+ self.assertEqual(logging.S3Log().read(self.remote_log_location),
+ 'content')
+
+ def test_read_key_empty(self):
+ self.hook_inst_mock.get_key.return_value = None
+ self.assertEqual(logging.S3Log().read(self.remote_log_location), '')
+
+ def test_read_raises(self):
+ self.hook_inst_mock.get_key.side_effect = Exception('error')
+ self.assertEqual(logging.S3Log().read(self.remote_log_location), '')
+
+ def test_read_raises_return_error(self):
+ self.hook_inst_mock.get_key.side_effect = Exception('error')
+ result = logging.S3Log().read(self.remote_log_location,
+ return_error=True)
+ msg = 'Could not read logs from %s' % self.remote_log_location
+ self.assertEqual(result, msg)
+ self.logging_mock.error.assert_called_once_with(msg)
+
+ def test_write(self):
+ logging.S3Log().write('text', self.remote_log_location)
+ self.hook_inst_mock.load_string.assert_called_once_with(
+ 'content\ntext',
+ key=self.remote_log_location,
+ replace=True,
+ encrypt=False,
+ )
+
+ def test_write_raises(self):
+ self.hook_inst_mock.load_string.side_effect = Exception('error')
+ logging.S3Log().write('text', self.remote_log_location)
+ msg = 'Could not write logs to %s' % self.remote_log_location
+ self.logging_mock.error.assert_called_once_with(msg)