Log store abstraction

pyproject.toml

@@ -38,6 +38,7 @@ dependencies = [
     "docker~=7.1.0",
     "gitpython>=3.1.18,<4.0.0",
     "jsonref",
+    "opentelemetry-sdk==1.38.0",
     "packaging>=24.1",
     "psutil>=5.0.0",
     "pydantic>=2.0,<=2.11.9",

src/zenml/artifacts/utils.py

@@ -539,7 +539,7 @@ def load_artifact_visualization(
             f"Artifact '{artifact.id}' cannot be visualized because the "
             "underlying artifact store was deleted."
         )
-    artifact_store = _load_artifact_store(
+    artifact_store = load_artifact_store(
         artifact_store_id=artifact.artifact_store_id, zen_store=zen_store
     )
     try:
@@ -821,7 +821,7 @@ def _load_artifact_from_uri(
     return artifact
 
 
-def _load_artifact_store(
+def load_artifact_store(
     artifact_store_id: Union[str, "UUID"],
     zen_store: Optional["BaseZenStore"] = None,
 ) -> "BaseArtifactStore":

src/zenml/cli/stack.py

@@ -202,6 +202,14 @@ def stack() -> None:
     type=str,
     required=False,
 )
+@click.option(
+    "-l",
+    "--log_store",
+    "log_store",
+    help="Name of the log store for this stack.",
+    type=str,
+    required=False,
+)
 @click.option(
     "--set",
     "set_stack",
@@ -255,6 +263,7 @@ def register_stack(
     data_validator: Optional[str] = None,
     image_builder: Optional[str] = None,
     deployer: Optional[str] = None,
+    log_store: Optional[str] = None,
     set_stack: bool = False,
     provider: Optional[str] = None,
     connector: Optional[str] = None,
@@ -278,6 +287,7 @@ def register_stack(
         data_validator: Name of the data validator for this stack.
         image_builder: Name of the new image builder for this stack.
         deployer: Name of the deployer for this stack.
+        log_store: Name of the log store for this stack.
         set_stack: Immediately set this stack as active.
         provider: Name of the cloud provider for this stack.
         connector: Name of the service connector for this stack.
@@ -522,6 +532,7 @@ def register_stack(
             (StackComponentType.DATA_VALIDATOR, data_validator),
             (StackComponentType.FEATURE_STORE, feature_store),
             (StackComponentType.IMAGE_BUILDER, image_builder),
+            (StackComponentType.LOG_STORE, log_store),
             (StackComponentType.MODEL_DEPLOYER, model_deployer),
             (StackComponentType.MODEL_REGISTRY, model_registry),
             (StackComponentType.STEP_OPERATOR, step_operator),

src/zenml/constants.py

@@ -228,15 +228,15 @@ def handle_int_env_var(var: str, default: int = 0) -> int:
         ENV_ZENML_LOGGING_VERBOSITY, default="DEBUG"
     ).upper()
     ZENML_STORAGE_LOGGING_VERBOSITY = os.getenv(
-        ENV_ZENML_STORAGE_LOGGING_VERBOSITY, default="DEBUG"
-    ).upper()
+        ENV_ZENML_STORAGE_LOGGING_VERBOSITY, default=None
+    )
 else:
     ZENML_LOGGING_VERBOSITY = os.getenv(
         ENV_ZENML_LOGGING_VERBOSITY, default="INFO"
     ).upper()
     ZENML_STORAGE_LOGGING_VERBOSITY = os.getenv(
-        ENV_ZENML_STORAGE_LOGGING_VERBOSITY, default="INFO"
-    ).upper()
+        ENV_ZENML_STORAGE_LOGGING_VERBOSITY, default=None
+    )
 
 INSIDE_ZENML_CONTAINER = handle_bool_env_var(ENV_ZENML_CONTAINER, False)
 
@@ -547,3 +547,28 @@ def handle_int_env_var(var: str, default: int = 0) -> int:
 LOGS_MERGE_INTERVAL_SECONDS = handle_int_env_var(
     ENV_ZENML_LOGS_MERGE_INTERVAL_SECONDS, default=10 * 60
 )
+
+# OpenTelemetry log storage constants
+ENV_ZENML_LOGS_OTEL_MAX_QUEUE_SIZE = "ZENML_LOGS_OTEL_MAX_QUEUE_SIZE"
+ENV_ZENML_LOGS_OTEL_SCHEDULE_DELAY_MILLIS = (
+    "ZENML_LOGS_OTEL_SCHEDULE_DELAY_MILLIS"
+)
+ENV_ZENML_LOGS_OTEL_MAX_EXPORT_BATCH_SIZE = (
+    "ZENML_LOGS_OTEL_MAX_EXPORT_BATCH_SIZE"
+)
+ENV_ZENML_LOGS_OTEL_EXPORT_TIMEOUT_MILLIS = (
+    "ZENML_LOGS_OTEL_EXPORT_TIMEOUT_MILLIS"
+)
+
+LOGS_OTEL_MAX_QUEUE_SIZE = handle_int_env_var(
+    ENV_ZENML_LOGS_OTEL_MAX_QUEUE_SIZE, default=100000
+)
+LOGS_OTEL_SCHEDULE_DELAY_MILLIS = handle_int_env_var(
+    ENV_ZENML_LOGS_OTEL_SCHEDULE_DELAY_MILLIS, default=5000
+)
+LOGS_OTEL_MAX_EXPORT_BATCH_SIZE = handle_int_env_var(
+    ENV_ZENML_LOGS_OTEL_MAX_EXPORT_BATCH_SIZE, default=5000
+)
+LOGS_OTEL_EXPORT_TIMEOUT_MILLIS = handle_int_env_var(
+    ENV_ZENML_LOGS_OTEL_EXPORT_TIMEOUT_MILLIS, default=15000
+)

src/zenml/deployers/server/service.py

@@ -63,6 +63,7 @@
 from zenml.stack import Stack
 from zenml.steps.utils import get_unique_step_output_names
 from zenml.utils import env_utils, source_utils
+from zenml.utils.logging_utils import setup_run_logging
 from zenml.zen_stores.rest_zen_store import RestZenStore
 
 if TYPE_CHECKING:
@@ -313,6 +314,9 @@ def initialize(self) -> None:
             )
             self._client.zen_store.reinitialize_session()
 
+        # Instantiate the active stack here to avoid race conditions later
+        self._client.active_stack.validate()
+
         # Execution tracking
         self.service_start_time = time.time()
         self.last_execution_time: Optional[datetime] = None
@@ -382,7 +386,7 @@ def execute_pipeline(
             )
 
         except Exception as e:
-            logger.error(f"❌ Pipeline execution failed: {e}")
+            logger.exception("❌ Pipeline execution failed")
             return self._build_response(
                 placeholder_run=placeholder_run,
                 mapped_outputs=None,
@@ -516,16 +520,13 @@ def _prepare_execute_with_orchestrator(
         deployment_snapshot = self._client.zen_store.create_snapshot(
             deployment_snapshot_request
         )
-
         # Create a placeholder run using the new deployment snapshot
         placeholder_run = run_utils.create_placeholder_run(
             snapshot=deployment_snapshot,
-            logs=None,
             trigger_info=PipelineRunTriggerInfo(
                 deployment_id=self.deployment.id,
             ),
         )
-
         return placeholder_run, deployment_snapshot
 
     def _execute_with_orchestrator(
@@ -576,23 +577,29 @@ def _execute_with_orchestrator(
         )
 
         captured_outputs: Optional[Dict[str, Dict[str, Any]]] = None
-        try:
-            # Use the new deployment snapshot with pre-configured settings
-            orchestrator.run(
-                snapshot=deployment_snapshot,
-                stack=active_stack,
-                placeholder_run=placeholder_run,
-            )
+        logging_context = setup_run_logging(
+            pipeline_run=placeholder_run,
+            source="deployment",
+        )
 
-            # Capture in-memory outputs before stopping the runtime context
-            if runtime.is_active():
-                captured_outputs = runtime.get_outputs()
-        except Exception as e:
-            logger.exception(f"Failed to execute pipeline: {e}")
-            raise RuntimeError(f"Failed to execute pipeline: {e}")
-        finally:
-            # Always stop deployment runtime context
-            runtime.stop()
+        with logging_context:
+            try:
+                # Use the new deployment snapshot with pre-configured settings
+                orchestrator.run(
+                    snapshot=deployment_snapshot,
+                    stack=active_stack,
+                    placeholder_run=placeholder_run,
+                )
+
+                # Capture in-memory outputs before stopping the runtime context
+                if runtime.is_active():
+                    captured_outputs = runtime.get_outputs()
+            except Exception as e:
+                logger.exception(f"Failed to execute pipeline: {e}")
+                raise RuntimeError(f"Failed to execute pipeline: {e}")
+            finally:
+                # Always stop deployment runtime context
+                runtime.stop()
 
         return captured_outputs
 

src/zenml/enums.py

@@ -157,6 +157,7 @@ class StackComponentType(StrEnum):
     EXPERIMENT_TRACKER = "experiment_tracker"
     FEATURE_STORE = "feature_store"
     IMAGE_BUILDER = "image_builder"
+    LOG_STORE = "log_store"
     MODEL_DEPLOYER = "model_deployer"
     ORCHESTRATOR = "orchestrator"
     STEP_OPERATOR = "step_operator"

src/zenml/execution/pipeline/dynamic/runner.py

@@ -18,6 +18,7 @@
 import inspect
 import itertools
 from concurrent.futures import ThreadPoolExecutor
+from contextlib import nullcontext
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -53,7 +54,6 @@
 from zenml.execution.pipeline.dynamic.utils import _Unmapped
 from zenml.execution.step.utils import launch_step
 from zenml.logger import get_logger
-from zenml.logging.step_logging import setup_pipeline_logging
 from zenml.models import (
     ArtifactVersionResponse,
     PipelineRunResponse,
@@ -70,6 +70,10 @@
 from zenml.steps.entrypoint_function_utils import StepArtifact
 from zenml.steps.utils import OutputSignature
 from zenml.utils import source_utils
+from zenml.utils.logging_utils import (
+    is_pipeline_logging_enabled,
+    setup_run_logging,
+)
 
 if TYPE_CHECKING:
     from zenml.config import DockerSettings
@@ -153,25 +157,27 @@ def pipeline(self) -> "DynamicPipeline":
 
     def run_pipeline(self) -> None:
         """Run the pipeline."""
-        with setup_pipeline_logging(
-            source="orchestrator",
-            snapshot=self._snapshot,
-        ) as logs_request:
-            if self._run:
-                run = Client().zen_store.update_run(
-                    run_id=self._run.id,
-                    run_update=PipelineRunUpdate(
-                        orchestrator_run_id=self._orchestrator_run_id,
-                        add_logs=[logs_request] if logs_request else None,
-                    ),
-                )
-            else:
-                run = create_placeholder_run(
-                    snapshot=self._snapshot,
+        if self._run:
+            run = Client().zen_store.update_run(
+                run_id=self._run.id,
+                run_update=PipelineRunUpdate(
                     orchestrator_run_id=self._orchestrator_run_id,
-                    logs=logs_request,
-                )
+                ),
+            )
+        else:
+            run = create_placeholder_run(
+                snapshot=self._snapshot,
+                orchestrator_run_id=self._orchestrator_run_id,
+            )
+
+        logging_context = nullcontext()
+        if is_pipeline_logging_enabled(self._snapshot.pipeline_configuration):
+            logging_context = setup_run_logging(
+                pipeline_run=run,
+                source="orchestrator",
+            )
 
+        with logging_context:
             with InMemoryArtifactCache():
                 with DynamicPipelineRunContext(
                     pipeline=self.pipeline,
@@ -181,7 +187,6 @@ def run_pipeline(self) -> None:
                 ):
                     self._orchestrator.run_init_hook(snapshot=self._snapshot)
                     try:
-                        # TODO: step logging isn't threadsafe
                         # TODO: what should be allowed as pipeline returns?
                         #  (artifacts, json serializable, anything?)
                         #  how do we show it in the UI?

src/zenml/integrations/kubernetes/orchestrators/kubernetes_orchestrator_entrypoint.py

@@ -18,6 +18,7 @@
 import socket
 import threading
 import time
+from contextlib import nullcontext
 from typing import List, Optional, Tuple, cast
 from uuid import UUID
 
@@ -56,7 +57,6 @@
     KubernetesOrchestrator,
 )
 from zenml.logger import get_logger
-from zenml.logging.step_logging import setup_orchestrator_logging
 from zenml.models import (
     PipelineRunResponse,
     PipelineRunUpdate,
@@ -74,6 +74,10 @@
 )
 from zenml.pipelines.run_utils import create_placeholder_run
 from zenml.utils import env_utils
+from zenml.utils.logging_utils import (
+    is_pipeline_logging_enabled,
+    setup_run_logging,
+)
 
 logger = get_logger(__name__)
 
@@ -244,7 +248,6 @@ def main() -> None:
         namespace=namespace,
         job_name=job_name,
     )
-    existing_logs_response = None
 
     if run_id and orchestrator_run_id:
         logger.info("Continuing existing run `%s`.", run_id)
@@ -257,13 +260,9 @@ def main() -> None:
         )
         logger.debug("Reconstructed nodes: %s", nodes)
 
-        # Continue logging to the same log file if it exists
-        for log_response in pipeline_run.log_collection or []:
-            if log_response.source == "orchestrator":
-                existing_logs_response = log_response
-                break
     else:
         orchestrator_run_id = orchestrator_pod_name
+
         if args.run_id:
             pipeline_run = client.zen_store.update_run(
                 run_id=args.run_id,
@@ -293,11 +292,12 @@ def main() -> None:
             for step_name, step in snapshot.step_configurations.items()
         ]
 
-    logs_context = setup_orchestrator_logging(
-        run_id=pipeline_run.id,
-        snapshot=snapshot,
-        logs_response=existing_logs_response,
-    )
+    logs_context = nullcontext()
+    if is_pipeline_logging_enabled(snapshot.pipeline_configuration):
+        logs_context = setup_run_logging(
+            pipeline_run=pipeline_run,
+            source="orchestrator",
+        )
 
     with logs_context:
         step_command = StepEntrypointConfiguration.get_entrypoint_command()