Logging

zenml.logging

Modules

step_logging

ZenML logging handler.

Classes

ArtifactStoreHandler(storage: PipelineLogsStorage)

Bases: Handler

Handler that writes log messages to artifact store storage.

Initialize the handler with a storage instance.

Parameters:

Name	Type	Description	Default
storage	PipelineLogsStorage	The PipelineLogsStorage instance to write to.	required

Source code in src/zenml/logging/step_logging.py

def __init__(self, storage: "PipelineLogsStorage"):
    """Initialize the handler with a storage instance.

    Args:
        storage: The PipelineLogsStorage instance to write to.
    """
    super().__init__()
    self.storage = storage

    # Get storage log level from environment
    self.setLevel(get_storage_log_level().value)

Functions

emit(record: logging.LogRecord) -> None

Emit a log record to the storage.

Parameters:

Name	Type	Description	Default
record	LogRecord	The log record to emit.	required

Source code in src/zenml/logging/step_logging.py

def emit(self, record: logging.LogRecord) -> None:
    """Emit a log record to the storage.

    Args:
        record: The log record to emit.
    """
    try:
        # Get level enum
        level = LoggingLevels.__members__.get(record.levelname.upper())

        # Get the message
        message = self.format(record)
        message = remove_ansi_escape_codes(message).rstrip()

        # Check if message needs to be chunked
        message_bytes = message.encode("utf-8")
        if len(message_bytes) <= DEFAULT_MESSAGE_SIZE:
            # Message is small enough, emit as-is
            log_record = LogEntry.model_construct(
                message=message,
                name=record.name,
                level=level,
                timestamp=utc_now(),
                module=record.module,
                filename=record.filename,
                lineno=record.lineno,
            )
            json_line = log_record.model_dump_json(exclude_none=True)
            self.storage.write(json_line)
        else:
            # Message is too large, split into chunks and emit each one
            chunks = self._split_to_chunks(message)
            entry_id = uuid4()
            for i, chunk in enumerate(chunks):
                log_record = LogEntry.model_construct(
                    message=chunk,
                    name=record.name,
                    level=level,
                    module=record.module,
                    filename=record.filename,
                    lineno=record.lineno,
                    timestamp=utc_now(),
                    chunk_index=i,
                    total_chunks=len(chunks),
                    id=entry_id,
                )

                json_line = log_record.model_dump_json(exclude_none=True)
                self.storage.write(json_line)
    except Exception:
        pass

LogEntry

Bases: BaseModel

A structured log entry with parsed information.

PipelineLogsStorage(logs_uri: str, artifact_store: BaseArtifactStore, max_queue_size: int = LOGS_STORAGE_MAX_QUEUE_SIZE, queue_timeout: int = LOGS_STORAGE_QUEUE_TIMEOUT, write_interval: int = LOGS_WRITE_INTERVAL_SECONDS, merge_files_interval: int = LOGS_MERGE_INTERVAL_SECONDS)

Helper class which buffers and stores logs to a given URI using a background thread.

Initialization.

Parameters:

Name	Type	Description	Default
logs_uri	str	the URI of the log file or folder.	required
artifact_store	BaseArtifactStore	Artifact Store from the current step context	required
max_queue_size	int	maximum number of individual messages to queue.	LOGS_STORAGE_MAX_QUEUE_SIZE
queue_timeout	int	timeout in seconds for putting items in queue when full. - Positive value: Wait N seconds, then drop logs if queue still full - Negative value: Block indefinitely until queue has space (never drop logs)	LOGS_STORAGE_QUEUE_TIMEOUT
write_interval	int	the amount of seconds before the created files get written to the artifact store.	LOGS_WRITE_INTERVAL_SECONDS
merge_files_interval	int	the amount of seconds before the created files get merged into a single file.	LOGS_MERGE_INTERVAL_SECONDS

Source code in src/zenml/logging/step_logging.py

