[AIRFLOW-2707] Validate task_log_reader on upgrade from <=1.9 (apache#3881)

We changed the default logging config and config from 1.9 to 1.10, but
anyone who upgrades and has an existing airflow.cfg won't know they need
to change this value - instead they will get nothing displayed in the UI
(ajax request fails) and see "'NoneType' object has no attribute 'read'"
in the error log.

This validates that config section at start up, and seamlessly upgrades
the old previous value.

Author: ashb

airflow/logging_config.py

@@ -18,6 +18,7 @@
 # under the License.
 #
 import logging
+import warnings
 from logging.config import dictConfig
 
 from airflow import configuration as conf
@@ -70,4 +71,36 @@ def configure_logging():
         # otherwise Airflow would silently fall back on the default config
         raise e
 
+    validate_logging_config(logging_config)
+
     return logging_config
+
+
+def validate_logging_config(logging_config):
+    # Now lets validate the other logging-related settings
+    task_log_reader = conf.get('core', 'task_log_reader')
+
+    logger = logging.getLogger('airflow.task')
+
+    def _get_handler(name):
+        return next((h for h in logger.handlers if h.name == name), None)
+
+    if _get_handler(task_log_reader) is None:
+        # Check for pre 1.10 setting that might be in deployed airflow.cfg files
+        if task_log_reader == "file.task" and _get_handler("task"):
+            warnings.warn(
+                "task_log_reader setting in [core] has a deprecated value of "
+                "{!r}, but no handler with this name was found. Please update "
+                "your config to use {!r}. Running config has been adjusted to "
+                "match".format(
+                    task_log_reader,
+                    "task",
+                ),
+                DeprecationWarning,
+            )
+            conf.set('core', 'task_log_reader', 'task')
+        else:
+            raise AirflowConfigException(
+                "Configured task_log_reader {!r} was not a handler of the 'airflow.task' "
+                "logger.".format(task_log_reader)
+            )

tests/test_logging_config.py

@@ -17,16 +17,22 @@
 # specific language governing permissions and limitations
 # under the License.
 import os
-import shutil
+import six
 import sys
 import tempfile
-import unittest
 from mock import patch, mock
 
 from airflow import configuration as conf
 from airflow.configuration import mkdir_p
 from airflow.exceptions import AirflowConfigException
 
+
+if six.PY2:
+    # Need `assertWarns` back-ported from unittest2
+    import unittest2 as unittest
+else:
+    import unittest
+
 SETTINGS_FILE_VALID = """
 LOGGING_CONFIG = {
     'version': 1,
@@ -41,14 +47,24 @@
             'class': 'logging.StreamHandler',
             'formatter': 'airflow.task',
             'stream': 'ext://sys.stdout'
-        }
+        },
+        'task': {
+            'class': 'logging.StreamHandler',
+            'formatter': 'airflow.task',
+            'stream': 'ext://sys.stdout'
+        },
     },
     'loggers': {
         'airflow': {
             'handlers': ['console'],
             'level': 'INFO',
             'propagate': False
-        }
+        },
+        'airflow.task': {
+            'handlers': ['task'],
+            'level': 'INFO',
+            'propagate': False,
+        },
     }
 }
 """
@@ -200,18 +216,21 @@ def test_loading_no_local_settings(self):
     # When the key is not available in the configuration
     def test_when_the_config_key_does_not_exists(self):
         from airflow import logging_config
+        conf_get = conf.get
 
         def side_effect(*args):
             if args[1] == 'logging_config_class':
                 raise AirflowConfigException
             else:
-                return "bla_bla_from_test"
+                return conf_get(*args)
 
         logging_config.conf.get = mock.Mock(side_effect=side_effect)
 
         with patch.object(logging_config.log, 'debug') as mock_debug:
             logging_config.configure_logging()
-            mock_debug.assert_any_call('Could not find key logging_config_class in config')
+            mock_debug.assert_any_call(
+                'Could not find key logging_config_class in config'
+            )
 
     # Just default
     def test_loading_local_settings_without_logging_config(self):
@@ -222,6 +241,16 @@ def test_loading_local_settings_without_logging_config(self):
                 'Unable to load custom logging, using default config instead'
             )
 
+    def test_1_9_config(self):
+        from airflow.logging_config import configure_logging
+        conf.set('core', 'task_log_reader', 'file.task')
+        try:
+            with self.assertWarnsRegex(DeprecationWarning, r'file.task'):
+                configure_logging()
+                self.assertEqual(conf.get('core', 'task_log_reader'), 'task')
+        finally:
+            conf.remove_option('core', 'task_log_reader', remove_default=False)
+
 
 if __name__ == '__main__':
     unittest.main()