diff --git a/homeassistant/components/logger/__init__.py b/homeassistant/components/logger/__init__.py
index ba0026fedc513..fb2920fb6e25d 100644
--- a/homeassistant/components/logger/__init__.py
+++ b/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.
 
diff --git a/tests/components/logger/test_init.py b/tests/components/logger/test_init.py
index 61818d57df96a..d2b0e8931b690 100644
--- a/tests/components/logger/test_init.py
+++ b/tests/components/logger/test_init.py
@@ -25,6 +25,72 @@ def restore_logging_class():
     logging.setLoggerClass(klass)
 
 
+async def test_log_filtering(hass, caplog):
+    """Test logging filters."""
+
+    assert await async_setup_component(
+        hass,
+        "logger",
+        {
+            "logger": {
+                "default": "warning",
+                "logs": {
+                    "test.filter": "info",
+                },
+                "filters": {
+                    "test.filter": [
+                        "doesntmatchanything",
+                        ".*shouldfilterall.*",
+                        "^filterthis:.*",
+                    ],
+                    "test.other_filter": [".*otherfilterer"],
+                },
+            }
+        },
+    )
+    await hass.async_block_till_done()
+
+    filter_logger = logging.getLogger("test.filter")
+
+    def msg_test(logger, result, message, *args):
+        logger.error(message, *args)
+        formatted_message = message % args
+        assert (formatted_message in caplog.text) == result
+        caplog.clear()
+
+    msg_test(
+        filter_logger, False, "this line containing shouldfilterall should be filtered"
+    )
+    msg_test(filter_logger, True, "this line should not be filtered filterthis:")
+    msg_test(filter_logger, False, "filterthis: should be filtered")
+    msg_test(filter_logger, False, "format string shouldfilter%s", "all")
+    msg_test(filter_logger, True, "format string shouldfilter%s", "not")
+
+    # Filtering should work even if log level is modified
+    await hass.services.async_call(
+        "logger",
+        "set_level",
+        {"test.filter": "warning"},
+        blocking=True,
+    )
+    assert filter_logger.getEffectiveLevel() == logging.WARNING
+    msg_test(
+        filter_logger,
+        False,
+        "this line containing shouldfilterall should still be filtered",
+    )
+
+    # Filtering should be scoped to a service
+    msg_test(
+        filter_logger, True, "this line containing otherfilterer should not be filtered"
+    )
+    msg_test(
+        logging.getLogger("test.other_filter"),
+        False,
+        "this line containing otherfilterer SHOULD be filtered",
+    )
+
+
 async def test_setting_level(hass):
     """Test we set log levels."""
     mocks = defaultdict(Mock)