def __init__(
    self,
    logs_uri: str,
    artifact_store: "BaseArtifactStore",
    max_queue_size: int = LOGS_STORAGE_MAX_QUEUE_SIZE,
    queue_timeout: int = LOGS_STORAGE_QUEUE_TIMEOUT,
    write_interval: int = LOGS_WRITE_INTERVAL_SECONDS,
    merge_files_interval: int = LOGS_MERGE_INTERVAL_SECONDS,
) -> None:
    """Initialization.

    Args:
        logs_uri: the URI of the log file or folder.
        artifact_store: Artifact Store from the current step context
        max_queue_size: maximum number of individual messages to queue.
        queue_timeout: timeout in seconds for putting items in queue when full.
            - Positive value: Wait N seconds, then drop logs if queue still full
            - Negative value: Block indefinitely until queue has space (never drop logs)
        write_interval: the amount of seconds before the created files
            get written to the artifact store.
        merge_files_interval: the amount of seconds before the created files
            get merged into a single file.
    """
    # Parameters
    self.logs_uri = logs_uri
    self.max_queue_size = max_queue_size
    self.queue_timeout = queue_timeout
    self.write_interval = write_interval
    self.merge_files_interval = merge_files_interval

    # State
    self.artifact_store = artifact_store

    # Immutable filesystems state
    self.last_merge_time = time.time()

    # Queue and log storage thread for async processing
    self.log_queue: queue.Queue[str] = queue.Queue(maxsize=max_queue_size)
    self.log_storage_thread: Optional[threading.Thread] = None
    self.shutdown_event = threading.Event()
    self.merge_event = threading.Event()

    # Start the log storage thread
    self._start_log_storage_thread()

Functions

merge_log_files(merge_all_files: bool = False) -> None

Merges all log files into one in the given URI.

Called on the logging context exit.

Parameters:

Name	Type	Description	Default
merge_all_files	bool	whether to merge all files or only raw files	False

Source code in src/zenml/logging/step_logging.py

def merge_log_files(self, merge_all_files: bool = False) -> None:
    """Merges all log files into one in the given URI.

    Called on the logging context exit.

    Args:
        merge_all_files: whether to merge all files or only raw files
    """
    from zenml.artifacts.utils import (
        _load_file_from_artifact_store,
    )

    # If the artifact store is immutable, merge the log files
    if self.artifact_store.config.IS_IMMUTABLE_FILESYSTEM:
        merged_file_suffix = "_merged"
        files_ = self.artifact_store.listdir(self.logs_uri)
        if not merge_all_files:
            # already merged files will not be merged again
            files_ = [
                f for f in files_ if merged_file_suffix not in str(f)
            ]
        file_name_ = self._get_timestamped_filename(
            suffix=merged_file_suffix
        )
        if len(files_) > 1:
            files_.sort()
            logger.debug("Log files count: %s", len(files_))

            missing_files = set()
            # dump all logs to a local file first
            with self.artifact_store.open(
                os.path.join(self.logs_uri, file_name_), "w"
            ) as merged_file:
                for file in files_:
                    try:
                        merged_file.write(
                            str(
                                _load_file_from_artifact_store(
                                    os.path.join(self.logs_uri, str(file)),
                                    artifact_store=self.artifact_store,
                                    mode="r",
                                )
                            )
                        )
                    except DoesNotExistException:
                        missing_files.add(file)

            # clean up left over files
            for file in files_:
                if file not in missing_files:
                    self.artifact_store.remove(
                        os.path.join(self.logs_uri, str(file))
                    )

        # Update the last merge time
        self.last_merge_time = time.time()

send_merge_event() -> None

Send a merge event to the log storage thread.

Source code in src/zenml/logging/step_logging.py

def send_merge_event(self) -> None:
    """Send a merge event to the log storage thread."""
    self.merge_event.set()

write(text: str) -> None

Main write method that sends individual messages directly to queue.

Parameters:

Name	Type	Description	Default
text	str	the incoming string.	required

Source code in src/zenml/logging/step_logging.py

