Log Stores

zenml.log_stores

Implements the log stores for ZenML.

Attributes

__all__ = ['ArtifactLogStore', 'BaseLogStore', 'BaseLogStoreConfig', 'BaseLogStoreFlavor', 'DatadogLogStore', 'DatadogLogStoreConfig', 'DatadogLogStoreFlavor', 'OtelLogStore', 'OtelLogStoreConfig', 'OtelLogStoreFlavor'] module-attribute

Classes

ArtifactLogStore(artifact_store: BaseArtifactStore, *args: Any, **kwargs: Any)

Bases: OtelLogStore

Log store that saves logs to the artifact store.

This implementation extends OtelLogStore and uses the ArtifactLogExporter to write logs to the artifact store. Inherits all OTEL infrastructure including shared BatchLogRecordProcessor and routing.

Initialize the artifact log store.

Parameters:

Name	Type	Description	Default
artifact_store	BaseArtifactStore	The artifact store to use for logging.	required
*args	Any	Positional arguments for the base class.	()
**kwargs	Any	Keyword arguments for the base class.	{}

Source code in src/zenml/log_stores/artifact/artifact_log_store.py

def __init__(
    self, artifact_store: "BaseArtifactStore", *args: Any, **kwargs: Any
) -> None:
    """Initialize the artifact log store.

    Args:
        artifact_store: The artifact store to use for logging.
        *args: Positional arguments for the base class.
        **kwargs: Keyword arguments for the base class.
    """
    super().__init__(*args, **kwargs)
    self._artifact_store = artifact_store

Attributes

config: ArtifactLogStoreConfig property

Returns the configuration of the artifact log store.

Returns:

Type	Description
ArtifactLogStoreConfig	The configuration.

origin_class: Type[ArtifactLogStoreOrigin] property

Class of the origin.

Returns:

Type	Description
Type[ArtifactLogStoreOrigin]	The class of the origin.

Methods:

cleanup() -> None

Cleanup the artifact log store.

This method is called to ensure that the artifact log store is cleaned up.

Source code in src/zenml/log_stores/artifact/artifact_log_store.py

def cleanup(self) -> None:
    """Cleanup the artifact log store.

    This method is called to ensure that the artifact log store is cleaned up.
    """
    self._artifact_store.cleanup()

fetch(logs_model: LogsResponse, start: Optional[str] = None, limit: Optional[int] = None, before: Optional[str] = None, after: Optional[str] = None, filter_: Optional[LogsEntriesFilter] = None) -> LogsEntriesResponse

Fetch a batch of log entries from the artifact store.

Parameters:

Name	Type	Description	Default
logs_model	LogsResponse	The logs model containing uri and artifact_store_id.	required
start	Optional[str]	Initial position. Only None and oldest are supported.	None
limit	Optional[int]	Maximum number of log entries to return.	None
before	Optional[str]	Not supported.	None
after	Optional[str]	Not supported.	None
filter_	Optional[LogsEntriesFilter]	Not supported.	None

Returns:

Type	Description
LogsEntriesResponse	The oldest entries up to the limit, without cursors.

Raises:

Type	Description
ValueError	If the logs model is invalid, the limit is not positive, or unsupported parameters are provided.

Source code in src/zenml/log_stores/artifact/artifact_log_store.py

def fetch(
    self,
    logs_model: "LogsResponse",
    start: Optional[str] = None,
    limit: Optional[int] = None,
    before: Optional[str] = None,
    after: Optional[str] = None,
    filter_: Optional[LogsEntriesFilter] = None,
) -> LogsEntriesResponse:
    """Fetch a batch of log entries from the artifact store.

    Args:
        logs_model: The logs model containing uri and artifact_store_id.
        start: Initial position. Only None and `oldest` are supported.
        limit: Maximum number of log entries to return.
        before: Not supported.
        after: Not supported.
        filter_: Not supported.

    Returns:
        The oldest entries up to the limit, without cursors.

    Raises:
        ValueError: If the logs model is invalid, the limit is not positive,
            or unsupported parameters are provided.
    """
    if not logs_model.uri:
        raise ValueError(
            "logs_model.uri is required for ArtifactLogStore.fetch()"
        )

    if not logs_model.artifact_store_id:
        raise ValueError(
            "logs_model.artifact_store_id is required "
            "for ArtifactLogStore.fetch()"
        )

    if logs_model.artifact_store_id != self._artifact_store.id:
        raise ValueError(
            "logs_model.artifact_store_id does not match the artifact store "
            "id of the log store."
        )

    unallowed = (
        start not in (None, "oldest")
        or before is not None
        or after is not None
    )
    if filter_ is not None:
        unallowed = unallowed or any(
            (
                filter_.search,
                filter_.level,
                filter_.since,
                filter_.until,
            )
        )
    if unallowed:
        raise ValueError(
            "The artifact log store only reads a log file from its start "
            "and serves what fits in one response. Drop start, before, "
            "after, search, level, since, and until, and narrow the result down where it is "
            "displayed."
        )

    return LogsEntriesResponse(
        items=fetch_log_records(
            artifact_store=self._artifact_store,
            logs_uri=logs_model.uri,
            limit=self.resolve_limit(limit),
        )
    )

from_artifact_store(artifact_store: BaseArtifactStore) -> ArtifactLogStore classmethod

Creates an artifact log store from an artifact store.

Parameters:

Name	Type	Description	Default
artifact_store	BaseArtifactStore	The artifact store to create the log store from.	required

Returns:

Type	Description
ArtifactLogStore	The created artifact log store.

Source code in src/zenml/log_stores/artifact/artifact_log_store.py

@classmethod
def from_artifact_store(
    cls, artifact_store: "BaseArtifactStore"
) -> "ArtifactLogStore":
    """Creates an artifact log store from an artifact store.

    Args:
        artifact_store: The artifact store to create the log store from.

    Returns:
        The created artifact log store.
    """
    return cls(
        artifact_store=artifact_store,
        id=artifact_store.id,
        name="default",
        config=ArtifactLogStoreConfig(endpoint=artifact_store.path),
        flavor="artifact",
        type=StackComponentType.LOG_STORE,
        user=artifact_store.user,
        created=artifact_store.created,
        updated=artifact_store.updated,
    )

get_exporter() -> LogRecordExporter

Get the artifact log exporter for this log store.

Returns:

Type	Description
LogRecordExporter	The ArtifactLogExporter instance.

Source code in src/zenml/log_stores/artifact/artifact_log_store.py

def get_exporter(self) -> "LogRecordExporter":
    """Get the artifact log exporter for this log store.

    Returns:
        The ArtifactLogExporter instance.
    """
    from zenml.log_stores.artifact.artifact_log_exporter import (
        ArtifactLogExporter,
    )

    return ArtifactLogExporter(artifact_store=self._artifact_store)

BaseLogStore(*args: Any, **kwargs: Any)

Bases: StackComponent, ABC

Base class for all ZenML log stores.

A log store is responsible for collecting, storing, and retrieving logs during pipeline and step execution. Different implementations may store logs in different backends (artifact store, OpenTelemetry, Datadog, etc.).

Initialize the log store.

Parameters:

Name	Type	Description	Default
*args	Any	Positional arguments for the base class.	()
**kwargs	Any	Keyword arguments for the base class.	{}

Source code in src/zenml/log_stores/base_log_store.py

def __init__(self, *args: Any, **kwargs: Any) -> None:
    """Initialize the log store.

    Args:
        *args: Positional arguments for the base class.
        **kwargs: Keyword arguments for the base class.
    """
    super().__init__(*args, **kwargs)
    self._origins: Dict[str, BaseLogStoreOrigin] = {}
    self._lock = threading.RLock()

Attributes

config: BaseLogStoreConfig property

Returns the configuration of the log store.

Returns:

Type	Description
BaseLogStoreConfig	The configuration.

default_query_size: int property

Default page size when no limit is specified.

Returns:

Type	Description
int	The default number of entries per page.

origin_class: Type[BaseLogStoreOrigin] property

Class of the origin.

Returns:

Type	Description
Type[BaseLogStoreOrigin]	The class of the origin used with this log store.

Methods:

deregister_origin(origin: BaseLogStoreOrigin, blocking: bool = True) -> None

Deregister an origin previously registered with the log store.

If no other origins are left, the log store will be flushed. The blocking parameter determines whether to block until the flush is complete.

Parameters:

Name	Type	Description	Default
origin	BaseLogStoreOrigin	The origin to deregister.	required
blocking	bool	Whether to block until the deregistration is complete and all logs are flushed if this is the last origin registered.	True

Source code in src/zenml/log_stores/base_log_store.py

def deregister_origin(
    self,
    origin: BaseLogStoreOrigin,
    blocking: bool = True,
) -> None:
    """Deregister an origin previously registered with the log store.

    If no other origins are left, the log store will be flushed. The
    `blocking` parameter determines whether to block until the flush is
    complete.

    Args:
        origin: The origin to deregister.
        blocking: Whether to block until the deregistration is complete
            and all logs are flushed if this is the last origin registered.
    """
    with self._lock:
        if origin.name not in self._origins:
            return
        self._release_origin(origin)
        del self._origins[origin.name]
        if len(self._origins) == 0:
            self.flush(blocking=blocking)

emit(origin: BaseLogStoreOrigin, record: logging.LogRecord, metadata: Optional[Dict[str, Any]] = None) -> None abstractmethod

Process a log record from the logging system.

Parameters:

Name	Type	Description	Default
origin	BaseLogStoreOrigin	The origin used to send the log record.	required
record	LogRecord	The Python logging.LogRecord to process.	required
metadata	Optional[Dict[str, Any]]	Additional metadata to attach to the log entry.	None

Source code in src/zenml/log_stores/base_log_store.py

@abstractmethod
def emit(
    self,
    origin: BaseLogStoreOrigin,
    record: logging.LogRecord,
    metadata: Optional[Dict[str, Any]] = None,
) -> None:
    """Process a log record from the logging system.

    Args:
        origin: The origin used to send the log record.
        record: The Python logging.LogRecord to process.
        metadata: Additional metadata to attach to the log entry.
    """

fetch(logs_model: LogsResponse, start: Optional[str] = None, limit: Optional[int] = None, before: Optional[str] = None, after: Optional[str] = None, filter_: Optional[LogsEntriesFilter] = None) -> LogsEntriesResponse abstractmethod

Fetch a page of log entries.

Implementations must use dependencies available on the server. Pages are ordered chronologically regardless of the starting end.

Return cursors only for directions supported by the backend. Each cursor preserves the native token and query parameters needed for continuation. Explicit continuation parameters must match the cursor.

Parameters:

Name	Type	Description	Default
logs_model	LogsResponse	The logs model containing metadata about the logs.	required
start	Optional[str]	Initial end of the stream: oldest or newest. Defaults to the log store's preferred end.	None
limit	Optional[int]	Maximum number of log entries to return. Defaults to the log store's default_query_size.	None
before	Optional[str]	Cursor towards older entries. Mutually exclusive with after.	None
after	Optional[str]	Cursor towards newer entries. Mutually exclusive with before.	None
filter_	Optional[LogsEntriesFilter]	Filters to apply while retrieving the entries.	None

Returns:

Type	Description
LogsEntriesResponse	Log entries ordered from oldest to newest and supported cursors.

Raises:

Type	Description
ValueError	If parameters are unsupported or conflict with a cursor.

Source code in src/zenml/log_stores/base_log_store.py

@abstractmethod
def fetch(
    self,
    logs_model: LogsResponse,
    start: Optional[str] = None,
    limit: Optional[int] = None,
    before: Optional[str] = None,
    after: Optional[str] = None,
    filter_: Optional[LogsEntriesFilter] = None,
) -> LogsEntriesResponse:
    """Fetch a page of log entries.

    Implementations must use dependencies available on the server. Pages
    are ordered chronologically regardless of the starting end.

    Return cursors only for directions supported by the backend. Each
    cursor preserves the native token and query parameters needed for
    continuation. Explicit continuation parameters must match the cursor.

    Args:
        logs_model: The logs model containing metadata about the logs.
        start: Initial end of the stream: `oldest` or `newest`.
            Defaults to the log store's preferred end.
        limit: Maximum number of log entries to return. Defaults to the
            log store's `default_query_size`.
        before: Cursor towards older entries. Mutually exclusive with after.
        after: Cursor towards newer entries. Mutually exclusive with before.
        filter_: Filters to apply while retrieving the entries.

    Returns:
        Log entries ordered from oldest to newest and supported cursors.

    Raises:
        ValueError: If parameters are unsupported or conflict with a cursor.
    """

flush(blocking: bool = True) -> None abstractmethod

Flush the log store.

This method is called to ensure that all logs are flushed to the backend.

Parameters:

Name	Type	Description	Default
blocking	bool	Whether to block until the flush is complete.	True

Source code in src/zenml/log_stores/base_log_store.py

@abstractmethod
def flush(self, blocking: bool = True) -> None:
    """Flush the log store.

    This method is called to ensure that all logs are flushed to the backend.

    Args:
        blocking: Whether to block until the flush is complete.
    """

register_origin(name: str, log_model: LogsResponse, metadata: Dict[str, Any]) -> BaseLogStoreOrigin

Register an origin for the log store.

Parameters:

Name	Type	Description	Default
name	str	The name of the origin.	required
log_model	LogsResponse	The log model associated with the origin.	required
metadata	Dict[str, Any]	Additional metadata to attach to the log entry.	required

