from typing import Any, Dict, NamedTuple, Optional, Union
import dagster._check as check
from dagster._annotations import PublicAttr, public
from dagster._core.errors import DagsterInvariantViolationError
from dagster._core.events import DagsterEvent
from dagster._core.utils import coerce_valid_log_level
from dagster._serdes.serdes import (
DefaultNamedTupleSerializer,
WhitelistMap,
deserialize_json_to_dagster_namedtuple,
register_serdes_tuple_fallbacks,
serialize_dagster_namedtuple,
whitelist_for_serdes,
)
from dagster._utils.error import SerializableErrorInfo
from dagster._utils.log import (
JsonEventLoggerHandler,
StructuredLoggerHandler,
StructuredLoggerMessage,
construct_single_handler_logger,
)
class EventLogEntrySerializer(DefaultNamedTupleSerializer):
@classmethod
def value_to_storage_dict(
cls,
value: NamedTuple,
whitelist_map: WhitelistMap,
descent_path: str,
) -> Dict[str, Any]:
storage_dict = super().value_to_storage_dict(value, whitelist_map, descent_path)
# include an empty string for the message field to allow older versions of dagster to load the events
storage_dict["message"] = ""
return storage_dict
[docs]@whitelist_for_serdes(serializer=EventLogEntrySerializer)
class EventLogEntry(
NamedTuple(
"_EventLogEntry",
[
("error_info", PublicAttr[Optional[SerializableErrorInfo]]),
("level", PublicAttr[Union[str, int]]),
("user_message", PublicAttr[str]),
("run_id", PublicAttr[str]),
("timestamp", PublicAttr[float]),
("step_key", PublicAttr[Optional[str]]),
("pipeline_name", Optional[str]),
("dagster_event", PublicAttr[Optional[DagsterEvent]]),
],
)
):
"""Entries in the event log.
Users should not instantiate this object directly. These entries may originate from the logging machinery (DagsterLogManager/context.log), from
framework events (e.g. EngineEvent), or they may correspond to events yielded by user code
(e.g. Output).
Args:
error_info (Optional[SerializableErrorInfo]): Error info for an associated exception, if
any, as generated by serializable_error_info_from_exc_info and friends.
level (Union[str, int]): The Python log level at which to log this event. Note that
framework and user code events are also logged to Python logging. This value may be an
integer or a (case-insensitive) string member of PYTHON_LOGGING_LEVELS_NAMES.
user_message (str): For log messages, this is the user-generated message.
run_id (str): The id of the run which generated this event.
timestamp (float): The Unix timestamp of this event.
step_key (Optional[str]): The step key for the step which generated this event. Some events
are generated outside of a step context.
job_name (Optional[str]): The job which generated this event. Some events are
generated outside of a job context.
dagster_event (Optional[DagsterEvent]): For framework and user events, the associated
structured event.
"""
def __new__(
cls,
error_info,
level,
user_message,
run_id,
timestamp,
step_key=None,
pipeline_name=None,
dagster_event=None,
job_name=None,
):
if pipeline_name and job_name:
raise DagsterInvariantViolationError(
"Provided both `pipeline_name` and `job_name` parameters to `EventLogEntry` "
"initialization. Please provide only one or the other."
)
pipeline_name = pipeline_name or job_name
return super(EventLogEntry, cls).__new__(
cls,
check.opt_inst_param(error_info, "error_info", SerializableErrorInfo),
coerce_valid_log_level(level),
check.str_param(user_message, "user_message"),
check.str_param(run_id, "run_id"),
check.float_param(timestamp, "timestamp"),
check.opt_str_param(step_key, "step_key"),
check.opt_str_param(pipeline_name, "pipeline_name"),
check.opt_inst_param(dagster_event, "dagster_event", DagsterEvent),
)
@public # type: ignore
@property
def is_dagster_event(self) -> bool:
return bool(self.dagster_event)
@public # type: ignore
@property
def job_name(self) -> Optional[str]:
return self.pipeline_name
@public # type: ignore
def get_dagster_event(self) -> DagsterEvent:
if not isinstance(self.dagster_event, DagsterEvent):
check.failed(
"Not a dagster event, check is_dagster_event before calling get_dagster_event",
)
return self.dagster_event
def to_json(self):
return serialize_dagster_namedtuple(self)
@staticmethod
def from_json(json_str):
return deserialize_json_to_dagster_namedtuple(json_str)
@public # type: ignore
@property
def dagster_event_type(self):
return self.dagster_event.event_type if self.dagster_event else None
@public # type: ignore
@property
def message(self) -> str:
"""
Return the message from the structured DagsterEvent if present, fallback to user_message
"""
if self.is_dagster_event:
msg = self.get_dagster_event().message
if msg is not None:
return msg
return self.user_message
def construct_event_record(logger_message: StructuredLoggerMessage) -> EventLogEntry:
check.inst_param(logger_message, "logger_message", StructuredLoggerMessage)
return EventLogEntry(
level=logger_message.level,
user_message=logger_message.meta["orig_message"],
run_id=logger_message.meta["run_id"],
timestamp=logger_message.record.created,
step_key=logger_message.meta.get("step_key"),
job_name=logger_message.meta.get("pipeline_name"),
dagster_event=logger_message.meta.get("dagster_event"),
error_info=None,
)
def construct_event_logger(event_record_callback):
"""
Callback receives a stream of event_records. Piggybacks on the logging machinery.
"""
check.callable_param(event_record_callback, "event_record_callback")
return construct_single_handler_logger(
"event-logger",
"debug",
StructuredLoggerHandler(
lambda logger_message: event_record_callback(construct_event_record(logger_message))
),
)
def construct_json_event_logger(json_path):
"""Record a stream of event records to json"""
check.str_param(json_path, "json_path")
return construct_single_handler_logger(
"json-event-record-logger",
"debug",
JsonEventLoggerHandler(
json_path,
lambda record: construct_event_record(
StructuredLoggerMessage(
name=record.name,
message=record.msg,
level=record.levelno,
meta=record.dagster_meta,
record=record,
)
),
),
)
register_serdes_tuple_fallbacks(
{
# These were originally distinguished from each other but ended up being empty subclasses
# of EventLogEntry -- instead of using the subclasses we were relying on
# EventLogEntry.is_dagster_event to distinguish events that originate in the logging
# machinery from events that are yielded by user code
"DagsterEventRecord": EventLogEntry,
"LogMessageRecord": EventLogEntry,
# renamed EventRecord -> EventLogEntry
"EventRecord": EventLogEntry,
}
)