def write(self, text: str) -> None:
    """Main write method that sends individual messages directly to queue.

    Args:
        text: the incoming string.
    """
    # Skip empty lines
    if text == "\n":
        return

    # If the current thread is the log storage thread, do nothing
    # to prevent recursion when the storage thread itself generates logs
    if (
        self.log_storage_thread
        and threading.current_thread() == self.log_storage_thread
    ):
        return

    # If the current thread is the fsspec IO thread, do nothing
    if self._is_fsspec_io_thread:
        return

    try:
        # Send individual message directly to queue
        if not self.shutdown_event.is_set():
            try:
                if self.queue_timeout < 0:
                    # Negative timeout = block indefinitely until queue has space
                    # Guarantees no log loss but may hang application
                    self.log_queue.put(text)
                else:
                    # Positive timeout = wait specified time then drop logs
                    # Prevents application hanging but may lose logs
                    self.log_queue.put(text, timeout=self.queue_timeout)
            except queue.Full:
                # This only happens with positive timeout
                # Queue is full - just skip this message to avoid blocking
                # Better to drop logs than hang the application
                pass

    except Exception:
        # Silently ignore errors to prevent recursion
        pass

write_buffer(buffer_to_write: List[str]) -> None

Write the given buffer to file. This runs in the log storage thread.

Parameters:

Name	Type	Description	Default
buffer_to_write	List[str]	The buffer contents to write to file.	required

Source code in src/zenml/logging/step_logging.py

def write_buffer(self, buffer_to_write: List[str]) -> None:
    """Write the given buffer to file. This runs in the log storage thread.

    Args:
        buffer_to_write: The buffer contents to write to file.
    """
    if not buffer_to_write:
        return

    try:
        # If the artifact store is immutable, write the buffer to a new file
        if self.artifact_store.config.IS_IMMUTABLE_FILESYSTEM:
            _logs_uri = self._get_timestamped_filename()
            with self.artifact_store.open(
                os.path.join(
                    self.logs_uri,
                    _logs_uri,
                ),
                "w",
            ) as file:
                for message in buffer_to_write:
                    file.write(f"{message}\n")

        # If the artifact store is mutable, append the buffer to the existing file
        else:
            with self.artifact_store.open(self.logs_uri, "a") as file:
                for message in buffer_to_write:
                    file.write(f"{message}\n")
            self.artifact_store._remove_previous_file_versions(
                self.logs_uri
            )

    except Exception as e:
        logger.error("Error in log storage thread: %s", e)

PipelineLogsStorageContext(logs_uri: str, artifact_store: BaseArtifactStore, prepend_step_name: bool = True)

Context manager which collects logs during pipeline run execution.

Initializes and prepares a storage object.

Parameters:

Name	Type	Description	Default
logs_uri	str	the URI of the logs file.	required
artifact_store	BaseArtifactStore	Artifact Store from the current pipeline run context.	required
prepend_step_name	bool	Whether to prepend the step name to the logs.	True

Source code in src/zenml/logging/step_logging.py

def __init__(
    self,
    logs_uri: str,
    artifact_store: "BaseArtifactStore",
    prepend_step_name: bool = True,
) -> None:
    """Initializes and prepares a storage object.

    Args:
        logs_uri: the URI of the logs file.
        artifact_store: Artifact Store from the current pipeline run context.
        prepend_step_name: Whether to prepend the step name to the logs.
    """
    # Create the storage object
    self.storage = PipelineLogsStorage(
        logs_uri=logs_uri, artifact_store=artifact_store
    )

    # Create the handler object
    self.artifact_store_handler: ArtifactStoreHandler = (
        ArtifactStoreHandler(self.storage)
    )

    # Additional configuration
    self.prepend_step_name = prepend_step_name
    self.original_step_names_in_console: Optional[bool] = None
    self._original_root_level: Optional[int] = None

Functions

Functions

fetch_log_records(zen_store: BaseZenStore, artifact_store_id: Union[str, UUID], logs_uri: str) -> List[LogEntry]

Fetches log entries.

Parameters:

Name	Type	Description	Default
zen_store	BaseZenStore	The store in which the artifact is stored.	required
artifact_store_id	Union[str, UUID]	The ID of the artifact store.	required
logs_uri	str	The URI of the artifact (file or directory).	required

Returns:

Type	Description
List[LogEntry]	List of log entries.