Returns:

Type	Description
BaseLogStoreOrigin	The origin.

Source code in src/zenml/log_stores/base_log_store.py

def register_origin(
    self, name: str, log_model: LogsResponse, metadata: Dict[str, Any]
) -> BaseLogStoreOrigin:
    """Register an origin for the log store.

    Args:
        name: The name of the origin.
        log_model: The log model associated with the origin.
        metadata: Additional metadata to attach to the log entry.

    Returns:
        The origin.
    """
    with self._lock:
        origin = self.origin_class(name, self, log_model, metadata)
        self._origins[name] = origin
        return origin

resolve_limit(limit: Optional[int]) -> int

Resolve the page size and apply the global limit.

Parameters:

Name	Type	Description	Default
limit	Optional[int]	The limit requested by the caller, if any.	required

Returns:

Type	Description
int	The page size, capped at LOGS_MAX_ENTRIES_PER_REQUEST.

Raises:

Type	Description
ValueError	If the requested limit is not positive.

Source code in src/zenml/log_stores/base_log_store.py

def resolve_limit(self, limit: Optional[int]) -> int:
    """Resolve the page size and apply the global limit.

    Args:
        limit: The limit requested by the caller, if any.

    Returns:
        The page size, capped at `LOGS_MAX_ENTRIES_PER_REQUEST`.

    Raises:
        ValueError: If the requested limit is not positive.
    """
    if limit is None:
        limit = self.default_query_size
    elif limit <= 0:
        raise ValueError("`limit` must be a positive integer.")

    return min(limit, LOGS_MAX_ENTRIES_PER_REQUEST)

BaseLogStoreConfig(warn_about_plain_text_secrets: bool = False, **kwargs: Any)

Bases: StackComponentConfig

Base configuration for all log stores.

Source code in src/zenml/stack/stack_component.py

def __init__(
    self, warn_about_plain_text_secrets: bool = False, **kwargs: Any
) -> None:
    """Ensures that secret references don't clash with pydantic validation.

    StackComponents allow the specification of all their string attributes
    using secret references of the form `{{secret_name.key}}`. This however
    is only possible when the stack component does not perform any explicit
    validation of this attribute using pydantic validators. If this were
    the case, the validation would run on the secret reference and would
    fail or in the worst case, modify the secret reference and lead to
    unexpected behavior. This method ensures that no attributes that require
    custom pydantic validation are set as secret references.

    Args:
        warn_about_plain_text_secrets: If true, then warns about using
            plain-text secrets.
        **kwargs: Arguments to initialize this stack component.

    Raises:
        ValueError: If an attribute that requires custom pydantic validation
            is passed as a secret reference, or if the `name` attribute
            was passed as a secret reference.
    """
    for key, value in kwargs.items():
        try:
            field = self.__class__.model_fields[key]
        except KeyError:
            # Value for a private attribute or non-existing field, this
            # will fail during the upcoming pydantic validation
            continue

        if value is None:
            continue

        if not secret_utils.is_secret_reference(value):
            if (
                secret_utils.is_secret_field(field)
                and warn_about_plain_text_secrets
            ):
                logger.warning(
                    "You specified a plain-text value for the sensitive "
                    f"attribute `{key}` for a `{self.__class__.__name__}` "
                    "stack component. This is currently only a warning, "
                    "but future versions of ZenML will require you to pass "
                    "in sensitive information as secrets. Check out the "
                    "documentation on how to configure your stack "
                    "components with secrets here: "
                    "https://docs.zenml.io/deploying-zenml/deploying-zenml/secret-management"
                )
            continue

        if pydantic_utils.has_validators(
            pydantic_class=self.__class__, field_name=key
        ):
            raise ValueError(
                f"Passing the stack component attribute `{key}` as a "
                "secret reference is not allowed as additional validation "
                "is required for this attribute."
            )

    super().__init__(**kwargs)

BaseLogStoreFlavor

Bases: Flavor

Base class for all ZenML log store flavors.

Attributes

config_class: Type[BaseLogStoreConfig] property

Config class for the base log store flavor.

Returns:

Type	Description
Type[BaseLogStoreConfig]	The config class.

implementation_class: Type[BaseLogStore] abstractmethod property

Implementation class for the base log store flavor.

Returns:

Type	Description
Type[BaseLogStore]	The implementation class.

type: StackComponentType property

Type of the flavor.

Returns:

Type	Description
StackComponentType	The type of the flavor.

DatadogLogStore(*args: Any, **kwargs: Any)

Bases: OtelLogStore

Log store that exports logs to Datadog.

This implementation extends OtelLogStore and configures it to send logs to Datadog's HTTP intake API.

Source code in src/zenml/log_stores/otel/otel_log_store.py

def __init__(self, *args: Any, **kwargs: Any) -> None:
    """Initialize the OpenTelemetry log store.

    Args:
        *args: Positional arguments for the base class.
        **kwargs: Keyword arguments for the base class.
    """
    super().__init__(*args, **kwargs)

    self._resource: Optional["Resource"] = None
    self._exporter: Optional["LogRecordExporter"] = None
    self._provider: Optional["LoggerProvider"] = None
    self._processor: Optional["OtelBatchLogRecordProcessor"] = None
    self._handler: Optional["LoggingHandler"] = None

Attributes

config: DatadogLogStoreConfig property

Returns the configuration of the Datadog log store.

Returns:

Type	Description
DatadogLogStoreConfig	The configuration.

default_query_size: int property

The maximum page size supported by Datadog's search API.

Returns:

Type	Description
int	The default number of entries per page.

Methods:

cleanup() -> None

Cleanup the Datadog log store.

This method is called when the log store is no longer needed.

Source code in src/zenml/log_stores/datadog/datadog_log_store.py

def cleanup(self) -> None:
    """Cleanup the Datadog log store.

    This method is called when the log store is no longer needed.
    """
    if self._datadog_exporter:
        self._datadog_exporter.shutdown()
        self._datadog_exporter = None

fetch(logs_model: LogsResponse, start: Optional[str] = None, limit: Optional[int] = None, before: Optional[str] = None, after: Optional[str] = None, filter_: Optional[LogsEntriesFilter] = None) -> LogsEntriesResponse

Fetch a page of log entries from the Datadog Logs API.

Cursors retain the filters, time bounds, direction, and page size.

Parameters:

Name	Type	Description	Default
logs_model	LogsResponse	The logs model containing run and step metadata.	required
start	Optional[str]	Initial end of the stream. Defaults to newest.	None
limit	Optional[int]	Maximum number of log entries to return.	None
before	Optional[str]	Cursor towards older entries, from a previous page.	None
after	Optional[str]	Cursor towards newer entries, from a previous page.	None
filter_	Optional[LogsEntriesFilter]	Filters for a new read. Explicit continuation filters must match the cursor.	None

Returns:

Type	Description
LogsEntriesResponse	A page of log entries, oldest first.

Raises:

Type	Description
ValueError	If the logs model does not belong to this log store, or the pagination parameters are invalid.
LogStoreError	If Datadog returns an error or unusable response.

Source code in src/zenml/log_stores/datadog/datadog_log_store.py

def fetch(
    self,
    logs_model: "LogsResponse",
    start: Optional[str] = None,
    limit: Optional[int] = None,
    before: Optional[str] = None,
    after: Optional[str] = None,
    filter_: Optional[LogsEntriesFilter] = None,
) -> LogsEntriesResponse:
    """Fetch a page of log entries from the Datadog Logs API.

    Cursors retain the filters, time bounds, direction, and page size.

    Args:
        logs_model: The logs model containing run and step metadata.
        start: Initial end of the stream. Defaults to `newest`.
        limit: Maximum number of log entries to return.
        before: Cursor towards older entries, from a previous page.
        after: Cursor towards newer entries, from a previous page.
        filter_: Filters for a new read. Explicit continuation filters
            must match the cursor.

    Returns:
        A page of log entries, oldest first.

    Raises:
        ValueError: If the logs model does not belong to this log store,
            or the pagination parameters are invalid.
        LogStoreError: If Datadog returns an error or unusable response.
    """  # noqa: DOC503
    if logs_model.log_store_id != self.id:
        raise ValueError(
            "logs_model.log_store_id does not match the id of the log "
            "store. These entries were collected by another log store, "
            "which is the one that can read them back."
        )

    if before is not None and after is not None:
        raise ValueError("Pass only one of `before` and `after`.")
    if start not in (None, "oldest", "newest"):
        raise ValueError("`start` must be `oldest` or `newest`.")

    filter_ = filter_ or LogsEntriesFilter()
    page_cursor = before if before is not None else after
    cursor = None
    direction: Literal["oldest", "newest"]
    if page_cursor is not None:
        cursor = _DatadogCursor.decode(page_cursor)
        direction = "newest" if before is not None else "oldest"
        if (
            cursor.log_store_id != self.id
            or cursor.logs_id != logs_model.id
            or cursor.direction != direction
        ):
            raise ValueError(
                "Invalid pagination cursor for this log store, stream, "
                "or direction."
            )
        if start is not None and start != cursor.direction:
            raise ValueError(
                "`start` conflicts with the pagination cursor."
            )
        if (
            limit is not None
            and min(self.resolve_limit(limit), DATADOG_MAX_PAGE_SIZE)
            != cursor.limit
        ):
            raise ValueError(
                "`limit` conflicts with the pagination cursor."
            )
        saved_filters = cursor.filters.model_dump()
        for name, value in filter_.model_dump(exclude_none=True).items():
            if value != saved_filters[name]:
                raise ValueError(
                    f"`{name}` conflicts with the pagination cursor."
                )
        filters = cursor.filters
        limit = self.resolve_limit(cursor.limit)
    else:
        direction = "oldest" if start == "oldest" else "newest"
        limit = min(self.resolve_limit(limit), DATADOG_MAX_PAGE_SIZE)
        filters = _DatadogFilters(
            search=filter_.search,
            level=filter_.level,
            since=filter_.since or to_utc_timezone(logs_model.created),
            until=filter_.until or utc_now(tz_aware=True),
        )
    descending = direction == "newest"

    body: Dict[str, Any] = {
        "filter": {
            "query": self._build_query(logs_model, filters),
            "from": filters.since.isoformat(),
            "to": filters.until.isoformat(),
        },
        "page": {"limit": limit},
        "sort": "-timestamp" if descending else "timestamp",
    }
    if cursor is not None:
        body["page"]["cursor"] = cursor.token

    events, native_cursor = self._search(body)
    entries = [
        entry
        for entry in (self._parse_log_entry(event) for event in events)
        if entry is not None
    ]
    if descending:
        entries.reverse()

    encoded = (
        _DatadogCursor(
            version=1,
            token=native_cursor,
            log_store_id=self.id,
            logs_id=logs_model.id,
            filters=filters,
            direction=direction,
            limit=limit,
        ).encode()
        if native_cursor
        else None
    )
    return LogsEntriesResponse(
        items=entries,
        before=encoded if descending else None,
        after=encoded if not descending else None,
    )

get_exporter() -> DatadogLogExporter

Get the Datadog log exporter.

Returns:

Type	Description
DatadogLogExporter	DatadogExporter with the proper configuration.

Source code in src/zenml/log_stores/datadog/datadog_log_store.py

def get_exporter(self) -> DatadogLogExporter:
    """Get the Datadog log exporter.

    Returns:
        DatadogExporter with the proper configuration.
    """
    if not self._datadog_exporter:
        self._datadog_exporter = DatadogLogExporter(
            endpoint=self.config.endpoint,
            headers=self._get_headers(),
            certificate_file=self.config.certificate_file,
            client_key_file=self.config.client_key_file,
            client_certificate_file=self.config.client_certificate_file,
            compression=self.config.compression,
        )
    return self._datadog_exporter

DatadogLogStoreConfig(warn_about_plain_text_secrets: bool = False, **kwargs: Any)

Bases: OtelLogStoreConfig

Configuration for Datadog log store.

Attributes:

Name	Type	Description
api_key	PlainSerializedSecretStr	Datadog API key for log ingestion.
application_key	PlainSerializedSecretStr	Datadog application key for log extraction.
site	str	Datadog site (e.g., "datadoghq.com", "datadoghq.eu").
max_export_batch_size	int	Maximum batch size for exports (Datadog limit: 1000).

Source code in src/zenml/stack/stack_component.py

def __init__(
    self, warn_about_plain_text_secrets: bool = False, **kwargs: Any
) -> None:
    """Ensures that secret references don't clash with pydantic validation.

    StackComponents allow the specification of all their string attributes
    using secret references of the form `{{secret_name.key}}`. This however
    is only possible when the stack component does not perform any explicit
    validation of this attribute using pydantic validators. If this were
    the case, the validation would run on the secret reference and would
    fail or in the worst case, modify the secret reference and lead to
    unexpected behavior. This method ensures that no attributes that require
    custom pydantic validation are set as secret references.

    Args:
        warn_about_plain_text_secrets: If true, then warns about using
            plain-text secrets.
        **kwargs: Arguments to initialize this stack component.

    Raises:
        ValueError: If an attribute that requires custom pydantic validation
            is passed as a secret reference, or if the `name` attribute
            was passed as a secret reference.
    """
    for key, value in kwargs.items():
        try:
            field = self.__class__.model_fields[key]
        except KeyError:
            # Value for a private attribute or non-existing field, this
            # will fail during the upcoming pydantic validation
            continue

        if value is None:
            continue

        if not secret_utils.is_secret_reference(value):
            if (
                secret_utils.is_secret_field(field)
                and warn_about_plain_text_secrets
            ):
                logger.warning(
                    "You specified a plain-text value for the sensitive "
                    f"attribute `{key}` for a `{self.__class__.__name__}` "
                    "stack component. This is currently only a warning, "
                    "but future versions of ZenML will require you to pass "
                    "in sensitive information as secrets. Check out the "
                    "documentation on how to configure your stack "
                    "components with secrets here: "
                    "https://docs.zenml.io/deploying-zenml/deploying-zenml/secret-management"
                )
            continue

        if pydantic_utils.has_validators(
            pydantic_class=self.__class__, field_name=key
        ):
            raise ValueError(
                f"Passing the stack component attribute `{key}` as a "
                "secret reference is not allowed as additional validation "
                "is required for this attribute."
            )

    super().__init__(**kwargs)

