RFC: Change logger behavior so that setting a default logger in a Definitions object changes the logger by default, without needing to set run config to enable it

python_modules/dagster/dagster/_core/definitions/logger_definition.py

@@ -16,6 +16,7 @@
 
     from dagster._core.definitions import JobDefinition
     from dagster._core.execution.context.logger import InitLoggerContext, UnboundInitLoggerContext
+    from dagster._core.instance import DagsterInstance
 
     InitLoggerFunction = Callable[[InitLoggerContext], logging.Logger]
 
@@ -67,7 +68,7 @@ def __call__(self, *args, **kwargs):
                 args[0],
                 context_param_name,
                 UnboundInitLoggerContext,
-                default=UnboundInitLoggerContext(logger_config=None, job_def=None),
+                default=UnboundInitLoggerContext(logger_config=None, job_def=None, instance=None),
             )
             return logger_invocation_result(self, context)
         else:
@@ -79,7 +80,11 @@ def __call__(self, *args, **kwargs):
                 kwargs[context_param_name],
                 context_param_name,
                 UnboundInitLoggerContext,
-                default=UnboundInitLoggerContext(logger_config=None, job_def=None),
+                default=UnboundInitLoggerContext(
+                    logger_config=None,
+                    job_def=None,
+                    instance=None,
+                ),
             )
 
             return logger_invocation_result(self, context)
@@ -161,6 +166,7 @@ def _wrap(logger_fn: "InitLoggerFunction") -> "LoggerDefinition":
 def build_init_logger_context(
     logger_config: Any = None,
     job_def: Optional["JobDefinition"] = None,
+    instance: Optional["DagsterInstance"] = None,
 ) -> "UnboundInitLoggerContext":
     """Builds logger initialization context from provided parameters.
 
@@ -184,4 +190,4 @@ def build_init_logger_context(
 
     check.opt_inst_param(job_def, "job_def", JobDefinition)
 
-    return UnboundInitLoggerContext(logger_config=logger_config, job_def=job_def)
+    return UnboundInitLoggerContext(logger_config=logger_config, job_def=job_def, instance=instance)

python_modules/dagster/dagster/_core/definitions/logger_invocation.py

@@ -8,7 +8,11 @@ def logger_invocation_result(logger_def: LoggerDefinition, init_context: Unbound
     logger_config = resolve_bound_config(init_context.logger_config, logger_def)
 
     bound_context = InitLoggerContext(
-        logger_config, logger_def, init_context.job_def, init_context.run_id
+        logger_config,
+        logger_def,
+        init_context.job_def,
+        init_context.run_id,
+        init_context.instance,
     )
 
     return logger_def.logger_fn(bound_context)

python_modules/dagster/dagster/_core/definitions/run_config.py

@@ -107,7 +107,7 @@ class RunConfigSchemaCreationData(NamedTuple):
 def define_logger_dictionary_cls(creation_data: RunConfigSchemaCreationData) -> Shape:
     return Shape(
         {
-            logger_name: def_config_field(logger_definition, is_required=False)
+            logger_name: def_config_field(logger_definition, is_required=None)
             for logger_name, logger_definition in creation_data.logger_defs.items()
         }
     )

python_modules/dagster/dagster/_core/execution/context/logger.py

@@ -6,6 +6,7 @@
 from dagster._core.definitions.logger_definition import LoggerDefinition
 from dagster._core.errors import DagsterInvariantViolationError
 from dagster._core.execution.context.output import RUN_ID_PLACEHOLDER
+from dagster._core.instance import DagsterInstance
 
 
 class InitLoggerContext:
@@ -32,11 +33,13 @@ def __init__(
         logger_def: Optional[LoggerDefinition] = None,
         job_def: Optional[JobDefinition] = None,
         run_id: Optional[str] = None,
+        instance: Optional[DagsterInstance] = None,
     ):
         self._logger_config = logger_config
         self._job_def = check.opt_inst_param(job_def, "job_def", JobDefinition)
         self._logger_def = check.opt_inst_param(logger_def, "logger_def", LoggerDefinition)
         self._run_id = check.opt_str_param(run_id, "run_id")
+        self._instance = check.opt_inst_param(instance, "instance", DagsterInstance)
 
     @public
     @property
@@ -57,12 +60,24 @@ def logger_def(self) -> Optional[LoggerDefinition]:
         """The logger definition for the logger being constructed."""
         return self._logger_def
 
+    @property
+    def instance(self) -> Optional[DagsterInstance]:
+        return self._instance
+
     @public
     @property
     def run_id(self) -> Optional[str]:
         """The ID for this run of the job."""
         return self._run_id
 
+    @property
+    def default_log_level(self) -> str:
+        return (
+            self.instance.python_log_level
+            if self.instance and self.instance.python_log_level
+            else "DEBUG"
+        )
+
 
 class UnboundInitLoggerContext(InitLoggerContext):
     """Logger initialization context outputted by ``build_init_logger_context``.