Source code in src/zenml/logging/step_logging.py

def fetch_log_records(
    zen_store: "BaseZenStore",
    artifact_store_id: Union[str, UUID],
    logs_uri: str,
) -> List[LogEntry]:
    """Fetches log entries.

    Args:
        zen_store: The store in which the artifact is stored.
        artifact_store_id: The ID of the artifact store.
        logs_uri: The URI of the artifact (file or directory).

    Returns:
        List of log entries.
    """
    log_entries = []

    for line in _stream_logs_line_by_line(
        zen_store, artifact_store_id, logs_uri
    ):
        if log_entry := parse_log_entry(line):
            log_entries.append(log_entry)

        if len(log_entries) >= MAX_ENTRIES_PER_REQUEST:
            break

    return log_entries

parse_log_entry(log_line: str) -> Optional[LogEntry]

Parse a single log entry into a LogEntry object.

Handles two formats: 1. JSON format: {"timestamp": "...", "level": "...", "message": "...", "location": "..."} Uses Pydantic's model_validate_json for automatic parsing and validation. 2. Plain text: Any other text (defaults to INFO level)

Parameters:

Name	Type	Description	Default
log_line	str	A single log line to parse	required

Returns:

Type	Description
Optional[LogEntry]	LogEntry object. For JSON logs, all fields are validated and parsed automatically.
Optional[LogEntry]	For plain text logs, only message is populated with INFO level default.
Optional[LogEntry]	Returns None only for empty lines.

Source code in src/zenml/logging/step_logging.py

def parse_log_entry(log_line: str) -> Optional[LogEntry]:
    """Parse a single log entry into a LogEntry object.

    Handles two formats:
    1. JSON format: {"timestamp": "...", "level": "...", "message": "...", "location": "..."}
       Uses Pydantic's model_validate_json for automatic parsing and validation.
    2. Plain text: Any other text (defaults to INFO level)

    Args:
        log_line: A single log line to parse

    Returns:
        LogEntry object. For JSON logs, all fields are validated and parsed automatically.
        For plain text logs, only message is populated with INFO level default.
        Returns None only for empty lines.
    """
    stripped_line = log_line.strip()
    if not stripped_line:
        return None

    # Try to parse JSON format first
    if stripped_line.startswith("{") and stripped_line.endswith("}"):
        try:
            return LogEntry.model_validate_json(stripped_line)
        except Exception:
            # If JSON parsing or validation fails, treat as plain text
            pass

    # For any other format (plain text), create LogEntry with defaults
    return LogEntry(
        message=stripped_line,
        name=None,  # No logger name available for plain text logs
        level=LoggingLevels.INFO,  # Default level for plain text logs
        timestamp=None,  # No timestamp available for plain text logs
    )

prepare_logs_uri(artifact_store: BaseArtifactStore, step_name: Optional[str] = None, log_key: Optional[str] = None) -> str

Generates and prepares a URI for the log file or folder for a step.

Parameters:

Name	Type	Description	Default
artifact_store	BaseArtifactStore	The artifact store on which the artifact will be stored.	required
step_name	Optional[str]	Name of the step. Skipped for global pipeline run logs.	None
log_key	Optional[str]	The unique identification key of the log file.	None

Returns:

Type	Description
str	The URI of the log storage (file or folder).

Source code in src/zenml/logging/step_logging.py

