Support for Python log filters (#19526)

We can change our python logger by adding handlers and formatters:
https://docs.dagster.io/concepts/logging/python-logging#configuring-python-log-handlers-

However, due to the [schema
check](https://github.com/dagster-io/dagster/blob/626fedef2922c423ba7131441a5196f5fab0764e/python_modules/dagster/dagster/_core/instance/config.py#L177)
we can't add filters. They are part of the [default python config
dict](https://docs.python.org/3/library/logging.config.html) and are
applied to handlers. They are used to have filter logic to discard
certain log records, or [add contextual information to the log
record](https://docs.python.org/3/howto/logging-cookbook.html). The
latter is what I'd like to do. The workaround is to just do this in a
formatter - however this isn't the correct python way of doing it.

As dagster is just [passing this as a dict to a dummy
logger](https://github.com/dagster-io/dagster/blob/626fedef2922c423ba7131441a5196f5fab0764e/python_modules/dagster/dagster/_core/instance/__init__.py#L2324)
- allowing for a filter value for the schema should be fine. (It works
fine too if you skip the schema check by using
DagsterInstance.ephemeral() )

I've added filters as a possible value to the schema - and changed
documentation to reflect this. Some tests will be added
[here](https://github.com/dagster-io/dagster/blob/626fedef2922c423ba7131441a5196f5fab0764e/python_modules/dagster/dagster_tests/logging_tests/test_logging.py#L429)
to test this filters change and also test formatters.

docs/content/concepts/logging/python-logging.mdx

@@ -98,7 +98,7 @@ python_logs:
 
 ## Configuring Python log handlers <Experimental />
 
-In your `dagster.yaml` file, you can configure handlers and formatters that will apply to the Dagster instance. This will apply the same logging configuration to all runs.
+In your `dagster.yaml` file, you can configure handlers, formatters and filters that will apply to the Dagster instance. This will apply the same logging configuration to all runs.
 
 For example:
 
@@ -111,14 +111,19 @@ python_logs:
         level: INFO
         stream: ext://sys.stdout
         formatter: myFormatter
+        filters:
+          - myFilter
     formatters:
       myFormatter:
         format: "My formatted message: %(message)s"
+    filters:
+      myFilter:
+        name: dagster
 ```
 
-Handler and formatter configuration follows the [dictionary config schema format](https://docs.python.org/3/library/logging.config.html#logging-config-dictschema) in the Python logging module. Only the `handlers` and `formatters` dictionary keys will be accepted, as Dagster creates loggers internally.
+Handler, filter and formatter configuration follows the [dictionary config schema format](https://docs.python.org/3/library/logging.config.html#logging-config-dictschema) in the Python logging module. Only the `handlers`, `formatters` and `filters` dictionary keys will be accepted, as Dagster creates loggers internally.
 
-From there, standard `context.log` calls will output with the configured handlers and formatters. After execution, read the output log file `my_dagster_logs.log`. As expected, the log file contains the formatted log:
+From there, standard `context.log` calls will output with the configured handlers, formatters and filters. After execution, read the output log file `my_dagster_logs.log`. As expected, the log file contains the formatted log:
 
 <Image
 alt="log-file-output"

examples/docs_snippets/docs_snippets/concepts/logging/python_logging_handler_config.yaml

@@ -6,6 +6,11 @@ python_logs:
         level: INFO
         stream: ext://sys.stdout
         formatter: myFormatter
+        filters:
+          - myFilter
     formatters:
       myFormatter:
         format: "My formatted message: %(message)s"
+    filters:
+      myFilter:
+        name: dagster

python_modules/dagster/dagster/_core/instance/config.py

@@ -183,6 +183,7 @@ def python_logs_config_schema() -> Field:
                 {
                     "handlers": Field(dict, is_required=False),
                     "formatters": Field(dict, is_required=False),
+                    "filters": Field(dict, is_required=False),
                 },
                 is_required=False,
             ),

python_modules/dagster/dagster_tests/logging_tests/test_logging.py

@@ -221,6 +221,22 @@ def emit(self, record):
         self.captured.append(record)
 
 
+class CustomFormatter(logging.Formatter):
+    def format(self, record):
+        record.msg = "I was formatted"
+        return super().format(record)
+
+
+class CustomLevelFilter(logging.Filter):
+    def __init__(self, filter_level):
+        super().__init__()
+        self.filter_level = filter_level
+
+    def filter(self, record):
+        record.msg = f"{record.msg} default logger is {DAGSTER_DEFAULT_LOGGER}"
+        return record.levelno == self.filter_level
+
+
 def test_capture_handler_log_records():
     capture_handler = CaptureHandler()
 
@@ -466,6 +482,128 @@ def test_error_when_logger_defined_yaml():
             log_job.execute_in_process(instance=instance)
 
 
+def test_conf_log_formatter(capsys):
+    config_settings = {
+        "python_logs": {
+            "dagster_handler_config": {
+                "handlers": {
+                    "handlerOne": {
+                        "class": "logging.StreamHandler",
+                        "level": "INFO",
+                        "stream": "ext://sys.stdout",
+                        "formatter": "myFormatter",
+                    },
+                },
+                "formatters": {
+                    "myFormatter": {
+                        "format": "My formatted message: %(message)s",
+                    }
+                },
+            }
+        }
+    }
+
+    with instance_for_test(overrides=config_settings) as instance:
+        log_job.execute_in_process(instance=instance)
+
+    out, _ = capsys.readouterr()
+
+    # currently the format of dict-inputted handlers is undetermined, so
+    # we only check for the expected message
+    assert re.search(r"My formatted message: ", out)
+
+
+def test_conf_log_formatter_custom(capsys):
+    config_settings = {
+        "python_logs": {
+            "dagster_handler_config": {
+                "handlers": {
+                    "handlerOne": {
+                        "class": "logging.StreamHandler",
+                        "level": "INFO",
+                        "stream": "ext://sys.stdout",
+                        "formatter": "myFormatter",
+                    },
+                },
+                "formatters": {
+                    "myFormatter": {
+                        "()": "dagster_tests.logging_tests.test_logging.CustomFormatter",
+                    }
+                },
+            }
+        }
+    }
+
+    with instance_for_test(overrides=config_settings) as instance:
+        log_job.execute_in_process(instance=instance)
+
+    out, _ = capsys.readouterr()
+
+    assert re.search(r"I was formatted", out)
+
+
+def test_conf_log_filter(capsys):
+    config_settings = {
+        "python_logs": {
+            "dagster_handler_config": {
+                "handlers": {
+                    "handlerOne": {
+                        "class": "logging.StreamHandler",
+                        "level": "INFO",
+                        "stream": "ext://sys.stderr",
+                        "formatter": "myFormatter",
+                        "filters": ["myFilter"],
+                    },
+                },
+                "formatters": {
+                    "myFormatter": {
+                        "format": "Filter me out: %(message)s",
+                    }
+                },
+                "filters": {"myFilter": {"name": "none"}},
+            }
+        }
+    }
+
+    with instance_for_test(overrides=config_settings) as instance:
+        log_job.execute_in_process(instance=instance)
+
+    _, err = capsys.readouterr()
+
+    assert not re.search(r"Filter me out", err)
+
+
+def test_conf_log_filter_custom_with_context(capsys):
+    config_settings = {
+        "python_logs": {
+            "dagster_handler_config": {
+                "handlers": {
+                    "handlerOne": {
+                        "class": "logging.StreamHandler",
+                        "level": "INFO",
+                        "stream": "ext://sys.stdout",
+                        "filters": ["myFilter"],
+                    },
+                },
+                "filters": {
+                    "myFilter": {
+                        "()": "dagster_tests.logging_tests.test_logging.CustomLevelFilter",
+                        "filter_level": logging.ERROR,
+                    }
+                },
+            }
+        }
+    }
+
+    with instance_for_test(overrides=config_settings) as instance:
+        log_job.execute_in_process(instance=instance)
+
+    out, _ = capsys.readouterr()
+
+    assert not re.search(r"Hello world", out)
+    assert re.search(rf"My test error default logger is {DAGSTER_DEFAULT_LOGGER}", out)
+
+
 def test_python_multithread_context_logging():
     def logging_background_thread(thread_name, context):
         for i in range(1, 4):