@@ -73,9 +88,18 @@ class UnboundInitLoggerContext(InitLoggerContext):
     and it is subsumed into an `InitLoggerContext`, which contains the logger_def validated against.
     """
 
-    def __init__(self, logger_config: Any, job_def: Optional[JobDefinition]):
+    def __init__(
+        self,
+        logger_config: Any,
+        job_def: Optional[JobDefinition],
+        instance: Optional[DagsterInstance] = None,
+    ):
         super(UnboundInitLoggerContext, self).__init__(
-            logger_config, logger_def=None, job_def=job_def, run_id=None
+            logger_config,
+            logger_def=None,
+            job_def=job_def,
+            run_id=None,
+            instance=instance,
         )
 
     @property

python_modules/dagster/dagster/_core/execution/context_creation_job.py

@@ -70,7 +70,10 @@ def initialize_console_manager(
         loggers.append(
             logger_def.logger_fn(
                 InitLoggerContext(
-                    logger_config, logger_def, run_id=dagster_run.run_id if dagster_run else None
+                    logger_config,
+                    logger_def,
+                    run_id=dagster_run.run_id if dagster_run else None,
+                    instance=instance,
                 )
             )
         )
@@ -470,6 +473,7 @@ def create_log_manager(
                         logger_def,
                         job_def=job_def,
                         run_id=dagster_run.run_id,
+                        instance=context_creation_data.instance,
                     )
                 )
             )
@@ -483,6 +487,7 @@ def create_log_manager(
                         logger_def,
                         job_def=job_def,
                         run_id=dagster_run.run_id,
+                        instance=context_creation_data.instance,
                     )
                 )
             )
@@ -515,6 +520,7 @@ def create_context_free_log_manager(
                     logger_def,
                     job_def=None,
                     run_id=dagster_run.run_id,
+                    instance=instance,
                 )
             )
         ]

python_modules/dagster/dagster/_core/execution/host_mode.py

@@ -101,6 +101,7 @@ def host_mode_execution_context_event_generator(
                     job_def=None,
                     logger_def=logger_def,
                     run_id=pipeline_run.run_id,
+                    instance=instance,
                 )
             )
         )

python_modules/dagster/dagster/_loggers/__init__.py

@@ -26,7 +26,6 @@
             "log_level": Field(
                 str,
                 is_required=False,
-                default_value="INFO",
                 description="The logger's threshold.",
             ),
             "name": Field(
@@ -44,9 +43,11 @@ def colored_console_logger(init_context: "InitLoggerContext") -> logging.Logger:
     """This logger provides support for sending Dagster logs to stdout in a colored format. It is
     included by default on jobs which do not otherwise specify loggers.
     """
+    log_level = init_context.logger_config.get("log_level", init_context.default_log_level)
+
     return create_console_logger(
         name=init_context.logger_config["name"],
-        level=coerce_valid_log_level(init_context.logger_config["log_level"]),
+        level=coerce_valid_log_level(log_level),
     )
 
 
@@ -56,7 +57,6 @@ def colored_console_logger(init_context: "InitLoggerContext") -> logging.Logger:
             "log_level": Field(
                 str,
                 is_required=False,
-                default_value="INFO",
                 description="The logger's threshold.",
             ),
             "name": Field(
@@ -89,7 +89,8 @@ def json_logged_job():
                 hello_op()
 
     """
-    level = coerce_valid_log_level(init_context.logger_config["log_level"])
+    log_level = init_context.logger_config.get("log_level", init_context.default_log_level)
+    level = coerce_valid_log_level(log_level)
     name = init_context.logger_config["name"]
 
     klass = logging.getLoggerClass()