def prepare_logs_uri(
    artifact_store: "BaseArtifactStore",
    step_name: Optional[str] = None,
    log_key: Optional[str] = None,
) -> str:
    """Generates and prepares a URI for the log file or folder for a step.

    Args:
        artifact_store: The artifact store on which the artifact will be stored.
        step_name: Name of the step. Skipped for global pipeline run logs.
        log_key: The unique identification key of the log file.

    Returns:
        The URI of the log storage (file or folder).
    """
    if log_key is None:
        log_key = str(uuid4())

    subfolder = step_name or PIPELINE_RUN_LOGS_FOLDER
    logs_base_uri = os.path.join(artifact_store.path, subfolder, "logs")

    # Create the dir
    if not artifact_store.exists(logs_base_uri):
        artifact_store.makedirs(logs_base_uri)

    # Delete the file if it already exists
    if artifact_store.config.IS_IMMUTABLE_FILESYSTEM:
        logs_uri = os.path.join(logs_base_uri, log_key)
        if artifact_store.exists(logs_uri):
            logger.warning(
                f"Logs directory {logs_uri} already exists! Removing old log directory..."
            )
            artifact_store.rmtree(logs_uri)

        artifact_store.makedirs(logs_uri)
    else:
        logs_uri = os.path.join(logs_base_uri, f"{log_key}{LOGS_EXTENSION}")
        if artifact_store.exists(logs_uri):
            logger.warning(
                f"Logs file {logs_uri} already exists! Removing old log file..."
            )
            artifact_store.remove(logs_uri)

    return sanitize_remote_path(logs_uri)

remove_ansi_escape_codes(text: str) -> str

Auxiliary function to remove ANSI escape codes from a given string.

Parameters:

Name	Type	Description	Default
text	str	the input string	required

Returns:

Type	Description
str	the version of the input string where the escape codes are removed.

Source code in src/zenml/logging/step_logging.py

def remove_ansi_escape_codes(text: str) -> str:
    """Auxiliary function to remove ANSI escape codes from a given string.

    Args:
        text: the input string

    Returns:
        the version of the input string where the escape codes are removed.
    """
    return ansi_escape.sub("", text)

setup_orchestrator_logging(run_id: UUID, deployment: PipelineDeploymentResponse, logs_response: Optional[LogsResponse] = None) -> Any

Set up logging for an orchestrator environment.

This function can be reused by different orchestrators to set up consistent logging behavior.

Parameters:

Name	Type	Description	Default
run_id	UUID	The pipeline run ID.	required
deployment	PipelineDeploymentResponse	The deployment of the pipeline run.	required
logs_response	Optional[LogsResponse]	The logs response to continue from.	None

Returns:

Type	Description
Any	The logs context (PipelineLogsStorageContext)

Source code in src/zenml/logging/step_logging.py

def setup_orchestrator_logging(
    run_id: UUID,
    deployment: "PipelineDeploymentResponse",
    logs_response: Optional[LogsResponse] = None,
) -> Any:
    """Set up logging for an orchestrator environment.

    This function can be reused by different orchestrators to set up
    consistent logging behavior.

    Args:
        run_id: The pipeline run ID.
        deployment: The deployment of the pipeline run.
        logs_response: The logs response to continue from.

    Returns:
        The logs context (PipelineLogsStorageContext)
    """
    try:
        logging_enabled = True

        if handle_bool_env_var(ENV_ZENML_DISABLE_PIPELINE_LOGS_STORAGE, False):
            logging_enabled = False
        else:
            if (
                deployment.pipeline_configuration.enable_pipeline_logs
                is not None
            ):
                logging_enabled = (
                    deployment.pipeline_configuration.enable_pipeline_logs
                )

        if not logging_enabled:
            return nullcontext()

        # Fetch the active stack
        client = Client()
        active_stack = client.active_stack

        if logs_response:
            logs_uri = logs_response.uri
        else:
            logs_uri = prepare_logs_uri(
                artifact_store=active_stack.artifact_store,
            )
            logs_model = LogsRequest(
                uri=logs_uri,
                source="orchestrator",
                artifact_store_id=active_stack.artifact_store.id,
            )

            # Add orchestrator logs to the pipeline run
            try:
                run_update = PipelineRunUpdate(add_logs=[logs_model])
                client.zen_store.update_run(
                    run_id=run_id, run_update=run_update
                )
            except Exception as e:
                logger.error(
                    f"Failed to add orchestrator logs to the run {run_id}: {e}"
                )
                raise e

        return PipelineLogsStorageContext(
            logs_uri=logs_uri,
            artifact_store=active_stack.artifact_store,
            prepend_step_name=False,
        )
    except Exception as e:
        logger.error(
            f"Failed to setup orchestrator logging for run {run_id}: {e}"
        )
        return nullcontext()