Methods:

set_default_endpoint(data: Dict[str, Any]) -> Dict[str, Any] classmethod

Set the endpoint based on site if not provided.

Parameters:

Name	Type	Description	Default
data	Dict[str, Any]	The input data dictionary.	required

Returns:

Type	Description
Dict[str, Any]	The data dictionary with the endpoint set if not provided.

Source code in src/zenml/log_stores/datadog/datadog_flavor.py

@model_validator(mode="before")
@classmethod
def set_default_endpoint(cls, data: Dict[str, Any]) -> Dict[str, Any]:
    """Set the endpoint based on site if not provided.

    Args:
        data: The input data dictionary.

    Returns:
        The data dictionary with the endpoint set if not provided.
    """
    if isinstance(data, dict) and not data.get("endpoint"):
        site = data.get("site", "datadoghq.com")
        data["endpoint"] = f"https://http-intake.logs.{site}/api/v2/logs"
    return data

validate_max_export_batch_size(v: int) -> int classmethod

Validate that max_export_batch_size doesn't exceed Datadog's limit.

Parameters:

Name	Type	Description	Default
v	int	The value to validate.	required

Returns:

Type	Description
int	The validated value.

Raises:

Type	Description
ValueError	If the value exceeds Datadog's limit.

Source code in src/zenml/log_stores/datadog/datadog_flavor.py

@field_validator("max_export_batch_size")
@classmethod
def validate_max_export_batch_size(cls, v: int) -> int:
    """Validate that max_export_batch_size doesn't exceed Datadog's limit.

    Args:
        v: The value to validate.

    Returns:
        The validated value.

    Raises:
        ValueError: If the value exceeds Datadog's limit.
    """
    if v > DATADOG_MAX_BATCH_SIZE:
        raise ValueError(
            f"max_export_batch_size cannot exceed {DATADOG_MAX_BATCH_SIZE} "
            f"(Datadog API limit). Got: {v}"
        )
    return v

DatadogLogStoreFlavor

Bases: Flavor

Datadog log store flavor.

Attributes

config_class: Type[BaseLogStoreConfig] property

Returns DatadogLogStoreConfig config class.

Returns:

Type	Description
Type[BaseLogStoreConfig]	The config class.

docs_url: str property

URL to the flavor documentation.

Returns:

Type	Description
str	The URL to the flavor documentation.

implementation_class: Type[BaseLogStore] property

Implementation class for this flavor.

Returns:

Type	Description
Type[BaseLogStore]	The implementation class.

logo_url: str property

URL to the flavor logo.

Returns:

Type	Description
str	The URL to the flavor logo.

name: str property

Name of the flavor.

Returns:

Type	Description
str	The name of the flavor.

sdk_docs_url: str property

URL to the SDK docs for this flavor.

Returns:

Type	Description
str	The URL to the SDK docs for this flavor.

type: StackComponentType property

Stack component type.

Returns:

Type	Description
StackComponentType	The stack component type.

OtelLogStore(*args: Any, **kwargs: Any)

Bases: BaseLogStore

Log store that exports logs using OpenTelemetry.

Subclasses should implement get_exporter() to provide the specific log exporter for their backend.

Initialize the OpenTelemetry log store.

Parameters:

Name	Type	Description	Default
*args	Any	Positional arguments for the base class.	()
**kwargs	Any	Keyword arguments for the base class.	{}

Source code in src/zenml/log_stores/otel/otel_log_store.py

def __init__(self, *args: Any, **kwargs: Any) -> None:
    """Initialize the OpenTelemetry log store.

    Args:
        *args: Positional arguments for the base class.
        **kwargs: Keyword arguments for the base class.
    """
    super().__init__(*args, **kwargs)

    self._resource: Optional["Resource"] = None
    self._exporter: Optional["LogRecordExporter"] = None
    self._provider: Optional["LoggerProvider"] = None
    self._processor: Optional["OtelBatchLogRecordProcessor"] = None
    self._handler: Optional["LoggingHandler"] = None

Attributes

config: OtelLogStoreConfig property

Returns the configuration of the OTel log store.

Returns:

Type	Description
OtelLogStoreConfig	The configuration.

origin_class: Type[OtelLogStoreOrigin] property

Class of the origin.

Returns:

Type	Description
Type[OtelLogStoreOrigin]	The class of the origin.

provider: LoggerProvider property

Returns the OpenTelemetry logger provider.

Returns:

Type	Description
LoggerProvider	The logger provider.

Raises:

Type	Description
RuntimeError	If the OpenTelemetry log store is not initialized.

Methods:

deactivate() -> None

Deactivate log collection and shut down the processor.

Flushes any pending logs and shuts down the processor's background thread.

Source code in src/zenml/log_stores/otel/otel_log_store.py

def deactivate(self) -> None:
    """Deactivate log collection and shut down the processor.

    Flushes any pending logs and shuts down the processor's background thread.
    """
    with self._lock:
        if self._processor:
            try:
                # Force flush any pending logs
                self._processor.flush(blocking=True)
                logger.debug("Flushed pending logs")
            except Exception as e:
                logger.warning(f"Error flushing logs: {e}")

            try:
                self._processor.shutdown()  # type: ignore[no-untyped-call]
                logger.debug(
                    "Shut down log processor and background thread"
                )
            except Exception as e:
                logger.warning(f"Error shutting down processor: {e}")
            else:
                self._processor = None
                self._handler = None
                self._provider = None
                self._resource = None
                self._exporter = None

    logger.debug("OtelLogStore deactivated")

emit(origin: BaseLogStoreOrigin, record: logging.LogRecord, metadata: Optional[Dict[str, Any]] = None) -> None

Process a log record by sending to OpenTelemetry.

Parameters:

Name	Type	Description	Default
origin	BaseLogStoreOrigin	The origin used to send the log record.	required
record	LogRecord	The log record to process.	required
metadata	Optional[Dict[str, Any]]	Additional metadata to attach to the log entry.	None

Raises:

Type	Description
RuntimeError	If the OpenTelemetry provider is not initialized.

Source code in src/zenml/log_stores/otel/otel_log_store.py

def emit(
    self,
    origin: BaseLogStoreOrigin,
    record: logging.LogRecord,
    metadata: Optional[Dict[str, Any]] = None,
) -> None:
    """Process a log record by sending to OpenTelemetry.

    Args:
        origin: The origin used to send the log record.
        record: The log record to process.
        metadata: Additional metadata to attach to the log entry.

    Raises:
        RuntimeError: If the OpenTelemetry provider is not initialized.
    """
    assert isinstance(origin, OtelLogStoreOrigin)
    with self._lock:
        if not self._provider:
            self._activate()

        if self._handler is None:
            raise RuntimeError("OpenTelemetry provider is not initialized")

        otel_log_record = self._handler._translate(record)
        merged_attrs = dict(otel_log_record.attributes or {})
        merged_attrs.update(origin.metadata)
        if metadata:
            merged_attrs.update(metadata)
        otel_log_record.attributes = merged_attrs
        origin.logger.emit(otel_log_record)

fetch(logs_model: LogsResponse, start: Optional[str] = None, limit: Optional[int] = None, before: Optional[str] = None, after: Optional[str] = None, filter_: Optional[LogsEntriesFilter] = None) -> LogsEntriesResponse

Fetch log entries from an OpenTelemetry backend.

OTLP does not support log retrieval. Subclasses must implement their backend's query API.

Parameters:

Name	Type	Description	Default
logs_model	LogsResponse	The logs model containing run and step metadata.	required
start	Optional[str]	Initial end of the stream, chosen by the subclass if omitted.	None
limit	Optional[int]	Maximum number of log entries to return.	None
before	Optional[str]	Cursor towards older entries, from a previous page.	None
after	Optional[str]	Cursor towards newer entries, from a previous page.	None
filter_	Optional[LogsEntriesFilter]	Filters to apply while retrieving the entries.	None

Returns:

Type	Description
LogsEntriesResponse	Never returns.

Raises:

Type	Description
NotImplementedError	Log retrieval is not supported by OTLP.

Source code in src/zenml/log_stores/otel/otel_log_store.py

def fetch(
    self,
    logs_model: "LogsResponse",
    start: Optional[str] = None,
    limit: Optional[int] = None,
    before: Optional[str] = None,
    after: Optional[str] = None,
    filter_: Optional["LogsEntriesFilter"] = None,
) -> "LogsEntriesResponse":
    """Fetch log entries from an OpenTelemetry backend.

    OTLP does not support log retrieval. Subclasses must implement their
    backend's query API.

    Args:
        logs_model: The logs model containing run and step metadata.
        start: Initial end of the stream, chosen by the subclass if omitted.
        limit: Maximum number of log entries to return.
        before: Cursor towards older entries, from a previous page.
        after: Cursor towards newer entries, from a previous page.
        filter_: Filters to apply while retrieving the entries.

    Returns:
        Never returns.

    Raises:
        NotImplementedError: Log retrieval is not supported by OTLP.
    """
    raise NotImplementedError(
        "Log fetching is not supported by the OTEL log store."
    )

flush(blocking: bool = True) -> None

Flush the log store.

Parameters:

Name	Type	Description	Default
blocking	bool	Whether to block until the flush is complete.	True

This method is called to ensure that all logs are flushed to the backend.

Source code in src/zenml/log_stores/otel/otel_log_store.py

def flush(self, blocking: bool = True) -> None:
    """Flush the log store.

    Args:
        blocking: Whether to block until the flush is complete.

    This method is called to ensure that all logs are flushed to the backend.
    """
    with self._lock:
        if self._processor:
            self._processor.flush(blocking=blocking)

get_exporter() -> LogRecordExporter

Get the Datadog log exporter.

Returns:

Type	Description
LogRecordExporter	OTLPLogExporter configured with API key and site.

Source code in src/zenml/log_stores/otel/otel_log_store.py

def get_exporter(self) -> "LogRecordExporter":
    """Get the Datadog log exporter.

    Returns:
        OTLPLogExporter configured with API key and site.
    """
    if not self._exporter:
        self._exporter = OTLPLogExporter(
            endpoint=self.config.endpoint,
            headers=self.config.headers,
            certificate_file=self.config.certificate_file,
            client_key_file=self.config.client_key_file,
            client_certificate_file=self.config.client_certificate_file,
            compression=self.config.compression,
        )

    return self._exporter

register_origin(name: str, log_model: LogsResponse, metadata: Dict[str, Any]) -> BaseLogStoreOrigin

Register an origin for the log store.

Parameters:

Name	Type	Description	Default
name	str	The name of the origin.	required
log_model	LogsResponse	The log model associated with the origin.	required
metadata	Dict[str, Any]	Additional metadata to attach to the log entry.	required

Returns:

Type	Description
BaseLogStoreOrigin	The origin.

Source code in src/zenml/log_stores/otel/otel_log_store.py

def register_origin(
    self, name: str, log_model: LogsResponse, metadata: Dict[str, Any]
) -> BaseLogStoreOrigin:
    """Register an origin for the log store.

    Args:
        name: The name of the origin.
        log_model: The log model associated with the origin.
        metadata: Additional metadata to attach to the log entry.

    Returns:
        The origin.
    """
    with self._lock:
        if not self._provider:
            self._activate()

    return super().register_origin(name, log_model, metadata)

OtelLogStoreConfig(warn_about_plain_text_secrets: bool = False, **kwargs: Any)

Bases: BaseLogStoreConfig

Configuration for OpenTelemetry log store.

Attributes:

Name	Type	Description
service_name	str	Name of the service (defaults to "zenml").
service_version	str	Version of the service (defaults to the ZenML version).
max_queue_size	int	Maximum queue size for batch processor.
schedule_delay_millis	int	Delay between batch exports in milliseconds.
max_export_batch_size	int	Maximum batch size for exports.
endpoint	str	The endpoint to export logs to.
headers	Optional[Dict[str, str]]	The headers to use for the export.
certificate_file	Optional[str]	The certificate file to use for the export.
client_key_file	Optional[str]	The client key file to use for the export.
client_certificate_file	Optional[str]	The client certificate file to use for the export.
compression	Compression	The compression to use for the export.

Source code in src/zenml/stack/stack_component.py

