DagsterDocs
Source code for dagster.core.log_manager
import datetime
import itertools
import logging
from collections import OrderedDict, namedtuple
from dagster import check, seven
from dagster.core.utils import make_new_run_id
from dagster.utils import frozendict, merge_dicts
from dagster.utils.error import SerializableErrorInfo
DAGSTER_META_KEY = "dagster_meta"
PYTHON_LOGGING_LEVELS_MAPPING = frozendict(
OrderedDict({"CRITICAL": 50, "ERROR": 40, "WARNING": 30, "INFO": 20, "DEBUG": 10})
)
PYTHON_LOGGING_LEVELS_ALIASES = frozendict(OrderedDict({"FATAL": "CRITICAL", "WARN": "WARNING"}))
PYTHON_LOGGING_LEVELS_NAMES = frozenset(
[
level_name.lower()
for level_name in sorted(
list(PYTHON_LOGGING_LEVELS_MAPPING.keys()) + list(PYTHON_LOGGING_LEVELS_ALIASES.keys())
)
]
)
def _dump_value(value):
# dump namedtuples as objects instead of arrays
if isinstance(value, tuple) and hasattr(value, "_asdict"):
return seven.json.dumps(value._asdict())
return seven.json.dumps(value)
def construct_log_string(synth_props, logging_tags, message_props):
# Handle this explicitly
dagster_event = (
message_props["dagster_event"]._asdict() if "dagster_event" in message_props else {}
)
event_specific_data = dagster_event.get("event_specific_data")
error = ""
if hasattr(event_specific_data, "error") and isinstance(
event_specific_data.error, SerializableErrorInfo
):
if hasattr(event_specific_data, "error_display_string"):
error = "\n\n" + event_specific_data.error_display_string
else:
error = "\n\n" + event_specific_data.error.to_string()
log_source_prefix = (
"resource:%s" % logging_tags["resource_name"]
if "resource_name" in logging_tags
else message_props.get("pipeline_name", "system")
)
prefix = " - ".join(
filter(
None,
(
log_source_prefix,
synth_props.get("run_id"),
str(dagster_event["pid"]) if dagster_event.get("pid") is not None else None,
logging_tags.get("step_key"),
dagster_event.get("event_type_value"),
synth_props.get("orig_message"),
),
)
)
return prefix + error
def coerce_valid_log_level(log_level):
"""Convert a log level into an integer for consumption by the low-level Python logging API."""
if isinstance(log_level, int):
return log_level
check.str_param(log_level, "log_level")
check.invariant(
log_level.lower() in PYTHON_LOGGING_LEVELS_NAMES,
"Bad value for log level {level}: permissible values are {levels}.".format(
level=log_level,
levels=", ".join(
["'{}'".format(level_name.upper()) for level_name in PYTHON_LOGGING_LEVELS_NAMES]
),
),
)
log_level = PYTHON_LOGGING_LEVELS_ALIASES.get(log_level.upper(), log_level.upper())
return PYTHON_LOGGING_LEVELS_MAPPING[log_level]
[docs]class DagsterLogManager(namedtuple("_DagsterLogManager", "run_id logging_tags loggers")):
"""Centralized dispatch for logging from user code.
Handles the construction of uniform structured log messages and passes them through to the
underlying loggers.
An instance of the log manager is made available to solids as ``context.log``. Users should not
initialize instances of the log manager directly. To configure custom loggers, set the
``logger_defs`` on a :py:class:`ModeDefinition` for a pipeline.
The log manager supports standard convenience methods like those exposed by the Python standard
library :py:mod:`python:logging` module (i.e., within the body of a solid,
``context.log.{debug, info, warning, warn, error, critical, fatal}``).
The underlying integer API can also be called directly using, e.g.
``context.log.log(5, msg)``, and the log manager will delegate to the ``log`` method
defined on each of the loggers it manages.
User-defined custom log levels are not supported, and calls to, e.g.,
``context.log.trace`` or ``context.log.notice`` will result in hard exceptions **at runtime**.
"""
def __new__(cls, run_id, logging_tags, loggers):
return super(DagsterLogManager, cls).__new__(
cls,
run_id=check.opt_str_param(run_id, "run_id"),
logging_tags=check.dict_param(
logging_tags, "logging_tags", key_type=str, value_type=str
),
loggers=check.list_param(loggers, "loggers", of_type=logging.Logger),
)
[docs] def with_tags(self, **new_tags):
"""Add new tags in "new_tags" to the set of tags attached to this log manager instance, and
return a new DagsterLogManager with the merged set of tags.
Args:
tags (Dict[str,str]): Dictionary of tags
Returns:
DagsterLogManager: a new DagsterLogManager namedtuple with updated tags for the same
run ID and loggers.
"""
return self._replace(logging_tags=merge_dicts(self.logging_tags, new_tags))
def _prepare_message(self, orig_message, message_props):
check.str_param(orig_message, "orig_message")
check.dict_param(message_props, "message_props")
# These are todos to further align with the Python logging API
check.invariant(
"extra" not in message_props, "do not allow until explicit support is handled"
)
check.invariant(
"exc_info" not in message_props, "do not allow until explicit support is handled"
)
# Reserved keys in the message_props -- these are system generated.
check.invariant("orig_message" not in message_props, "orig_message reserved value")
check.invariant("message" not in message_props, "message reserved value")
check.invariant("log_message_id" not in message_props, "log_message_id reserved value")
check.invariant("log_timestamp" not in message_props, "log_timestamp reserved value")
log_message_id = make_new_run_id()
log_timestamp = datetime.datetime.utcnow().isoformat()
synth_props = {
"orig_message": orig_message,
"log_message_id": log_message_id,
"log_timestamp": log_timestamp,
"run_id": self.run_id,
}
# We first generate all props for the purpose of producing the semi-structured
# log message via _kv_messsage
all_props = dict(
itertools.chain(synth_props.items(), self.logging_tags.items(), message_props.items())
)
# So here we use the arbitrary key DAGSTER_META_KEY to store a dictionary of
# all the meta information that dagster injects into log message.
# The python logging module, in its infinite wisdom, actually takes all the
# keys in extra and unconditionally smashes them into the internal dictionary
# of the logging.LogRecord class. We used a reserved key here to avoid naming
# collisions with internal variables of the LogRecord class.
# See __init__.py:363 (makeLogRecord) in the python 3.6 logging module source
# for the gory details.
return (
construct_log_string(synth_props, self.logging_tags, message_props),
{DAGSTER_META_KEY: all_props},
)
def _log(self, level, orig_message, message_props):
"""Invoke the underlying loggers for a given log level.
Args:
level (Union[str, int]): An integer represeting a Python logging level or one of the
standard Python string representations of a logging level.
orig_message (str): The log message generated in user code.
message_props (dict): Additional properties for the structured log message.
"""
if not self.loggers:
return
level = coerce_valid_log_level(level)
message, extra = self._prepare_message(orig_message, message_props)
for logger_ in self.loggers:
logger_.log(level, message, extra=extra)
[docs] def log(self, level, msg, **kwargs):
"""Invoke the underlying loggers for a given integer log level.
Args:
level (int): An integer represeting a Python logging level.
orig_message (str): The message to log.
"""
check.str_param(msg, "msg")
check.int_param(level, "level")
return self._log(level, msg, kwargs)
[docs] def debug(self, msg, **kwargs):
"""Log at the ``logging.DEBUG`` level.
The message will be automatically adorned with contextual information about the name
of the pipeline, the name of the solid, etc., so it is generally unnecessary to include
this type of information in the log message.
You can optionally additional key-value pairs to an individual log message using the kwargs
to this method.
Args:
msg (str): The message to log.
**kwargs (Optional[Any]): Any additional key-value pairs for only this log message.
"""
check.str_param(msg, "msg")
return self._log(logging.DEBUG, msg, kwargs)
[docs] def info(self, msg, **kwargs):
"""Log at the ``logging.INFO`` level.
See :py:meth:`~DagsterLogManager.debug`.
"""
check.str_param(msg, "msg")
return self._log(logging.INFO, msg, kwargs)
[docs] def warning(self, msg, **kwargs):
"""Log at the ``logging.WARNING`` level.
See :py:meth:`~DagsterLogManager.debug`.
"""
check.str_param(msg, "msg")
return self._log(logging.WARNING, msg, kwargs)
# Define the alias .warn()
warn = warning
"""Alias for :py:meth:`~DagsterLogManager.warning`"""
[docs] def error(self, msg, **kwargs):
"""Log at the ``logging.ERROR`` level.
See :py:meth:`~DagsterLogManager.debug`.
"""
check.str_param(msg, "msg")
return self._log(logging.ERROR, msg, kwargs)
[docs] def critical(self, msg, **kwargs):
"""Log at the ``logging.CRITICAL`` level.
See :py:meth:`~DagsterLogManager.debug`.
"""
check.str_param(msg, "msg")
return self._log(logging.CRITICAL, msg, kwargs)
# Define the alias .fatal()
fatal = critical
"""Alias for :py:meth:`~DagsterLogManager.critical`"""