Store client-side pipeline run logs in the artifact store (#3498)

* Store client-side pipeline logs in the artifact store

* Update endpoint to return pipeline logs

* Add project scope to pipeline logs endpoint

* Add pipeline log settings and pipeline run logs endpoint

* Fix linter errors

* Use enable_pipeline_logs instead of enable_step_logs

* Apply code review suggestions and add docs

* Optimised images with calibre/image-actions

* Fix linter errors

* Use the enable_pipeline_logs in the pipeline configuration

* Fix deadlock bug and remaining unit tests

* Remove redundant Kubernetes orchestrator logs

---------

Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>

Author: stefannica

docs/book/.gitbook/assets/zenml_pipeline_run_logs.png

@@ -1,6 +1,11 @@
 # Enable or disable logs storing
 
-By default, ZenML uses a logging handler to capture the logs that occur during the execution of a step. Users are free to use the default python logging module or print statements, and ZenML's logging handler will catch these logs and store them. 
+By default, ZenML uses a logging handler to capture two types of logs:
+
+* the logs collected from your ZenML client while triggering and waiting for a pipeline to run. These logs cover everything that happens client-side: building and pushing container images, triggering the pipeline, waiting for it to start, and waiting for it to finish. Note that these logs are not available for scheduled pipelines and the logs might not be available for pipeline runs that fail while building or pushing the container images.
+* the logs collected from the execution of a step. These logs only cover what happens during the execution of a single step and originate mostly from the user-provided step code and the libraries it calls.
+
+For step logs, users are free to use the default python logging module or print statements, and ZenML's logging handler will catch these logs and store them.
 
 ```python
 import logging
@@ -13,16 +18,38 @@ def my_step() -> None:
     print("World.")  # You can utilize `print` statements as well. 
 ```
 
-These logs are stored within the respective artifact store of your stack. You can display the logs in the dashboard as follows:
+All these logs are stored within the respective artifact store of your stack. You can visualize the pipeline run logs and step logs in the dashboard as follows:
 
+![Displaying pipeline run logs on the dashboard](../../.gitbook/assets/zenml_pipeline_run_logs.png)
 ![Displaying step logs on the dashboard](../../.gitbook/assets/zenml_step_logs.png)
 
 {% hint style="warning" %}
 Note that if you are not connected to a cloud artifact store with a service connector configured then you will not
 be able to view your logs in the dashboard. Read more [here](./view-logs-on-the-dasbhoard.md).
 {% endhint %}
 
-If you do not want to store the logs in your artifact store, you can:
+If you do not want to store the logs in your artifact store, you can do the following:
+
+For pipeline run logs:
+
+1.  Disable it by using the `enable_pipeline_logs` parameter with your `@pipeline` decorator:
+
+    ```python
+    from zenml import pipeline, step
+
+    @pipeline(enable_pipeline_logs=False)  # disables logging for pipeline
+    def my_pipeline():
+        ...
+    ```
+2. Disable it by using the environmental variable `ZENML_DISABLE_PIPELINE_LOGS_STORAGE` and setting it to `true`. This environmental variable takes precedence over the parameter mentioned above. Note this environmental variable needs to be set on the [execution environment](../pipeline-development/configure-python-environments/README.md#execution-environments), i.e., on the orchestrator level:
+
+```shell
+export ZENML_DISABLE_PIPELINE_LOGS_STORAGE=true
+
+python my_pipeline.py
+```
+
+For step logs:
 
 1.  Disable it by using the `enable_step_logs` parameter either with your `@pipeline` or `@step` decorator:
 

docs/book/how-to/control-logging/enable-or-disable-logs-storing.md

@@ -210,6 +210,7 @@ def _apply_run_configuration(
                 enable_artifact_metadata=config.enable_artifact_metadata,
                 enable_artifact_visualization=config.enable_artifact_visualization,
                 enable_step_logs=config.enable_step_logs,
+                enable_pipeline_logs=config.enable_pipeline_logs,
                 settings=config.settings,
                 tags=config.tags,
                 extra=config.extra,

src/zenml/config/compiler.py

@@ -41,6 +41,7 @@ class PipelineConfigurationUpdate(StrictBaseModel):
     enable_artifact_metadata: Optional[bool] = None
     enable_artifact_visualization: Optional[bool] = None
     enable_step_logs: Optional[bool] = None
+    enable_pipeline_logs: Optional[bool] = None
     settings: Dict[str, SerializeAsAny[BaseSettings]] = {}
     tags: Optional[List[Union[str, "Tag"]]] = None
     extra: Dict[str, Any] = {}

src/zenml/config/pipeline_configurations.py

@@ -40,6 +40,7 @@ class PipelineRunConfiguration(
     enable_artifact_metadata: Optional[bool] = None
     enable_artifact_visualization: Optional[bool] = None
     enable_step_logs: Optional[bool] = None
+    enable_pipeline_logs: Optional[bool] = None
     schedule: Optional[Schedule] = None
     build: Union[PipelineBuildBase, UUID, None] = Field(
         default=None, union_mode="left_to_right"

src/zenml/config/pipeline_run_configuration.py

@@ -168,6 +168,7 @@ def handle_int_env_var(var: str, default: int = 0) -> int:
 ENV_ZENML_SERVER = "ZENML_SERVER"
 ENV_ZENML_ENFORCE_TYPE_ANNOTATIONS = "ZENML_ENFORCE_TYPE_ANNOTATIONS"
 ENV_ZENML_ENABLE_IMPLICIT_AUTH_METHODS = "ZENML_ENABLE_IMPLICIT_AUTH_METHODS"
+ENV_ZENML_DISABLE_PIPELINE_LOGS_STORAGE = "ZENML_DISABLE_PIPELINE_LOGS_STORAGE"
 ENV_ZENML_DISABLE_STEP_LOGS_STORAGE = "ZENML_DISABLE_STEP_LOGS_STORAGE"
 ENV_ZENML_DISABLE_STEP_NAMES_IN_LOGS = "ZENML_DISABLE_STEP_NAMES_IN_LOGS"
 ENV_ZENML_IGNORE_FAILURE_HOOK = "ZENML_IGNORE_FAILURE_HOOK"

src/zenml/constants.py

@@ -543,7 +543,6 @@ def prepare_or_run_pipeline(
                 mount_local_stores=self.config.is_local,
             )
 
-            logger.info("Waiting for Kubernetes orchestrator pod to start...")
             kube_utils.create_and_wait_for_pod_to_start(
                 core_api=self._k8s_core_api,
                 pod_display_name="Kubernetes orchestrator pod",

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

@@ -173,7 +173,6 @@ def run_step_on_kubernetes(step_name: str) -> None:
             mount_local_stores=mount_local_stores,
         )
 
-        logger.info(f"Waiting for pod of step `{step_name}` to start...")
         kube_utils.create_and_wait_for_pod_to_start(
             core_api=core_api,
             pod_display_name=f"pod for step `{step_name}`",

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

@@ -218,9 +218,6 @@ def launch(
             mount_local_stores=False,
         )
 
-        logger.info(
-            "Waiting for pod of step `%s` to start...", info.pipeline_step_name
-        )
         kube_utils.create_and_wait_for_pod_to_start(
             core_api=self._k8s_core_api,
             pod_display_name=f"pod of step `{info.pipeline_step_name}`",

src/zenml/integrations/kubernetes/step_operators/kubernetes_step_operator.py

@@ -13,6 +13,7 @@
 #  permissions and limitations under the License.
 """ZenML logging handler."""
 
+import logging
 import os
 import re
 import sys
@@ -48,6 +49,7 @@
 redirected: ContextVar[bool] = ContextVar("redirected", default=False)
 
 LOGS_EXTENSION = ".log"
+PIPELINE_RUN_LOGS_FOLDER = "pipeline_runs"
 
 
 def remove_ansi_escape_codes(text: str) -> str:
@@ -65,14 +67,14 @@ def remove_ansi_escape_codes(text: str) -> str:
 
 def prepare_logs_uri(
     artifact_store: "BaseArtifactStore",
-    step_name: str,
+    step_name: Optional[str] = None,
     log_key: Optional[str] = None,
 ) -> str:
     """Generates and prepares a URI for the log file or folder for a step.
 
     Args:
         artifact_store: The artifact store on which the artifact will be stored.
-        step_name: Name of the step.
+        step_name: Name of the step. Skipped for global pipeline run logs.
         log_key: The unique identification key of the log file.
 
     Returns:
@@ -81,11 +83,8 @@ def prepare_logs_uri(
     if log_key is None:
         log_key = str(uuid4())
 
-    logs_base_uri = os.path.join(
-        artifact_store.path,
-        step_name,
-        "logs",
-    )
+    subfolder = step_name or PIPELINE_RUN_LOGS_FOLDER
+    logs_base_uri = os.path.join(artifact_store.path, subfolder, "logs")
 
     # Create the dir
     if not artifact_store.exists(logs_base_uri):
@@ -210,7 +209,7 @@ def _read_file(
         artifact_store.cleanup()
 
 
-class StepLogsStorage:
+class PipelineLogsStorage:
     """Helper class which buffers and stores logs to a given URI."""
 
     def __init__(
@@ -324,6 +323,18 @@ def save_to_file(self, force: bool = False) -> None:
             self.disabled = True
 
             try:
+                # The configured logging handler uses a lock to ensure that
+                # logs generated by different threads are not interleaved.
+                # Given that most artifact stores are based on fsspec, which
+                # use a separate thread for async operations, it may happen that
+                # the fsspec library itself will log something, which will end
+                # up in a deadlock.
+                # To avoid this, we temporarily disable the lock in the logging
+                # handler while writing to the file.
+                logging_handler = logging.getLogger().handlers[0]
+                logging_lock = logging_handler.lock
+                logging_handler.lock = None
+
                 if self.buffer:
                     if self.artifact_store.config.IS_IMMUTABLE_FILESYSTEM:
                         _logs_uri = self._get_timestamped_filename()
@@ -353,6 +364,9 @@ def save_to_file(self, force: bool = False) -> None:
                 # I/O errors.
                 logger.error(f"Error while trying to write logs: {e}")
             finally:
+                # Restore the original logging handler lock
+                logging_handler.lock = logging_lock
+
                 self.buffer = []
                 self.last_save_time = time.time()
 
@@ -418,8 +432,8 @@ def merge_log_files(self, merge_all_files: bool = False) -> None:
                         )
 
 
-class StepLogsStorageContext:
-    """Context manager which patches stdout and stderr during step execution."""
+class PipelineLogsStorageContext:
+    """Context manager which patches stdout and stderr during pipeline run execution."""
 
     def __init__(
         self, logs_uri: str, artifact_store: "BaseArtifactStore"
@@ -428,17 +442,17 @@ def __init__(
 
         Args:
             logs_uri: the URI of the logs file.
-            artifact_store: Artifact Store from the current step context.
+            artifact_store: Artifact Store from the current pipeline run context.
         """
-        self.storage = StepLogsStorage(
+        self.storage = PipelineLogsStorage(
             logs_uri=logs_uri, artifact_store=artifact_store
         )
 
-    def __enter__(self) -> "StepLogsStorageContext":
+    def __enter__(self) -> "PipelineLogsStorageContext":
         """Enter condition of the context manager.
 
         Wraps the `write` method of both stderr and stdout, so each incoming
-        message gets stored in the step logs storage.
+        message gets stored in the pipeline logs storage.
 
         Returns:
             self