def __init__(
    self, warn_about_plain_text_secrets: bool = False, **kwargs: Any
) -> None:
    """Ensures that secret references don't clash with pydantic validation.

    StackComponents allow the specification of all their string attributes
    using secret references of the form `{{secret_name.key}}`. This however
    is only possible when the stack component does not perform any explicit
    validation of this attribute using pydantic validators. If this were
    the case, the validation would run on the secret reference and would
    fail or in the worst case, modify the secret reference and lead to
    unexpected behavior. This method ensures that no attributes that require
    custom pydantic validation are set as secret references.

    Args:
        warn_about_plain_text_secrets: If true, then warns about using
            plain-text secrets.
        **kwargs: Arguments to initialize this stack component.

    Raises:
        ValueError: If an attribute that requires custom pydantic validation
            is passed as a secret reference, or if the `name` attribute
            was passed as a secret reference.
    """
    for key, value in kwargs.items():
        try:
            field = self.__class__.model_fields[key]
        except KeyError:
            # Value for a private attribute or non-existing field, this
            # will fail during the upcoming pydantic validation
            continue

        if value is None:
            continue

        if not secret_utils.is_secret_reference(value):
            if (
                secret_utils.is_secret_field(field)
                and warn_about_plain_text_secrets
            ):
                logger.warning(
                    "You specified a plain-text value for the sensitive "
                    f"attribute `{key}` for a `{self.__class__.__name__}` "
                    "stack component. This is currently only a warning, "
                    "but future versions of ZenML will require you to pass "
                    "in sensitive information as secrets. Check out the "
                    "documentation on how to configure your stack "
                    "components with secrets here: "
                    "https://docs.zenml.io/deploying-zenml/deploying-zenml/secret-management"
                )
            continue

        if pydantic_utils.has_validators(
            pydantic_class=self.__class__, field_name=key
        ):
            raise ValueError(
                f"Passing the stack component attribute `{key}` as a "
                "secret reference is not allowed as additional validation "
                "is required for this attribute."
            )

    super().__init__(**kwargs)

OtelLogStoreFlavor

Bases: Flavor

OpenTelemetry log store flavor.

Attributes

config_class: Type[BaseLogStoreConfig] property

Returns DatadogLogStoreConfig config class.

Returns:

Type	Description
Type[BaseLogStoreConfig]	The config class.

docs_url: str property

URL to the flavor documentation.

Returns:

Type	Description
str	The URL to the flavor documentation.

implementation_class: Type[BaseLogStore] property

Implementation class for this flavor.

Returns:

Type	Description
Type[BaseLogStore]	The implementation class.

logo_url: str property

URL to the flavor logo.

Returns:

Type	Description
str	The URL to the flavor logo.

name: str property

Name of the flavor.

Returns:

Type	Description
str	The name of the flavor.

sdk_docs_url: str property

URL to the SDK docs for this flavor.

Returns:

Type	Description
str	The URL to the SDK docs for this flavor.

type: StackComponentType property

Stack component type.

Returns:

Type	Description
StackComponentType	The stack component type.

Modules

artifact

Artifact log store implementation.

Modules

artifact_log_exporter

OpenTelemetry exporter that writes logs to ZenML artifact store.

Classes

ArtifactLogExporter(artifact_store: BaseArtifactStore)

Bases: LogRecordExporter

OpenTelemetry exporter that writes logs to ZenML artifact store.

Initialize the exporter.

Parameters:

Name	Type	Description	Default
artifact_store	BaseArtifactStore	The artifact store to write logs to.	required

Source code in src/zenml/log_stores/artifact/artifact_log_exporter.py

def __init__(self, artifact_store: "BaseArtifactStore") -> None:
    """Initialize the exporter.

    Args:
        artifact_store: The artifact store to write logs to.
    """
    self.artifact_store = artifact_store

Methods:

export(batch: Sequence[ReadableLogRecord]) -> LogRecordExportResult

Export a batch of logs to the artifact store.

Parameters:

Name	Type	Description	Default
batch	Sequence[ReadableLogRecord]	Readable log records to export (can be from multiple contexts).	required

Returns:

Type	Description
LogRecordExportResult	LogRecordExportResult indicating success or failure.

Source code in src/zenml/log_stores/artifact/artifact_log_exporter.py

def export(
    self, batch: Sequence[ReadableLogRecord]
) -> LogRecordExportResult:
    """Export a batch of logs to the artifact store.

    Args:
        batch: Readable log records to export (can be from multiple contexts).

    Returns:
        LogRecordExportResult indicating success or failure.
    """
    if not batch:
        return LogRecordExportResult.SUCCESS

    try:
        logs_by_uri: Dict[str, List[str]] = defaultdict(list)
        finalized_log_streams: List[str] = []

        for readable in batch:
            attrs = readable.log_record.attributes
            if not attrs:
                continue

            log_uri = attrs.get("zenml.log.uri")
            if not log_uri or not isinstance(log_uri, str):
                continue

            if readable.log_record.body is END_OF_STREAM_MESSAGE:
                finalized_log_streams.append(log_uri)
                continue

            entries = self._otel_record_to_log_entries(readable)
            for entry in entries:
                json_line = entry.model_dump_json(exclude_none=True)
                logs_by_uri[log_uri].append(json_line)

        for log_uri, log_lines in logs_by_uri.items():
            if log_lines:
                self._write(log_lines, log_uri)

        for log_uri in finalized_log_streams:
            self._finalize(log_uri)

        return LogRecordExportResult.SUCCESS

    except Exception:
        logger.exception("Failed to export logs to artifact store")
        return LogRecordExportResult.FAILURE

shutdown() -> None

Shutdown the exporter.

Source code in src/zenml/log_stores/artifact/artifact_log_exporter.py

def shutdown(self) -> None:
    """Shutdown the exporter."""
    self.artifact_store.cleanup()

Functions:

artifact_log_store

Artifact log store implementation.

Classes

ArtifactLogStore(artifact_store: BaseArtifactStore, *args: Any, **kwargs: Any)

Bases: OtelLogStore

Log store that saves logs to the artifact store.

This implementation extends OtelLogStore and uses the ArtifactLogExporter to write logs to the artifact store. Inherits all OTEL infrastructure including shared BatchLogRecordProcessor and routing.

Initialize the artifact log store.

Parameters:

Name	Type	Description	Default
artifact_store	BaseArtifactStore	The artifact store to use for logging.	required
*args	Any	Positional arguments for the base class.	()
**kwargs	Any	Keyword arguments for the base class.	{}

Source code in src/zenml/log_stores/artifact/artifact_log_store.py

def __init__(
    self, artifact_store: "BaseArtifactStore", *args: Any, **kwargs: Any
) -> None:
    """Initialize the artifact log store.

    Args:
        artifact_store: The artifact store to use for logging.
        *args: Positional arguments for the base class.
        **kwargs: Keyword arguments for the base class.
    """
    super().__init__(*args, **kwargs)
    self._artifact_store = artifact_store

Attributes

config: ArtifactLogStoreConfig property

Returns the configuration of the artifact log store.

Returns:

Type	Description
ArtifactLogStoreConfig	The configuration.

origin_class: Type[ArtifactLogStoreOrigin] property

Class of the origin.

Returns:

Type	Description
Type[ArtifactLogStoreOrigin]	The class of the origin.

Methods:

cleanup() -> None

Cleanup the artifact log store.

This method is called to ensure that the artifact log store is cleaned up.

Source code in src/zenml/log_stores/artifact/artifact_log_store.py

def cleanup(self) -> None:
    """Cleanup the artifact log store.

    This method is called to ensure that the artifact log store is cleaned up.
    """
    self._artifact_store.cleanup()

fetch(logs_model: LogsResponse, start: Optional[str] = None, limit: Optional[int] = None, before: Optional[str] = None, after: Optional[str] = None, filter_: Optional[LogsEntriesFilter] = None) -> LogsEntriesResponse

Fetch a batch of log entries from the artifact store.

Parameters:

Name	Type	Description	Default
logs_model	LogsResponse	The logs model containing uri and artifact_store_id.	required
start	Optional[str]	Initial position. Only None and oldest are supported.	None
limit	Optional[int]	Maximum number of log entries to return.	None
before	Optional[str]	Not supported.	None
after	Optional[str]	Not supported.	None
filter_	Optional[LogsEntriesFilter]	Not supported.	None

Returns:

Type	Description
LogsEntriesResponse	The oldest entries up to the limit, without cursors.

Raises:

Type	Description
ValueError	If the logs model is invalid, the limit is not positive, or unsupported parameters are provided.

Source code in src/zenml/log_stores/artifact/artifact_log_store.py

def fetch(
    self,
    logs_model: "LogsResponse",
    start: Optional[str] = None,
    limit: Optional[int] = None,
    before: Optional[str] = None,
    after: Optional[str] = None,
    filter_: Optional[LogsEntriesFilter] = None,
) -> LogsEntriesResponse:
    """Fetch a batch of log entries from the artifact store.

    Args:
        logs_model: The logs model containing uri and artifact_store_id.
        start: Initial position. Only None and `oldest` are supported.
        limit: Maximum number of log entries to return.
        before: Not supported.
        after: Not supported.
        filter_: Not supported.

    Returns:
        The oldest entries up to the limit, without cursors.

    Raises:
        ValueError: If the logs model is invalid, the limit is not positive,
            or unsupported parameters are provided.
    """
    if not logs_model.uri:
        raise ValueError(
            "logs_model.uri is required for ArtifactLogStore.fetch()"
        )

    if not logs_model.artifact_store_id:
        raise ValueError(
            "logs_model.artifact_store_id is required "
            "for ArtifactLogStore.fetch()"
        )

    if logs_model.artifact_store_id != self._artifact_store.id:
        raise ValueError(
            "logs_model.artifact_store_id does not match the artifact store "
            "id of the log store."
        )

    unallowed = (
        start not in (None, "oldest")
        or before is not None
        or after is not None
    )
    if filter_ is not None:
        unallowed = unallowed or any(
            (
                filter_.search,
                filter_.level,
                filter_.since,
                filter_.until,
            )
        )
    if unallowed:
        raise ValueError(
            "The artifact log store only reads a log file from its start "
            "and serves what fits in one response. Drop start, before, "
            "after, search, level, since, and until, and narrow the result down where it is "
            "displayed."
        )

    return LogsEntriesResponse(
        items=fetch_log_records(
            artifact_store=self._artifact_store,
            logs_uri=logs_model.uri,
            limit=self.resolve_limit(limit),
        )
    )

from_artifact_store(artifact_store: BaseArtifactStore) -> ArtifactLogStore classmethod

Creates an artifact log store from an artifact store.

Parameters:

Name	Type	Description	Default
artifact_store	BaseArtifactStore	The artifact store to create the log store from.	required

Returns:

Type	Description
ArtifactLogStore	The created artifact log store.

Source code in src/zenml/log_stores/artifact/artifact_log_store.py

@classmethod
def from_artifact_store(
    cls, artifact_store: "BaseArtifactStore"
) -> "ArtifactLogStore":
    """Creates an artifact log store from an artifact store.

    Args:
        artifact_store: The artifact store to create the log store from.

    Returns:
        The created artifact log store.
    """
    return cls(
        artifact_store=artifact_store,
        id=artifact_store.id,
        name="default",
        config=ArtifactLogStoreConfig(endpoint=artifact_store.path),
        flavor="artifact",
        type=StackComponentType.LOG_STORE,
        user=artifact_store.user,
        created=artifact_store.created,
        updated=artifact_store.updated,
    )

get_exporter() -> LogRecordExporter

Get the artifact log exporter for this log store.

Returns:

Type	Description
LogRecordExporter	The ArtifactLogExporter instance.

Source code in src/zenml/log_stores/artifact/artifact_log_store.py

def get_exporter(self) -> "LogRecordExporter":
    """Get the artifact log exporter for this log store.

    Returns:
        The ArtifactLogExporter instance.
    """
    from zenml.log_stores.artifact.artifact_log_exporter import (
        ArtifactLogExporter,
    )

    return ArtifactLogExporter(artifact_store=self._artifact_store)

ArtifactLogStoreConfig(warn_about_plain_text_secrets: bool = False, **kwargs: Any)

Bases: OtelLogStoreConfig

Configuration for the artifact log store.

Source code in src/zenml/stack/stack_component.py

def __init__(
    self, warn_about_plain_text_secrets: bool = False, **kwargs: Any
) -> None:
    """Ensures that secret references don't clash with pydantic validation.

    StackComponents allow the specification of all their string attributes
    using secret references of the form `{{secret_name.key}}`. This however
    is only possible when the stack component does not perform any explicit
    validation of this attribute using pydantic validators. If this were
    the case, the validation would run on the secret reference and would
    fail or in the worst case, modify the secret reference and lead to
    unexpected behavior. This method ensures that no attributes that require
    custom pydantic validation are set as secret references.

    Args:
        warn_about_plain_text_secrets: If true, then warns about using
            plain-text secrets.
        **kwargs: Arguments to initialize this stack component.

    Raises:
        ValueError: If an attribute that requires custom pydantic validation
            is passed as a secret reference, or if the `name` attribute
            was passed as a secret reference.
    """
    for key, value in kwargs.items():
        try:
            field = self.__class__.model_fields[key]
        except KeyError:
            # Value for a private attribute or non-existing field, this
            # will fail during the upcoming pydantic validation
            continue

        if value is None:
            continue

        if not secret_utils.is_secret_reference(value):
            if (
                secret_utils.is_secret_field(field)
                and warn_about_plain_text_secrets
            ):
                logger.warning(
                    "You specified a plain-text value for the sensitive "
                    f"attribute `{key}` for a `{self.__class__.__name__}` "
                    "stack component. This is currently only a warning, "
                    "but future versions of ZenML will require you to pass "
                    "in sensitive information as secrets. Check out the "
                    "documentation on how to configure your stack "
                    "components with secrets here: "
                    "https://docs.zenml.io/deploying-zenml/deploying-zenml/secret-management"
                )
            continue

        if pydantic_utils.has_validators(
            pydantic_class=self.__class__, field_name=key
        ):
            raise ValueError(
                f"Passing the stack component attribute `{key}` as a "
                "secret reference is not allowed as additional validation "
                "is required for this attribute."
            )

    super().__init__(**kwargs)

