Add info log message about duration taken to load plugins (#12308)

Loading plugins, particularly from setuptools entry points can be slow,
and since by default this happens per-task, it can slow down task
execution unexpectedly.

By having this log message users can know the source of the delay

(The change to test_standard_task_runner was to remove logging-config
side effects from that test)

airflow/plugins_manager.py

@@ -23,6 +23,7 @@ import inspect
 import logging
 import os
 import sys
+import time
 import types
 from typing import TYPE_CHECKING, Any, Dict, List, Optional, Type
 
@@ -262,6 +263,8 @@ def ensure_plugins_loaded():
 
     log.debug("Loading plugins")
 
+    start = time.monotonic()
+
     plugins = []
     registered_hooks = []
 
@@ -273,6 +276,12 @@ def ensure_plugins_loaded():
     for plugin in plugins:
         registered_hooks.extend(plugin.hooks)
 
+    end = time.monotonic()
+
+    num_loaded = len(plugins)
+    if num_loaded > 0:
+        log.info("Loading %d plugin(s) took %.2f seconds", num_loaded, end - start)
+
 
 def initialize_web_ui_plugins():
     """Collect extension points for WEB UI"""

airflow/utils/log/colored_log.py

@@ -90,8 +90,9 @@ class CustomTTYColoredFormatter(TTYColoredFormatter):
 
     def format(self, record: LogRecord) -> str:
         try:
-            record = self._color_record_args(record)
-            record = self._color_record_traceback(record)
+            if self.stream.isatty():
+                record = self._color_record_args(record)
+                record = self._color_record_traceback(record)
             return super().format(record)
         except ValueError:  # I/O operation on closed file
             from logging import Formatter

tests/plugins/test_plugins_manager.py

@@ -15,6 +15,7 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
+import logging
 import unittest
 from unittest import mock
 
@@ -102,8 +103,17 @@ class TestPluginsRBAC(unittest.TestCase):
             assert ("test.plugins.test_plugins_manager", "Version Conflict") in import_errors.items()
 
 
-class TestPluginsManager(unittest.TestCase):
-    def test_should_load_plugins_from_property(self):
+class TestPluginsManager:
+    def test_no_log_when_no_plugins(self, caplog):
+
+        with mock_plugin_manager(plugins=[]):
+            from airflow import plugins_manager
+
+            plugins_manager.ensure_plugins_loaded()
+
+        assert caplog.record_tuples == []
+
+    def test_should_load_plugins_from_property(self, caplog):
         class AirflowTestPropertyPlugin(AirflowPlugin):
             name = "test_property_plugin"
 
@@ -119,10 +129,13 @@ class TestPluginsManager(unittest.TestCase):
 
             plugins_manager.ensure_plugins_loaded()
 
-            self.assertIn('AirflowTestPropertyPlugin', str(plugins_manager.plugins))
-            self.assertIn("TestPropertyHook", str(plugins_manager.registered_hooks))
+            assert 'AirflowTestPropertyPlugin' in str(plugins_manager.plugins)
+            assert 'TestPropertyHook' in str(plugins_manager.registered_hooks)
 
-    def test_should_warning_about_incompatible_plugins(self):
+        assert caplog.records[0].levelname == 'INFO'
+        assert caplog.records[0].msg == 'Loading %d plugin(s) took %.2f seconds'
+
+    def test_should_warning_about_incompatible_plugins(self, caplog):
         class AirflowAdminViewsPlugin(AirflowPlugin):
             name = "test_admin_views_plugin"
 
@@ -133,26 +146,29 @@ class TestPluginsManager(unittest.TestCase):
 
             menu_links = [mock.MagicMock()]
 
-        with mock_plugin_manager(plugins=[AirflowAdminViewsPlugin(), AirflowAdminMenuLinksPlugin()]):
+        with mock_plugin_manager(
+            plugins=[AirflowAdminViewsPlugin(), AirflowAdminMenuLinksPlugin()]
+        ), caplog.at_level(logging.WARNING, logger='airflow.plugins_manager'):
             from airflow import plugins_manager
 
-            # assert not logs
-            with self.assertLogs(plugins_manager.log) as cm:
-                plugins_manager.initialize_web_ui_plugins()
+            plugins_manager.initialize_web_ui_plugins()
 
-        self.assertEqual(
-            cm.output,
-            [
-                'WARNING:airflow.plugins_manager:Plugin \'test_admin_views_plugin\' may not be '
-                'compatible with the current Airflow version. Please contact the author of '
-                'the plugin.',
-                'WARNING:airflow.plugins_manager:Plugin \'test_menu_links_plugin\' may not be '
-                'compatible with the current Airflow version. Please contact the author of '
-                'the plugin.',
-            ],
-        )
+        assert caplog.record_tuples == [
+            (
+                "airflow.plugins_manager",
+                logging.WARNING,
+                "Plugin 'test_admin_views_plugin' may not be compatible with the current Airflow version. "
+                "Please contact the author of the plugin.",
+            ),
+            (
+                "airflow.plugins_manager",
+                logging.WARNING,
+                "Plugin 'test_menu_links_plugin' may not be compatible with the current Airflow version. "
+                "Please contact the author of the plugin.",
+            ),
+        ]
 
-    def test_should_not_warning_about_fab_plugins(self):
+    def test_should_not_warning_about_fab_plugins(self, caplog):
         class AirflowAdminViewsPlugin(AirflowPlugin):
             name = "test_admin_views_plugin"
 
@@ -163,14 +179,16 @@ class TestPluginsManager(unittest.TestCase):
 
             appbuilder_menu_items = [mock.MagicMock()]
 
-        with mock_plugin_manager(plugins=[AirflowAdminViewsPlugin(), AirflowAdminMenuLinksPlugin()]):
+        with mock_plugin_manager(
+            plugins=[AirflowAdminViewsPlugin(), AirflowAdminMenuLinksPlugin()]
+        ), caplog.at_level(logging.WARNING, logger='airflow.plugins_manager'):
             from airflow import plugins_manager
 
-            # assert not logs
-            with self.assertRaises(AssertionError), self.assertLogs(plugins_manager.log):
-                plugins_manager.initialize_web_ui_plugins()
+            plugins_manager.initialize_web_ui_plugins()
 
-    def test_should_not_warning_about_fab_and_flask_admin_plugins(self):
+        assert caplog.record_tuples == []
+
+    def test_should_not_warning_about_fab_and_flask_admin_plugins(self, caplog):
         class AirflowAdminViewsPlugin(AirflowPlugin):
             name = "test_admin_views_plugin"
 
@@ -183,12 +201,14 @@ class TestPluginsManager(unittest.TestCase):
             menu_links = [mock.MagicMock()]
             appbuilder_menu_items = [mock.MagicMock()]
 
-        with mock_plugin_manager(plugins=[AirflowAdminViewsPlugin(), AirflowAdminMenuLinksPlugin()]):
+        with mock_plugin_manager(
+            plugins=[AirflowAdminViewsPlugin(), AirflowAdminMenuLinksPlugin()]
+        ), caplog.at_level(logging.WARNING, logger='airflow.plugins_manager'):
             from airflow import plugins_manager
 
-            # assert not logs
-            with self.assertRaises(AssertionError), self.assertLogs(plugins_manager.log):
-                plugins_manager.initialize_web_ui_plugins()
+            plugins_manager.initialize_web_ui_plugins()
+
+        assert caplog.record_tuples == []
 
 
 class TestPluginsDirectorySource(unittest.TestCase):

tests/task/task_runner/test_standard_task_runner.py

@@ -16,6 +16,7 @@
 # specific language governing permissions and limitations
 # under the License.
 import getpass
+import logging
 import os
 import time
 import unittest
@@ -60,6 +61,9 @@ class TestStandardTaskRunner(unittest.TestCase):
 
     @classmethod
     def tearDownClass(cls):
+        airflow_logger = logging.getLogger('airflow')
+        airflow_logger.handlers = []
+        airflow_logger.propagate = True
         try:
             clear_db_runs()
         except Exception:  # noqa pylint: disable=broad-except