Add telemetry pipeline run ends

docs/book/user-guide/starter-guide/fetching-pipelines.md

@@ -101,6 +101,16 @@ Calling a pipeline executes it and then returns the response of the freshly exec
 last_run = training_pipeline()
 ```
 
+{% hint style="warning" %}
+The run that you get back is the model stored in the ZenML database at the point of the method call. This means the pipeline run is still initializing and no steps have been run. To get the latest state can get a refreshed version from the client:
+
+```python
+from zenml.client import Client
+
+Client().get_pipeline_run(last_run.id) to get a refreshed version
+```
+{% endhint %}
+
 ### Get a run via the client
 
 If you already know the exact run that you want to fetch (e.g., from looking at the dashboard), you can use the [`Client.get_pipeline_run()`](https://sdkdocs.zenml.io/latest/core\_code\_docs/core-client/#zenml.client.Client.get\_pipeline\_run) method to fetch the run directly without having to query the pipeline first:

examples/e2e/steps/training/model_trainer.py

@@ -101,12 +101,12 @@ def model_trainer(
     # keep track of mlflow version for future use
     model_registry = Client().active_stack.model_registry
     if model_registry:
-        versions = model_registry.list_model_versions(name=name)
-        if versions:
+        version = model_registry.get_latest_model_version(
+            name=name, stage=None
+        )
+        if version:
             model_ = get_step_context().model
-            model_.log_metadata(
-                {"model_registry_version": versions[-1].version}
-            )
+            model_.log_metadata({"model_registry_version": version.version})
     ### YOUR CODE ENDS HERE ###
 
     return model

examples/e2e/utils/promote_in_model_registry.py

@@ -35,6 +35,7 @@ def promote_in_model_registry(
         target_env: stage for promotion
     """
     model_registry = Client().active_stack.model_registry
+    model_registry.configure_mlflow()
     if latest_version != current_version:
         model_registry.update_model_version(
             name=model_name,

src/zenml/analytics/enums.py

@@ -20,6 +20,7 @@ class AnalyticsEvent(str, Enum):
 
     # Pipelines
     RUN_PIPELINE = "Pipeline run"
+    RUN_PIPELINE_ENDED = "Pipeline run ended"
     CREATE_PIPELINE = "Pipeline created"
     BUILD_PIPELINE = "Pipeline built"
 

src/zenml/new/pipelines/pipeline.py

@@ -734,14 +734,12 @@ def _run(
                 deployment=deployment_request
             )
 
-            analytics_handler.metadata = self._get_pipeline_analytics_metadata(
-                deployment=deployment_model, stack=stack
-            )
             stack.prepare_pipeline_deployment(deployment=deployment_model)
 
             self.log_pipeline_deployment_metadata(deployment_model)
 
             run = None
+            run_id = None
             if not schedule:
                 run_request = PipelineRunRequest(
                     name=get_run_name(
@@ -764,7 +762,11 @@ def _run(
                     status=ExecutionStatus.INITIALIZING,
                 )
                 run = Client().zen_store.create_run(run_request)
+                run_id = run.id
 
+            analytics_handler.metadata = self._get_pipeline_analytics_metadata(
+                deployment=deployment_model, stack=stack, run_id=run_id
+            )
             # Prevent execution of nested pipelines which might lead to
             # unexpected behavior
             constants.SHOULD_PREVENT_PIPELINE_EXECUTION = True
@@ -1081,12 +1083,14 @@ def _get_pipeline_analytics_metadata(
         self,
         deployment: "PipelineDeploymentResponse",
         stack: "Stack",
+        run_id: Optional[UUID] = None,
     ) -> Dict[str, Any]:
         """Returns the pipeline deployment metadata.
 
         Args:
             deployment: The pipeline deployment to track.
             stack: The stack on which the pipeline will be deployed.
+            run_id: The ID of the pipeline run.
 
         Returns:
             the metadata about the pipeline deployment
@@ -1113,6 +1117,7 @@ def _get_pipeline_analytics_metadata(
             "schedule": bool(deployment.schedule),
             "custom_materializer": custom_materializer,
             "own_stack": own_stack,
+            "pipeline_run_id": str(run_id) if run_id else None,
         }
 
     def _compile(

src/zenml/zen_stores/sql_zen_store.py

@@ -61,7 +61,11 @@
 from sqlmodel.sql.expression import Select, SelectOfScalar
 
 from zenml.analytics.enums import AnalyticsEvent
-from zenml.analytics.utils import analytics_disabler, track_decorator
+from zenml.analytics.utils import (
+    analytics_disabler,
+    track_decorator,
+    track_handler,
+)
 from zenml.config.global_config import GlobalConfiguration
 from zenml.config.secrets_store_config import SecretsStoreConfiguration
 from zenml.config.server_config import ServerConfiguration
@@ -6688,7 +6692,20 @@ def _update_pipeline_run_status(
                 ExecutionStatus.FAILED,
             }:
                 run_update.end_time = datetime.utcnow()
-
+                with track_handler(
+                    AnalyticsEvent.RUN_PIPELINE_ENDED
+                ) as analytics_handler:
+                    analytics_handler.metadata = {
+                        "pipeline_run_id": pipeline_run_id,
+                        "status": new_status,
+                        "num_steps": num_steps,
+                        "start_time": pipeline_run.start_time.strftime(
+                            "%Y-%m-%dT%H:%M:%S.%fZ"
+                        ),
+                        "end_time": run_update.end_time.strftime(
+                            "%Y-%m-%dT%H:%M:%S.%fZ"
+                        ),
+                    }
             pipeline_run.update(run_update)
             session.add(pipeline_run)