ArtifactLogStoreOrigin(name: str, log_store: BaseLogStore, log_model: LogsResponse, metadata: Dict[str, Any])

Bases: OtelLogStoreOrigin

Artifact log store origin.

Initialize a log store origin.

Parameters:

Name	Type	Description	Default
name	str	The name of the origin.	required
log_store	BaseLogStore	The log store to emit logs to.	required
log_model	LogsResponse	The log model associated with the origin.	required
metadata	Dict[str, Any]	Additional metadata to attach to all log entries that will be emitted by this origin.	required

Source code in src/zenml/log_stores/artifact/artifact_log_store.py

def __init__(
    self,
    name: str,
    log_store: "BaseLogStore",
    log_model: LogsResponse,
    metadata: Dict[str, Any],
) -> None:
    """Initialize a log store origin.

    Args:
        name: The name of the origin.
        log_store: The log store to emit logs to.
        log_model: The log model associated with the origin.
        metadata: Additional metadata to attach to all log entries that will
            be emitted by this origin.
    """
    super().__init__(name, log_store, log_model, metadata)

    if log_model.uri:
        self.metadata["zenml.log.uri"] = log_model.uri

Methods:

Functions:

fetch_log_records(artifact_store: BaseArtifactStore, logs_uri: str, limit: int) -> List[LogEntry]

Fetches log entries.

Parameters:

Name	Type	Description	Default
artifact_store	BaseArtifactStore	The artifact store.	required
logs_uri	str	The URI of the artifact (file or directory).	required
limit	int	Maximum number of log entries to return.	required

Returns:

Type	Description
List[LogEntry]	List of log entries.

Source code in src/zenml/log_stores/artifact/artifact_log_store.py

def fetch_log_records(
    artifact_store: "BaseArtifactStore",
    logs_uri: str,
    limit: int,
) -> List[LogEntry]:
    """Fetches log entries.

    Args:
        artifact_store: The artifact store.
        logs_uri: The URI of the artifact (file or directory).
        limit: Maximum number of log entries to return.

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

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

        if len(log_entries) >= limit:
            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/log_stores/artifact/artifact_log_store.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.
    """
    line = log_line.strip()
    if not line:
        return None

    if line.startswith("{") and line.endswith("}"):
        try:
            return LogEntry.model_validate_json(line)
        except Exception:
            pass

    old_format = re.search(
        r"^\[(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\s+UTC\]", line
    )

    timestamp = None
    if old_format:
        timestamp = old_format.group(1) + "Z"
        line = line.replace(old_format.group(0), "").strip()

    return LogEntry(
        message=line,
        name=None,
        level=LoggingLevels.INFO,
        timestamp=timestamp,
    )

prepare_logs_uri(artifact_store: BaseArtifactStore, log_id: UUID) -> 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
log_id	UUID	The ID of the logs entity	required

Returns:

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

Source code in src/zenml/log_stores/artifact/artifact_log_store.py

