Add regex-based filters to logger component

homeassistant/components/logger/__init__.py

@@ -26,6 +26,7 @@
 
 LOGGER_DEFAULT = "default"
 LOGGER_LOGS = "logs"
+LOGGER_FILTERS = "filters"
 
 ATTR_LEVEL = "level"
 
@@ -40,6 +41,7 @@
             {
                 vol.Optional(LOGGER_DEFAULT): _VALID_LOG_LEVEL,
                 vol.Optional(LOGGER_LOGS): vol.Schema({cv.string: _VALID_LOG_LEVEL}),
+                vol.Optional(LOGGER_FILTERS): vol.Schema({cv.string: [cv.is_regex]}),
             }
         )
     },
@@ -70,6 +72,11 @@ def set_log_levels(logpoints):
     if LOGGER_LOGS in config[DOMAIN]:
         set_log_levels(config[DOMAIN][LOGGER_LOGS])
 
+    if LOGGER_FILTERS in config[DOMAIN]:
+        for key, value in config[DOMAIN][LOGGER_FILTERS].items():
+            logger = logging.getLogger(key)
+            _add_log_filter(logger, value)
+
     @callback
     def async_service_handler(service):
         """Handle logger services."""
@@ -103,6 +110,15 @@ def _set_log_level(logger, level):
     getattr(logger, "orig_setLevel", logger.setLevel)(LOGSEVERITY[level])
 
 
+def _add_log_filter(logger, patterns):
+    """Add a Filter to the logger based on a regexp of the filter_str."""
+
+    def filter_func(logrecord):
+        return not any(p.match(logrecord.getMessage()) for p in patterns)
+
+    logger.addFilter(filter_func)
+
+
 def _get_logger_class(hass_overrides):
     """Create a logger subclass.
 

tests/components/logger/test_init.py

@@ -25,6 +25,54 @@ def restore_logging_class():
     logging.setLoggerClass(klass)
 
 
+async def test_log_filtering(hass):
+    """Test logging filters."""
+
+    assert await async_setup_component(
+        hass,
+        "logger",
+        {
+            "logger": {
+                "default": "warning",
+                "logs": {
+                    "test.filter": "info",
+                },
+                "filters": {
+                    "test.filter": [
+                        "doesntmatchanything",
+                        ".*shouldfilterall.*",
+                        "^filterthis:.*",
+                    ],
+                },
+            }
+        },
+    )
+    await hass.async_block_till_done()
+
+    logger = logging.getLogger("test.filter")
+
+    def msg_test(msg, result, args=None):
+        record = logger.makeRecord(
+            "test.filter", logging.WARNING, "test_log_filtering", 0, msg, args, None
+        )
+        assert logger.filter(record) == result
+
+    msg_test("this line containing shouldfilterall should be filtered", False)
+    msg_test("this line should not be filtered filterthis:", True)
+    msg_test("filterthis: should be filtered", False)
+    msg_test("format string shouldfilter%s", False, args=("all"))
+    msg_test("format string shouldfilter%s", True, args=("none"))
+    # Filtering should work even if log level is modified
+    await hass.services.async_call(
+        "logger",
+        "set_level",
+        {"test.filter": "warning"},
+        blocking=True,
+    )
+    assert logger.getEffectiveLevel() == logging.WARNING
+    msg_test("this line containing shouldfilterall should still be filtered", False)
+
+
 async def test_setting_level(hass):
     """Test we set log levels."""
     mocks = defaultdict(Mock)