def prepare_logs_uri(
    artifact_store: "BaseArtifactStore",
    log_id: UUID,
) -> 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.
        log_id: The ID of the logs entity

    Returns:
        The URI of the log storage (file or folder).
    """
    logs_base_uri = os.path.join(artifact_store.path, "logs")

    if artifact_store.config.IS_IMMUTABLE_FILESYSTEM:
        logs_uri = os.path.join(logs_base_uri, str(log_id))
    else:
        logs_uri = os.path.join(logs_base_uri, f"{log_id}{LOGS_EXTENSION}")

    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/log_stores/artifact/artifact_log_store.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)

base_log_store

Base class for log stores.

Classes

BaseLogStore(*args: Any, **kwargs: Any)

Bases: StackComponent, ABC

Base class for all ZenML log stores.

A log store is responsible for collecting, storing, and retrieving logs during pipeline and step execution. Different implementations may store logs in different backends (artifact store, OpenTelemetry, Datadog, etc.).

Initialize the log store.

Parameters:

Name	Type	Description	Default
*args	Any	Positional arguments for the base class.	()
**kwargs	Any	Keyword arguments for the base class.	{}

Source code in src/zenml/log_stores/base_log_store.py

def __init__(self, *args: Any, **kwargs: Any) -> None:
    """Initialize the log store.

    Args:
        *args: Positional arguments for the base class.
        **kwargs: Keyword arguments for the base class.
    """
    super().__init__(*args, **kwargs)
    self._origins: Dict[str, BaseLogStoreOrigin] = {}
    self._lock = threading.RLock()

Attributes

config: BaseLogStoreConfig property

Returns the configuration of the log store.

Returns:

Type	Description
BaseLogStoreConfig	The configuration.

default_query_size: int property

Default page size when no limit is specified.

Returns:

Type	Description
int	The default number of entries per page.

origin_class: Type[BaseLogStoreOrigin] property

Class of the origin.

Returns:

Type	Description
Type[BaseLogStoreOrigin]	The class of the origin used with this log store.

Methods:

deregister_origin(origin: BaseLogStoreOrigin, blocking: bool = True) -> None

Deregister an origin previously registered with the log store.

If no other origins are left, the log store will be flushed. The blocking parameter determines whether to block until the flush is complete.

Parameters:

Name	Type	Description	Default
origin	BaseLogStoreOrigin	The origin to deregister.	required
blocking	bool	Whether to block until the deregistration is complete and all logs are flushed if this is the last origin registered.	True

Source code in src/zenml/log_stores/base_log_store.py

def deregister_origin(
    self,
    origin: BaseLogStoreOrigin,
    blocking: bool = True,
) -> None:
    """Deregister an origin previously registered with the log store.

    If no other origins are left, the log store will be flushed. The
    `blocking` parameter determines whether to block until the flush is
    complete.

    Args:
        origin: The origin to deregister.
        blocking: Whether to block until the deregistration is complete
            and all logs are flushed if this is the last origin registered.
    """
    with self._lock:
        if origin.name not in self._origins:
            return
        self._release_origin(origin)
        del self._origins[origin.name]
        if len(self._origins) == 0:
            self.flush(blocking=blocking)

emit(origin: BaseLogStoreOrigin, record: logging.LogRecord, metadata: Optional[Dict[str, Any]] = None) -> None abstractmethod

Process a log record from the logging system.

Parameters:

Name	Type	Description	Default
origin	BaseLogStoreOrigin	The origin used to send the log record.	required
record	LogRecord	The Python logging.LogRecord to process.	required
metadata	Optional[Dict[str, Any]]	Additional metadata to attach to the log entry.	None

Source code in src/zenml/log_stores/base_log_store.py

@abstractmethod
def emit(
    self,
    origin: BaseLogStoreOrigin,
    record: logging.LogRecord,
    metadata: Optional[Dict[str, Any]] = None,
) -> None:
    """Process a log record from the logging system.

    Args:
        origin: The origin used to send the log record.
        record: The Python logging.LogRecord to process.
        metadata: Additional metadata to attach to the log entry.
    """

fetch(logs_model: LogsResponse, start: Optional[str] = None, limit: Optional[int] = None, before: Optional[str] = None, after: Optional[str] = None, filter_: Optional[LogsEntriesFilter] = None) -> LogsEntriesResponse abstractmethod

Fetch a page of log entries.

Implementations must use dependencies available on the server. Pages are ordered chronologically regardless of the starting end.

Return cursors only for directions supported by the backend. Each cursor preserves the native token and query parameters needed for continuation. Explicit continuation parameters must match the cursor.

Parameters:

Name	Type	Description	Default
logs_model	LogsResponse	The logs model containing metadata about the logs.	required
start	Optional[str]	Initial end of the stream: oldest or newest. Defaults to the log store's preferred end.	None
limit	Optional[int]	Maximum number of log entries to return. Defaults to the log store's default_query_size.	None
before	Optional[str]	Cursor towards older entries. Mutually exclusive with after.	None
after	Optional[str]	Cursor towards newer entries. Mutually exclusive with before.	None
filter_	Optional[LogsEntriesFilter]	Filters to apply while retrieving the entries.	None

Returns:

Type	Description
LogsEntriesResponse	Log entries ordered from oldest to newest and supported cursors.

Raises:

Type	Description
ValueError	If parameters are unsupported or conflict with a cursor.

Source code in src/zenml/log_stores/base_log_store.py

@abstractmethod
def fetch(
    self,
    logs_model: LogsResponse,
    start: Optional[str] = None,
    limit: Optional[int] = None,
    before: Optional[str] = None,
    after: Optional[str] = None,
    filter_: Optional[LogsEntriesFilter] = None,
) -> LogsEntriesResponse:
    """Fetch a page of log entries.

    Implementations must use dependencies available on the server. Pages
    are ordered chronologically regardless of the starting end.

    Return cursors only for directions supported by the backend. Each
    cursor preserves the native token and query parameters needed for
    continuation. Explicit continuation parameters must match the cursor.

    Args:
        logs_model: The logs model containing metadata about the logs.
        start: Initial end of the stream: `oldest` or `newest`.
            Defaults to the log store's preferred end.
        limit: Maximum number of log entries to return. Defaults to the
            log store's `default_query_size`.
        before: Cursor towards older entries. Mutually exclusive with after.
        after: Cursor towards newer entries. Mutually exclusive with before.
        filter_: Filters to apply while retrieving the entries.

    Returns:
        Log entries ordered from oldest to newest and supported cursors.

    Raises:
        ValueError: If parameters are unsupported or conflict with a cursor.
    """

flush(blocking: bool = True) -> None abstractmethod

Flush the log store.

This method is called to ensure that all logs are flushed to the backend.

Parameters:

Name	Type	Description	Default
blocking	bool	Whether to block until the flush is complete.	True

Source code in src/zenml/log_stores/base_log_store.py

@abstractmethod
def flush(self, blocking: bool = True) -> None:
    """Flush the log store.

    This method is called to ensure that all logs are flushed to the backend.

    Args:
        blocking: Whether to block until the flush is complete.
    """

register_origin(name: str, log_model: LogsResponse, metadata: Dict[str, Any]) -> BaseLogStoreOrigin

Register an origin for the log store.

Parameters:

Name	Type	Description	Default
name	str	The name of the origin.	required
log_model	LogsResponse	The log model associated with the origin.	required
metadata	Dict[str, Any]	Additional metadata to attach to the log entry.	required

Returns:

Type	Description
BaseLogStoreOrigin	The origin.

Source code in src/zenml/log_stores/base_log_store.py

def register_origin(
    self, name: str, log_model: LogsResponse, metadata: Dict[str, Any]
) -> BaseLogStoreOrigin:
    """Register an origin for the log store.

    Args:
        name: The name of the origin.
        log_model: The log model associated with the origin.
        metadata: Additional metadata to attach to the log entry.

    Returns:
        The origin.
    """
    with self._lock:
        origin = self.origin_class(name, self, log_model, metadata)
        self._origins[name] = origin
        return origin

resolve_limit(limit: Optional[int]) -> int

Resolve the page size and apply the global limit.

Parameters:

Name	Type	Description	Default
limit	Optional[int]	The limit requested by the caller, if any.	required

Returns:

Type	Description
int	The page size, capped at LOGS_MAX_ENTRIES_PER_REQUEST.

Raises:

Type	Description
ValueError	If the requested limit is not positive.

Source code in src/zenml/log_stores/base_log_store.py

def resolve_limit(self, limit: Optional[int]) -> int:
    """Resolve the page size and apply the global limit.

    Args:
        limit: The limit requested by the caller, if any.

    Returns:
        The page size, capped at `LOGS_MAX_ENTRIES_PER_REQUEST`.

    Raises:
        ValueError: If the requested limit is not positive.
    """
    if limit is None:
        limit = self.default_query_size
    elif limit <= 0:
        raise ValueError("`limit` must be a positive integer.")

    return min(limit, LOGS_MAX_ENTRIES_PER_REQUEST)

BaseLogStoreConfig(warn_about_plain_text_secrets: bool = False, **kwargs: Any)

Bases: StackComponentConfig

Base configuration for all log stores.

Source code in src/zenml/stack/stack_component.py

def __init__(
    self, warn_about_plain_text_secrets: bool = False, **kwargs: Any
) -> None:
    """Ensures that secret references don't clash with pydantic validation.

    StackComponents allow the specification of all their string attributes
    using secret references of the form `{{secret_name.key}}`. This however
    is only possible when the stack component does not perform any explicit
    validation of this attribute using pydantic validators. If this were
    the case, the validation would run on the secret reference and would
    fail or in the worst case, modify the secret reference and lead to
    unexpected behavior. This method ensures that no attributes that require
    custom pydantic validation are set as secret references.

    Args:
        warn_about_plain_text_secrets: If true, then warns about using
            plain-text secrets.
        **kwargs: Arguments to initialize this stack component.

    Raises:
        ValueError: If an attribute that requires custom pydantic validation
            is passed as a secret reference, or if the `name` attribute
            was passed as a secret reference.
    """
    for key, value in kwargs.items():
        try:
            field = self.__class__.model_fields[key]
        except KeyError:
            # Value for a private attribute or non-existing field, this
            # will fail during the upcoming pydantic validation
            continue

        if value is None:
            continue

        if not secret_utils.is_secret_reference(value):
            if (
                secret_utils.is_secret_field(field)
                and warn_about_plain_text_secrets
            ):
                logger.warning(
                    "You specified a plain-text value for the sensitive "
                    f"attribute `{key}` for a `{self.__class__.__name__}` "
                    "stack component. This is currently only a warning, "
                    "but future versions of ZenML will require you to pass "
                    "in sensitive information as secrets. Check out the "
                    "documentation on how to configure your stack "
                    "components with secrets here: "
                    "https://docs.zenml.io/deploying-zenml/deploying-zenml/secret-management"
                )
            continue

        if pydantic_utils.has_validators(
            pydantic_class=self.__class__, field_name=key
        ):
            raise ValueError(
                f"Passing the stack component attribute `{key}` as a "
                "secret reference is not allowed as additional validation "
                "is required for this attribute."
            )

    super().__init__(**kwargs)

BaseLogStoreFlavor

Bases: Flavor

Base class for all ZenML log store flavors.

Attributes

config_class: Type[BaseLogStoreConfig] property

Config class for the base log store flavor.

Returns:

Type	Description
Type[BaseLogStoreConfig]	The config class.

implementation_class: Type[BaseLogStore] abstractmethod property

Implementation class for the base log store flavor.

Returns:

Type	Description
Type[BaseLogStore]	The implementation class.

type: StackComponentType property

Type of the flavor.

Returns:

Type	Description
StackComponentType	The type of the flavor.

BaseLogStoreOrigin(name: str, log_store: BaseLogStore, log_model: LogsResponse, metadata: Dict[str, Any])

Base class for all ZenML log store origins.

The origin is the entry point for all log records to be sent to the log store for processing. The process of sending a log record is as follows:

1. instantiate the log store or use the active log store
2. register an origin by calling log_store.register_origin() and passing the log model and optional metadata to be attached to each log record
3. emit the log record by calling log_store.emit() and passing the origin and log record
4. deregister the origin when all logs have been emitted by calling log_store.deregister(origin)

Initialize a log store origin.

Parameters:

Name	Type	Description	Default
name	str	The name of the origin.	required
log_store	BaseLogStore	The log store to emit logs to.	required
log_model	LogsResponse	The log model associated with the origin.	required
metadata	Dict[str, Any]	Additional metadata to attach to all log entries that will be emitted by this origin.	required

Source code in src/zenml/log_stores/base_log_store.py

def __init__(
    self,
    name: str,
    log_store: "BaseLogStore",
    log_model: LogsResponse,
    metadata: Dict[str, Any],
) -> None:
    """Initialize a log store origin.

    Args:
        name: The name of the origin.
        log_store: The log store to emit logs to.
        log_model: The log model associated with the origin.
        metadata: Additional metadata to attach to all log entries that will
            be emitted by this origin.
    """
    self.name = name
    self.log_store = log_store
    self.log_model = log_model
    self.metadata = metadata

Methods:

datadog

Datadog log store implementation.

Modules

datadog_flavor

Datadog log store flavor.

Classes

DatadogLogStoreConfig(warn_about_plain_text_secrets: bool = False, **kwargs: Any)

Bases: OtelLogStoreConfig

Configuration for Datadog log store.

Attributes:

Name	Type	Description
api_key	PlainSerializedSecretStr	Datadog API key for log ingestion.
application_key	PlainSerializedSecretStr	Datadog application key for log extraction.
site	str	Datadog site (e.g., "datadoghq.com", "datadoghq.eu").
max_export_batch_size	int	Maximum batch size for exports (Datadog limit: 1000).

Source code in src/zenml/stack/stack_component.py

def __init__(
    self, warn_about_plain_text_secrets: bool = False, **kwargs: Any
) -> None:
    """Ensures that secret references don't clash with pydantic validation.

    StackComponents allow the specification of all their string attributes
    using secret references of the form `{{secret_name.key}}`. This however
    is only possible when the stack component does not perform any explicit
    validation of this attribute using pydantic validators. If this were
    the case, the validation would run on the secret reference and would
    fail or in the worst case, modify the secret reference and lead to
    unexpected behavior. This method ensures that no attributes that require
    custom pydantic validation are set as secret references.

    Args:
        warn_about_plain_text_secrets: If true, then warns about using
            plain-text secrets.
        **kwargs: Arguments to initialize this stack component.

    Raises:
        ValueError: If an attribute that requires custom pydantic validation
            is passed as a secret reference, or if the `name` attribute
            was passed as a secret reference.
    """
    for key, value in kwargs.items():
        try:
            field = self.__class__.model_fields[key]
        except KeyError:
            # Value for a private attribute or non-existing field, this
            # will fail during the upcoming pydantic validation
            continue

        if value is None:
            continue

        if not secret_utils.is_secret_reference(value):
            if (
                secret_utils.is_secret_field(field)
                and warn_about_plain_text_secrets
            ):
                logger.warning(
                    "You specified a plain-text value for the sensitive "
                    f"attribute `{key}` for a `{self.__class__.__name__}` "
                    "stack component. This is currently only a warning, "
                    "but future versions of ZenML will require you to pass "
                    "in sensitive information as secrets. Check out the "
                    "documentation on how to configure your stack "
                    "components with secrets here: "
                    "https://docs.zenml.io/deploying-zenml/deploying-zenml/secret-management"
                )
            continue

        if pydantic_utils.has_validators(
            pydantic_class=self.__class__, field_name=key
        ):
            raise ValueError(
                f"Passing the stack component attribute `{key}` as a "
                "secret reference is not allowed as additional validation "
                "is required for this attribute."
            )

    super().__init__(**kwargs)

Methods:

set_default_endpoint(data: Dict[str, Any]) -> Dict[str, Any] classmethod

Set the endpoint based on site if not provided.

Parameters:

Name	Type	Description	Default
data	Dict[str, Any]	The input data dictionary.	required

Returns:

Type	Description
Dict[str, Any]	The data dictionary with the endpoint set if not provided.

Source code in src/zenml/log_stores/datadog/datadog_flavor.py

@model_validator(mode="before")
@classmethod
def set_default_endpoint(cls, data: Dict[str, Any]) -> Dict[str, Any]:
    """Set the endpoint based on site if not provided.

    Args:
        data: The input data dictionary.

    Returns:
        The data dictionary with the endpoint set if not provided.
    """
    if isinstance(data, dict) and not data.get("endpoint"):
        site = data.get("site", "datadoghq.com")
        data["endpoint"] = f"https://http-intake.logs.{site}/api/v2/logs"
    return data

validate_max_export_batch_size(v: int) -> int classmethod

Validate that max_export_batch_size doesn't exceed Datadog's limit.

Parameters:

Name	Type	Description	Default
v	int	The value to validate.	required

Returns:

Type	Description
int	The validated value.

Raises:

Type	Description
ValueError	If the value exceeds Datadog's limit.

Source code in src/zenml/log_stores/datadog/datadog_flavor.py

@field_validator("max_export_batch_size")
@classmethod
def validate_max_export_batch_size(cls, v: int) -> int:
    """Validate that max_export_batch_size doesn't exceed Datadog's limit.

    Args:
        v: The value to validate.

    Returns:
        The validated value.

    Raises:
        ValueError: If the value exceeds Datadog's limit.
    """
    if v > DATADOG_MAX_BATCH_SIZE:
        raise ValueError(
            f"max_export_batch_size cannot exceed {DATADOG_MAX_BATCH_SIZE} "
            f"(Datadog API limit). Got: {v}"
        )
    return v

DatadogLogStoreFlavor

Bases: Flavor

Datadog log store flavor.

Attributes

config_class: Type[BaseLogStoreConfig] property

Returns DatadogLogStoreConfig config class.

Returns:

Type	Description
Type[BaseLogStoreConfig]	The config class.

docs_url: str property

URL to the flavor documentation.

Returns:

Type	Description
str	The URL to the flavor documentation.

implementation_class: Type[BaseLogStore] property

Implementation class for this flavor.

Returns:

Type	Description
Type[BaseLogStore]	The implementation class.

logo_url: str property

URL to the flavor logo.

Returns:

Type	Description
str	The URL to the flavor logo.

name: str property

Name of the flavor.

Returns:

Type	Description
str	The name of the flavor.

sdk_docs_url: str property

URL to the SDK docs for this flavor.

Returns:

Type	Description
str	The URL to the SDK docs for this flavor.

type: StackComponentType property

Stack component type.

Returns:

Type	Description
StackComponentType	The stack component type.

datadog_log_exporter

Log exporter that writes logs to Datadog.

Classes

DatadogLogExporter(endpoint: str, certificate_file: Optional[str] = None, client_key_file: Optional[str] = None, client_certificate_file: Optional[str] = None, headers: Optional[Dict[str, str]] = None, timeout: float = DEFAULT_TIMEOUT, compression: Compression = Compression.NoCompression)

Bases: OTLPLogExporter

Datadog log exporter.

This exporter writes OpenTelemetry logs to Datadog's HTTP intake API with slightly modified log records to adapt to Datadog's format.

Source code in src/zenml/log_stores/otel/otel_log_exporter.py

def __init__(
    self,
    endpoint: str,
    certificate_file: Optional[str] = None,
    client_key_file: Optional[str] = None,
    client_certificate_file: Optional[str] = None,
    headers: Optional[Dict[str, str]] = None,
    timeout: float = DEFAULT_TIMEOUT,
    compression: Compression = Compression.NoCompression,
):
    """Initialize the exporter.

    Args:
        endpoint: The endpoint to export logs to.
        certificate_file: The certificate file to use for the export.
        client_key_file: The client key file to use for the export.
        client_certificate_file: The client certificate file to use for the export.
        headers: The headers to use for the export.
        timeout: The timeout to use for the export.
        compression: The compression to use for the export.
    """
    self._shutdown_is_occurring = threading.Event()
    self._endpoint = endpoint
    self._certificate_file = certificate_file
    self._client_key_file = client_key_file
    self._client_certificate_file = client_certificate_file
    self._client_cert = (
        (self._client_certificate_file, self._client_key_file)
        if self._client_certificate_file and self._client_key_file
        else self._client_certificate_file
    )
    self._timeout = timeout
    self._compression = compression
    self._session = requests.Session()
    self._session.headers.update(
        {
            "Content-Type": "application/json",
            "Accept": "application/json",
            "User-Agent": f"zenml/{zenml_version}",
        }
    )
    if headers:
        self._session.headers.update(headers)

    # Retries are triggered on specific HTTP status codes:
    #
    #     408: Request Timeout.
    #     429: Too Many Requests.
    #     502: Bad Gateway.
    #     503: Service Unavailable.
    #     504: Gateway Timeout
    #
    # This also handles connection level errors, if a connection attempt
    # fails due to transient issues like:
    #
    #     DNS resolution errors.
    #     Connection timeouts.
    #     Network disruptions.
    #
    # Additional errors retried:
    #
    #     Read Timeouts: If the server does not send a response within
    #     the timeout period.
    #     Connection Refused: If the server refuses the connection.
    #
    retries = Retry(
        connect=5,
        read=5,
        redirect=3,
        status=5,
        allowed_methods=[
            "POST",
        ],
        status_forcelist=[
            408,  # Request Timeout
            429,  # Too Many Requests
            500,  # Internal Server Error
            502,  # Bad Gateway
            503,  # Service Unavailable
            504,  # Gateway Timeout
        ],
        other=3,
        backoff_factor=0.5,
        respect_retry_after_header=True,
        raise_on_status=False,
    )
    http_adapter = HTTPAdapter(
        max_retries=retries,
        pool_maxsize=1,
    )
    self._session.mount("https://", http_adapter)
    self._session.mount("http://", http_adapter)

    self._shutdown = False

Functions:

datadog_log_store

Datadog log store implementation.

Classes

DatadogLogStore(*args: Any, **kwargs: Any)

Bases: OtelLogStore

Log store that exports logs to Datadog.

This implementation extends OtelLogStore and configures it to send logs to Datadog's HTTP intake API.

Source code in src/zenml/log_stores/otel/otel_log_store.py

def __init__(self, *args: Any, **kwargs: Any) -> None:
    """Initialize the OpenTelemetry log store.

    Args:
        *args: Positional arguments for the base class.
        **kwargs: Keyword arguments for the base class.
    """
    super().__init__(*args, **kwargs)

    self._resource: Optional["Resource"] = None
    self._exporter: Optional["LogRecordExporter"] = None
    self._provider: Optional["LoggerProvider"] = None
    self._processor: Optional["OtelBatchLogRecordProcessor"] = None
    self._handler: Optional["LoggingHandler"] = None

Attributes

config: DatadogLogStoreConfig property

Returns the configuration of the Datadog log store.

Returns:

Type	Description
DatadogLogStoreConfig	The configuration.

default_query_size: int property

The maximum page size supported by Datadog's search API.

Returns:

Type	Description
int	The default number of entries per page.

Methods:

cleanup() -> None

Cleanup the Datadog log store.

This method is called when the log store is no longer needed.

Source code in src/zenml/log_stores/datadog/datadog_log_store.py

def cleanup(self) -> None:
    """Cleanup the Datadog log store.

    This method is called when the log store is no longer needed.
    """
    if self._datadog_exporter:
        self._datadog_exporter.shutdown()
        self._datadog_exporter = None

fetch(logs_model: LogsResponse, start: Optional[str] = None, limit: Optional[int] = None, before: Optional[str] = None, after: Optional[str] = None, filter_: Optional[LogsEntriesFilter] = None) -> LogsEntriesResponse

Fetch a page of log entries from the Datadog Logs API.

Cursors retain the filters, time bounds, direction, and page size.

Parameters:

Name	Type	Description	Default
logs_model	LogsResponse	The logs model containing run and step metadata.	required
start	Optional[str]	Initial end of the stream. Defaults to newest.	None
limit	Optional[int]	Maximum number of log entries to return.	None
before	Optional[str]	Cursor towards older entries, from a previous page.	None
after	Optional[str]	Cursor towards newer entries, from a previous page.	None
filter_	Optional[LogsEntriesFilter]	Filters for a new read. Explicit continuation filters must match the cursor.	None

Returns:

Type	Description
LogsEntriesResponse	A page of log entries, oldest first.

Raises:

Type	Description
ValueError	If the logs model does not belong to this log store, or the pagination parameters are invalid.
LogStoreError	If Datadog returns an error or unusable response.

Source code in src/zenml/log_stores/datadog/datadog_log_store.py

def fetch(
    self,
    logs_model: "LogsResponse",
    start: Optional[str] = None,
    limit: Optional[int] = None,
    before: Optional[str] = None,
    after: Optional[str] = None,
    filter_: Optional[LogsEntriesFilter] = None,
) -> LogsEntriesResponse:
    """Fetch a page of log entries from the Datadog Logs API.

    Cursors retain the filters, time bounds, direction, and page size.

    Args:
        logs_model: The logs model containing run and step metadata.
        start: Initial end of the stream. Defaults to `newest`.
        limit: Maximum number of log entries to return.
        before: Cursor towards older entries, from a previous page.
        after: Cursor towards newer entries, from a previous page.
        filter_: Filters for a new read. Explicit continuation filters
            must match the cursor.

    Returns:
        A page of log entries, oldest first.

    Raises:
        ValueError: If the logs model does not belong to this log store,
            or the pagination parameters are invalid.
        LogStoreError: If Datadog returns an error or unusable response.
    """  # noqa: DOC503
    if logs_model.log_store_id != self.id:
        raise ValueError(
            "logs_model.log_store_id does not match the id of the log "
            "store. These entries were collected by another log store, "
            "which is the one that can read them back."
        )

    if before is not None and after is not None:
        raise ValueError("Pass only one of `before` and `after`.")
    if start not in (None, "oldest", "newest"):
        raise ValueError("`start` must be `oldest` or `newest`.")

    filter_ = filter_ or LogsEntriesFilter()
    page_cursor = before if before is not None else after
    cursor = None
    direction: Literal["oldest", "newest"]
    if page_cursor is not None:
        cursor = _DatadogCursor.decode(page_cursor)
        direction = "newest" if before is not None else "oldest"
        if (
            cursor.log_store_id != self.id
            or cursor.logs_id != logs_model.id
            or cursor.direction != direction
        ):
            raise ValueError(
                "Invalid pagination cursor for this log store, stream, "
                "or direction."
            )
        if start is not None and start != cursor.direction:
            raise ValueError(
                "`start` conflicts with the pagination cursor."
            )
        if (
            limit is not None
            and min(self.resolve_limit(limit), DATADOG_MAX_PAGE_SIZE)
            != cursor.limit
        ):
            raise ValueError(
                "`limit` conflicts with the pagination cursor."
            )
        saved_filters = cursor.filters.model_dump()
        for name, value in filter_.model_dump(exclude_none=True).items():
            if value != saved_filters[name]:
                raise ValueError(
                    f"`{name}` conflicts with the pagination cursor."
                )
        filters = cursor.filters
        limit = self.resolve_limit(cursor.limit)
    else:
        direction = "oldest" if start == "oldest" else "newest"
        limit = min(self.resolve_limit(limit), DATADOG_MAX_PAGE_SIZE)
        filters = _DatadogFilters(
            search=filter_.search,
            level=filter_.level,
            since=filter_.since or to_utc_timezone(logs_model.created),
            until=filter_.until or utc_now(tz_aware=True),
        )
    descending = direction == "newest"

    body: Dict[str, Any] = {
        "filter": {
            "query": self._build_query(logs_model, filters),
            "from": filters.since.isoformat(),
            "to": filters.until.isoformat(),
        },
        "page": {"limit": limit},
        "sort": "-timestamp" if descending else "timestamp",
    }
    if cursor is not None:
        body["page"]["cursor"] = cursor.token

    events, native_cursor = self._search(body)
    entries = [
        entry
        for entry in (self._parse_log_entry(event) for event in events)
        if entry is not None
    ]
    if descending:
        entries.reverse()

    encoded = (
        _DatadogCursor(
            version=1,
            token=native_cursor,
            log_store_id=self.id,
            logs_id=logs_model.id,
            filters=filters,
            direction=direction,
            limit=limit,
        ).encode()
        if native_cursor
        else None
    )
    return LogsEntriesResponse(
        items=entries,
        before=encoded if descending else None,
        after=encoded if not descending else None,
    )

get_exporter() -> DatadogLogExporter

Get the Datadog log exporter.

Returns:

Type	Description
DatadogLogExporter	DatadogExporter with the proper configuration.

Source code in src/zenml/log_stores/datadog/datadog_log_store.py

def get_exporter(self) -> DatadogLogExporter:
    """Get the Datadog log exporter.

    Returns:
        DatadogExporter with the proper configuration.
    """
    if not self._datadog_exporter:
        self._datadog_exporter = DatadogLogExporter(
            endpoint=self.config.endpoint,
            headers=self._get_headers(),
            certificate_file=self.config.certificate_file,
            client_key_file=self.config.client_key_file,
            client_certificate_file=self.config.client_certificate_file,
            compression=self.config.compression,
        )
    return self._datadog_exporter

Functions:

otel

OpenTelemetry log store implementation.

Modules

otel_flavor

OpenTelemetry log store flavor.

Classes

Compression

Bases: StrEnum

Compression types.

OtelLogStoreConfig(warn_about_plain_text_secrets: bool = False, **kwargs: Any)

Bases: BaseLogStoreConfig

Configuration for OpenTelemetry log store.

Attributes:

Name	Type	Description
service_name	str	Name of the service (defaults to "zenml").
service_version	str	Version of the service (defaults to the ZenML version).
max_queue_size	int	Maximum queue size for batch processor.
schedule_delay_millis	int	Delay between batch exports in milliseconds.
max_export_batch_size	int	Maximum batch size for exports.
endpoint	str	The endpoint to export logs to.
headers	Optional[Dict[str, str]]	The headers to use for the export.
certificate_file	Optional[str]	The certificate file to use for the export.
client_key_file	Optional[str]	The client key file to use for the export.
client_certificate_file	Optional[str]	The client certificate file to use for the export.
compression	Compression	The compression to use for the export.

Source code in src/zenml/stack/stack_component.py

def __init__(
    self, warn_about_plain_text_secrets: bool = False, **kwargs: Any
) -> None:
    """Ensures that secret references don't clash with pydantic validation.

    StackComponents allow the specification of all their string attributes
    using secret references of the form `{{secret_name.key}}`. This however
    is only possible when the stack component does not perform any explicit
    validation of this attribute using pydantic validators. If this were
    the case, the validation would run on the secret reference and would
    fail or in the worst case, modify the secret reference and lead to
    unexpected behavior. This method ensures that no attributes that require
    custom pydantic validation are set as secret references.

    Args:
        warn_about_plain_text_secrets: If true, then warns about using
            plain-text secrets.
        **kwargs: Arguments to initialize this stack component.

    Raises:
        ValueError: If an attribute that requires custom pydantic validation
            is passed as a secret reference, or if the `name` attribute
            was passed as a secret reference.
    """
    for key, value in kwargs.items():
        try:
            field = self.__class__.model_fields[key]
        except KeyError:
            # Value for a private attribute or non-existing field, this
            # will fail during the upcoming pydantic validation
            continue

        if value is None:
            continue

        if not secret_utils.is_secret_reference(value):
            if (
                secret_utils.is_secret_field(field)
                and warn_about_plain_text_secrets
            ):
                logger.warning(
                    "You specified a plain-text value for the sensitive "
                    f"attribute `{key}` for a `{self.__class__.__name__}` "
                    "stack component. This is currently only a warning, "
                    "but future versions of ZenML will require you to pass "
                    "in sensitive information as secrets. Check out the "
                    "documentation on how to configure your stack "
                    "components with secrets here: "
                    "https://docs.zenml.io/deploying-zenml/deploying-zenml/secret-management"
                )
            continue

        if pydantic_utils.has_validators(
            pydantic_class=self.__class__, field_name=key
        ):
            raise ValueError(
                f"Passing the stack component attribute `{key}` as a "
                "secret reference is not allowed as additional validation "
                "is required for this attribute."
            )

    super().__init__(**kwargs)

OtelLogStoreFlavor

Bases: Flavor

OpenTelemetry log store flavor.

Attributes

config_class: Type[BaseLogStoreConfig] property

Returns DatadogLogStoreConfig config class.

Returns:

Type	Description
Type[BaseLogStoreConfig]	The config class.

docs_url: str property

URL to the flavor documentation.

Returns:

Type	Description
str	The URL to the flavor documentation.

implementation_class: Type[BaseLogStore] property

Implementation class for this flavor.

Returns:

Type	Description
Type[BaseLogStore]	The implementation class.

logo_url: str property

URL to the flavor logo.

Returns:

Type	Description
str	The URL to the flavor logo.

name: str property

Name of the flavor.

Returns:

Type	Description
str	The name of the flavor.

sdk_docs_url: str property

URL to the SDK docs for this flavor.

Returns:

Type	Description
str	The URL to the SDK docs for this flavor.

type: StackComponentType property

Stack component type.

Returns:

Type	Description
StackComponentType	The stack component type.

otel_log_exporter

OpenTelemetry exporter that writes logs to any OpenTelemetry backend.

Classes

OTLPLogExporter(endpoint: str, certificate_file: Optional[str] = None, client_key_file: Optional[str] = None, client_certificate_file: Optional[str] = None, headers: Optional[Dict[str, str]] = None, timeout: float = DEFAULT_TIMEOUT, compression: Compression = Compression.NoCompression)

Bases: LogRecordExporter

OpenTelemetry exporter using JSON protocol.

This exporter is a placeholder until the actual implementation of the OpenTelemetry exporter is available in the opentelemetry-exporter-otlp-proto-json package.

Initialize the exporter.

Parameters:

Name	Type	Description	Default
endpoint	str	The endpoint to export logs to.	required
certificate_file	Optional[str]	The certificate file to use for the export.	None
client_key_file	Optional[str]	The client key file to use for the export.	None
client_certificate_file	Optional[str]	The client certificate file to use for the export.	None
headers	Optional[Dict[str, str]]	The headers to use for the export.	None
timeout	float	The timeout to use for the export.	DEFAULT_TIMEOUT
compression	Compression	The compression to use for the export.	NoCompression

Source code in src/zenml/log_stores/otel/otel_log_exporter.py

def __init__(
    self,
    endpoint: str,
    certificate_file: Optional[str] = None,
    client_key_file: Optional[str] = None,
    client_certificate_file: Optional[str] = None,
    headers: Optional[Dict[str, str]] = None,
    timeout: float = DEFAULT_TIMEOUT,
    compression: Compression = Compression.NoCompression,
):
    """Initialize the exporter.

    Args:
        endpoint: The endpoint to export logs to.
        certificate_file: The certificate file to use for the export.
        client_key_file: The client key file to use for the export.
        client_certificate_file: The client certificate file to use for the export.
        headers: The headers to use for the export.
        timeout: The timeout to use for the export.
        compression: The compression to use for the export.
    """
    self._shutdown_is_occurring = threading.Event()
    self._endpoint = endpoint
    self._certificate_file = certificate_file
    self._client_key_file = client_key_file
    self._client_certificate_file = client_certificate_file
    self._client_cert = (
        (self._client_certificate_file, self._client_key_file)
        if self._client_certificate_file and self._client_key_file
        else self._client_certificate_file
    )
    self._timeout = timeout
    self._compression = compression
    self._session = requests.Session()
    self._session.headers.update(
        {
            "Content-Type": "application/json",
            "Accept": "application/json",
            "User-Agent": f"zenml/{zenml_version}",
        }
    )
    if headers:
        self._session.headers.update(headers)

    # Retries are triggered on specific HTTP status codes:
    #
    #     408: Request Timeout.
    #     429: Too Many Requests.
    #     502: Bad Gateway.
    #     503: Service Unavailable.
    #     504: Gateway Timeout
    #
    # This also handles connection level errors, if a connection attempt
    # fails due to transient issues like:
    #
    #     DNS resolution errors.
    #     Connection timeouts.
    #     Network disruptions.
    #
    # Additional errors retried:
    #
    #     Read Timeouts: If the server does not send a response within
    #     the timeout period.
    #     Connection Refused: If the server refuses the connection.
    #
    retries = Retry(
        connect=5,
        read=5,
        redirect=3,
        status=5,
        allowed_methods=[
            "POST",
        ],
        status_forcelist=[
            408,  # Request Timeout
            429,  # Too Many Requests
            500,  # Internal Server Error
            502,  # Bad Gateway
            503,  # Service Unavailable
            504,  # Gateway Timeout
        ],
        other=3,
        backoff_factor=0.5,
        respect_retry_after_header=True,
        raise_on_status=False,
    )
    http_adapter = HTTPAdapter(
        max_retries=retries,
        pool_maxsize=1,
    )
    self._session.mount("https://", http_adapter)
    self._session.mount("http://", http_adapter)

    self._shutdown = False

Methods:

export(batch: Sequence[ReadableLogRecord]) -> LogRecordExportResult

Export a batch of logs to the OpenTelemetry backend.

Parameters:

Name	Type	Description	Default
batch	Sequence[ReadableLogRecord]	The batch of logs to export.	required

Returns:

Type	Description
LogRecordExportResult	LogRecordExportResult indicating success or failure.

Source code in src/zenml/log_stores/otel/otel_log_exporter.py

def export(
    self, batch: Sequence[ReadableLogRecord]
) -> LogRecordExportResult:
    """Export a batch of logs to the OpenTelemetry backend.

    Args:
        batch: The batch of logs to export.

    Returns:
        LogRecordExportResult indicating success or failure.
    """
    if self._shutdown:
        logger.warning("Exporter already shutdown, ignoring batch")
        return LogRecordExportResult.FAILURE
    encoded_logs = self._encode_logs(batch)

    serialized_data = json.dumps(
        encoded_logs,
        default=pydantic_encoder,
    ).encode("utf-8")

    try:
        resp = self._export(serialized_data, self._timeout)
        if resp.ok:
            return LogRecordExportResult.SUCCESS
        return LogRecordExportResult.FAILURE
    except Exception as e:
        logger.error(f"Error exporting logs: {e}")
        return LogRecordExportResult.FAILURE

shutdown() -> None

Shutdown the exporter.

Source code in src/zenml/log_stores/otel/otel_log_exporter.py

def shutdown(self) -> None:
    """Shutdown the exporter."""
    if self._shutdown:
        logger.warning("Exporter already shutdown, ignoring call")
        return
    self._shutdown = True
    self._shutdown_is_occurring.set()
    self._session.close()

Functions:

otel_log_store

OpenTelemetry log store implementation.

Classes

OtelBatchLogRecordProcessor

Bases: BatchLogRecordProcessor

OpenTelemetry batch log record processor.

This is a subclass of the BatchLogRecordProcessor that allows for a non-blocking flush.

Methods:

flush(blocking: bool = True) -> bool

Force flush the batch log record processor.

Parameters:

Name	Type	Description	Default
blocking	bool	Whether to block until the flush is complete.	True

Returns:

Type	Description
bool	True if the flush is successful, False otherwise.

Source code in src/zenml/log_stores/otel/otel_log_store.py

def flush(self, blocking: bool = True) -> bool:
    """Force flush the batch log record processor.

    Args:
        blocking: Whether to block until the flush is complete.

    Returns:
        True if the flush is successful, False otherwise.
    """
    if not blocking:
        # For a non-blocking flush, we simply need to wake up the worker
        # thread and it will handle the flush in the background.
        self._batch_processor._worker_awaken.set()
        return True
    else:
        return self.force_flush()

OtelLogStore(*args: Any, **kwargs: Any)

Bases: BaseLogStore

Log store that exports logs using OpenTelemetry.

Subclasses should implement get_exporter() to provide the specific log exporter for their backend.

Initialize the OpenTelemetry log store.

Parameters:

Name	Type	Description	Default
*args	Any	Positional arguments for the base class.	()
**kwargs	Any	Keyword arguments for the base class.	{}

Source code in src/zenml/log_stores/otel/otel_log_store.py

def __init__(self, *args: Any, **kwargs: Any) -> None:
    """Initialize the OpenTelemetry log store.

    Args:
        *args: Positional arguments for the base class.
        **kwargs: Keyword arguments for the base class.
    """
    super().__init__(*args, **kwargs)

    self._resource: Optional["Resource"] = None
    self._exporter: Optional["LogRecordExporter"] = None
    self._provider: Optional["LoggerProvider"] = None
    self._processor: Optional["OtelBatchLogRecordProcessor"] = None
    self._handler: Optional["LoggingHandler"] = None

Attributes

config: OtelLogStoreConfig property

Returns the configuration of the OTel log store.

Returns:

Type	Description
OtelLogStoreConfig	The configuration.

origin_class: Type[OtelLogStoreOrigin] property

Class of the origin.

Returns:

Type	Description
Type[OtelLogStoreOrigin]	The class of the origin.

provider: LoggerProvider property

Returns the OpenTelemetry logger provider.

Returns:

Type	Description
LoggerProvider	The logger provider.

Raises:

Type	Description
RuntimeError	If the OpenTelemetry log store is not initialized.

Methods:

deactivate() -> None

Deactivate log collection and shut down the processor.

Flushes any pending logs and shuts down the processor's background thread.

Source code in src/zenml/log_stores/otel/otel_log_store.py

def deactivate(self) -> None:
    """Deactivate log collection and shut down the processor.

    Flushes any pending logs and shuts down the processor's background thread.
    """
    with self._lock:
        if self._processor:
            try:
                # Force flush any pending logs
                self._processor.flush(blocking=True)
                logger.debug("Flushed pending logs")
            except Exception as e:
                logger.warning(f"Error flushing logs: {e}")

            try:
                self._processor.shutdown()  # type: ignore[no-untyped-call]
                logger.debug(
                    "Shut down log processor and background thread"
                )
            except Exception as e:
                logger.warning(f"Error shutting down processor: {e}")
            else:
                self._processor = None
                self._handler = None
                self._provider = None
                self._resource = None
                self._exporter = None

    logger.debug("OtelLogStore deactivated")

emit(origin: BaseLogStoreOrigin, record: logging.LogRecord, metadata: Optional[Dict[str, Any]] = None) -> None

Process a log record by sending to OpenTelemetry.

Parameters:

Name	Type	Description	Default
origin	BaseLogStoreOrigin	The origin used to send the log record.	required
record	LogRecord	The log record to process.	required
metadata	Optional[Dict[str, Any]]	Additional metadata to attach to the log entry.	None

Raises:

Type	Description
RuntimeError	If the OpenTelemetry provider is not initialized.

Source code in src/zenml/log_stores/otel/otel_log_store.py

def emit(
    self,
    origin: BaseLogStoreOrigin,
    record: logging.LogRecord,
    metadata: Optional[Dict[str, Any]] = None,
) -> None:
    """Process a log record by sending to OpenTelemetry.

    Args:
        origin: The origin used to send the log record.
        record: The log record to process.
        metadata: Additional metadata to attach to the log entry.

    Raises:
        RuntimeError: If the OpenTelemetry provider is not initialized.
    """
    assert isinstance(origin, OtelLogStoreOrigin)
    with self._lock:
        if not self._provider:
            self._activate()

        if self._handler is None:
            raise RuntimeError("OpenTelemetry provider is not initialized")

        otel_log_record = self._handler._translate(record)
        merged_attrs = dict(otel_log_record.attributes or {})
        merged_attrs.update(origin.metadata)
        if metadata:
            merged_attrs.update(metadata)
        otel_log_record.attributes = merged_attrs
        origin.logger.emit(otel_log_record)

fetch(logs_model: LogsResponse, start: Optional[str] = None, limit: Optional[int] = None, before: Optional[str] = None, after: Optional[str] = None, filter_: Optional[LogsEntriesFilter] = None) -> LogsEntriesResponse

Fetch log entries from an OpenTelemetry backend.

OTLP does not support log retrieval. Subclasses must implement their backend's query API.

Parameters:

Name	Type	Description	Default
logs_model	LogsResponse	The logs model containing run and step metadata.	required
start	Optional[str]	Initial end of the stream, chosen by the subclass if omitted.	None
limit	Optional[int]	Maximum number of log entries to return.	None
before	Optional[str]	Cursor towards older entries, from a previous page.	None
after	Optional[str]	Cursor towards newer entries, from a previous page.	None
filter_	Optional[LogsEntriesFilter]	Filters to apply while retrieving the entries.	None

Returns:

Type	Description
LogsEntriesResponse	Never returns.

Raises:

Type	Description
NotImplementedError	Log retrieval is not supported by OTLP.

Source code in src/zenml/log_stores/otel/otel_log_store.py

def fetch(
    self,
    logs_model: "LogsResponse",
    start: Optional[str] = None,
    limit: Optional[int] = None,
    before: Optional[str] = None,
    after: Optional[str] = None,
    filter_: Optional["LogsEntriesFilter"] = None,
) -> "LogsEntriesResponse":
    """Fetch log entries from an OpenTelemetry backend.

    OTLP does not support log retrieval. Subclasses must implement their
    backend's query API.

    Args:
        logs_model: The logs model containing run and step metadata.
        start: Initial end of the stream, chosen by the subclass if omitted.
        limit: Maximum number of log entries to return.
        before: Cursor towards older entries, from a previous page.
        after: Cursor towards newer entries, from a previous page.
        filter_: Filters to apply while retrieving the entries.

    Returns:
        Never returns.

    Raises:
        NotImplementedError: Log retrieval is not supported by OTLP.
    """
    raise NotImplementedError(
        "Log fetching is not supported by the OTEL log store."
    )

flush(blocking: bool = True) -> None

Flush the log store.

Parameters:

Name	Type	Description	Default
blocking	bool	Whether to block until the flush is complete.	True

This method is called to ensure that all logs are flushed to the backend.

Source code in src/zenml/log_stores/otel/otel_log_store.py

def flush(self, blocking: bool = True) -> None:
    """Flush the log store.

    Args:
        blocking: Whether to block until the flush is complete.

    This method is called to ensure that all logs are flushed to the backend.
    """
    with self._lock:
        if self._processor:
            self._processor.flush(blocking=blocking)

get_exporter() -> LogRecordExporter

Get the Datadog log exporter.

Returns:

Type	Description
LogRecordExporter	OTLPLogExporter configured with API key and site.

Source code in src/zenml/log_stores/otel/otel_log_store.py

def get_exporter(self) -> "LogRecordExporter":
    """Get the Datadog log exporter.

    Returns:
        OTLPLogExporter configured with API key and site.
    """
    if not self._exporter:
        self._exporter = OTLPLogExporter(
            endpoint=self.config.endpoint,
            headers=self.config.headers,
            certificate_file=self.config.certificate_file,
            client_key_file=self.config.client_key_file,
            client_certificate_file=self.config.client_certificate_file,
            compression=self.config.compression,
        )

    return self._exporter

register_origin(name: str, log_model: LogsResponse, metadata: Dict[str, Any]) -> BaseLogStoreOrigin

Register an origin for the log store.

Parameters:

Name	Type	Description	Default
name	str	The name of the origin.	required
log_model	LogsResponse	The log model associated with the origin.	required
metadata	Dict[str, Any]	Additional metadata to attach to the log entry.	required

Returns:

Type	Description
BaseLogStoreOrigin	The origin.

Source code in src/zenml/log_stores/otel/otel_log_store.py

def register_origin(
    self, name: str, log_model: LogsResponse, metadata: Dict[str, Any]
) -> BaseLogStoreOrigin:
    """Register an origin for the log store.

    Args:
        name: The name of the origin.
        log_model: The log model associated with the origin.
        metadata: Additional metadata to attach to the log entry.

    Returns:
        The origin.
    """
    with self._lock:
        if not self._provider:
            self._activate()

    return super().register_origin(name, log_model, metadata)

OtelLogStoreOrigin(name: str, log_store: BaseLogStore, log_model: LogsResponse, metadata: Dict[str, Any])

Bases: BaseLogStoreOrigin

OpenTelemetry log store origin.

Initialize a log store origin.

Parameters:

Name	Type	Description	Default
name	str	The name of the origin.	required
log_store	BaseLogStore	The log store to emit logs to.	required
log_model	LogsResponse	The log model associated with the origin.	required
metadata	Dict[str, Any]	Additional metadata to attach to all log entries that will be emitted by this origin.	required

Source code in src/zenml/log_stores/otel/otel_log_store.py

def __init__(
    self,
    name: str,
    log_store: "BaseLogStore",
    log_model: LogsResponse,
    metadata: Dict[str, Any],
) -> None:
    """Initialize a log store origin.

    Args:
        name: The name of the origin.
        log_store: The log store to emit logs to.
        log_model: The log model associated with the origin.
        metadata: Additional metadata to attach to all log entries that will
            be emitted by this origin.
    """
    metadata.update(
        {
            "zenml.log.id": str(log_model.id),
            "zenml.log.source": log_model.source,
            "zenml.log_store.id": str(log_store.id),
            "zenml.log_store.name": log_store.name,
        }
    )

    super().__init__(name, log_store, log_model, metadata)
    assert isinstance(log_store, OtelLogStore)

    self._logger = log_store.provider.get_logger(name)

Attributes

logger: Logger property

Returns the OpenTelemetry logger for this origin.

Returns:

Type	Description
Logger	The logger.

Methods:

Functions: