Running multiple functions with ExitHandler#

This example uses the Kubernetes function ExitHandler to run multiple functions before continuing to the next function. The ExitHandler function is especially useful to do something that needs to be performed as the very last step; for example, sending an email or updating the run results of the previous steps somewhere else.

import mlrun

Create the project and its functions#

Function D is designed to fail, to show the ExitHandler running after a failure.

project = mlrun.get_or_create_project(
    "waiting-for-multiple-steps", "./", user_project=True
> 2024-04-16 09:28:55,586 [info] Project loaded successfully: {'project_name': 'waiting-for-multiple-steps'}
import time

# First 3 functions to run
def func_A(context):
    time.sleep(5)"Function A is running now")
    return "Function A has been triggered"

def func_B(context):
    time.sleep(10)"Function B is running now")        
    return "Function B has been triggered"

def func_C(context):
    time.sleep(15)"Function C is running now")        
    return "Function C has been triggered"

# This function waits for the 3 functions to complete, then logs their results before raising an exception.
def func_D(context, func_a_res, func_b_res, func_c_res):"Function D is running now, logging the results of previous functions.")
    context.log_artifact("func_a_result", str(func_a_res))
    context.log_artifact("func_b_result", str(func_b_res))
    context.log_artifact("func_c_result", str(func_c_res))"Function D has been triggered")
    # Now it raises an Exception so the function fails.
    raise Exception("This is an example exception")

# This function will be part of the ExitHandler.
def func_final(context):"The final function is now running regardless of whether all preceding functions succeeded.")
    return "Function final has been triggered"
# Set the functions in the project
func_A = project.set_function(
    func="", name="func-A", handler="func_A", image="mlrun/mlrun", kind="job"
func_B = project.set_function(
    func="", name="func-B", handler="func_B", image="mlrun/mlrun", kind="job"
func_C = project.set_function(
    func="", name="func-C", handler="func_C", image="mlrun/mlrun", kind="job"
func_D = project.set_function(
    func="", name="func-D", handler="func_D", image="mlrun/mlrun", kind="job"
func_final = project.set_function(

Create the pipeline#

The ExitHandler function runs after the functions, A, B, C, D. It is triggered regardless of whether or not all of the preceding functions succeeded.


from kfp import dsl
from mlrun.platforms import auto_mount
import os
import sys
import mlrun

@dsl.pipeline(name="ExitHandler and multiple wait pipeline", description="Pipeline that runs 3 functions simultaneously, waits for all of them to finish, and then runs another function that logs their results and fails. After all of this, an exit function is triggered.")
def kfpipeline(input_val):
    project = mlrun.get_current_project()
    # 'func_final' is executed after everything inside the block finishes or crashes.
    with dsl.ExitHandler(mlrun.run_function("func-final")):
        # Start 3 functions simultaneously
        step_1 = mlrun.run_function('func-A', returns=['first_func_res'])
        step_2 = mlrun.run_function('func-B', returns=['second_func_res'])
        step_3 = mlrun.run_function('func-C', returns=['third_func_res'])

        # Start the function only after the 3 first functions are done. This function logs the outputs of the previous functions as artifacts, and then crashes.
        step_4 = mlrun.run_function('func-D', params = {"func_a_res":step_1.outputs["first_func_res"],
                                                        "func_c_res":step_3.outputs["third_func_res"]}, returns=["fourth_func_res"]).after(step_1, step_2, step_3)  
project.set_workflow(name="workflow-func", workflow_path="")
<mlrun.projects.project.MlrunProject at 0x7f1fe10d5250>

Run the workflow#

Now run the workflow and check the workflow graph in the UI."", watch=True, local=False)
Pipeline running (id=2cb4810e-78bc-4722-9f45-6f8b33144df6), click here to view the details in MLRun UI

Run Results

[info] Workflow 2cb4810e-78bc-4722-9f45-6f8b33144df6 finished with 1 errors

click the hyper links below to see detailed results
uid start state name parameters results
Apr 16 09:35:45 completed func-final
return=Function final has been triggered
Apr 16 09:35:18
func_a_res=Function A has been triggered
func_b_res=Function B has been triggered
func_c_res=Function C has been triggered
Apr 16 09:34:39 completed func-b
second_func_res=Function B has been triggered
Apr 16 09:34:39 completed func-c
third_func_res=Function C has been triggered
Apr 16 09:34:39 completed func-a
first_func_res=Function A has been triggered
RuntimeError                              Traceback (most recent call last)
Cell In[16], line 1
----> 1'',watch=True, local=False)

File /conda/envs/mlrun-extended/lib/python3.9/site-packages/mlrun/projects/, in, name, workflow_path, arguments, artifact_path, workflow_handler, namespace, sync, watch, dirty, engine, local, schedule, timeout, source, cleanup_ttl, notifications)
   2716     if engine == "remote" and status_engine.engine != "local":
   2717         status_engine = _RemoteRunner
-> 2719     status_engine.get_run_status(project=self, run=run, timeout=timeout)
   2720 return run

File /conda/envs/mlrun-extended/lib/python3.9/site-packages/mlrun/projects/, in _KFPRunner.get_run_status(project, run, timeout, expected_statuses, notifiers)
    697 notifiers.push(text, "info", runs)
    699 if raise_error:
--> 700     raise raise_error
    701 return state, had_errors, text

File /conda/envs/mlrun-extended/lib/python3.9/site-packages/mlrun/projects/, in _KFPRunner.get_run_status(project, run, timeout, expected_statuses, notifiers)
    672     if timeout:
    673"Waiting for pipeline run completion")
--> 674         state = run.wait_for_completion(
    675             timeout=timeout, expected_statuses=expected_statuses
    676         )
    677 except RuntimeError as exc:
    678     # push runs table also when we have errors
    679     raise_error = exc

File /conda/envs/mlrun-extended/lib/python3.9/site-packages/mlrun/projects/, in _PipelineRunStatus.wait_for_completion(self, timeout, expected_statuses)
    459 def wait_for_completion(self, timeout=None, expected_statuses=None):
--> 460     self._state = self._engine.wait_for_completion(
    461         self.run_id,
    462         project=self.project,
    463         timeout=timeout,
    464         expected_statuses=expected_statuses,
    465     )
    466     return self._state

File /conda/envs/mlrun-extended/lib/python3.9/site-packages/mlrun/projects/, in _KFPRunner.wait_for_completion(run_id, project, timeout, expected_statuses)
    638     timeout = 60 * 60
    639 project_name = if project else ""
--> 640 run_info = wait_for_pipeline_completion(
    641     run_id,
    642     timeout=timeout,
    643     expected_statuses=expected_statuses,
    644     project=project_name,
    645 )
    646 status = ""
    647 if run_info:

File /conda/envs/mlrun-extended/lib/python3.9/site-packages/mlrun/, in wait_for_pipeline_completion(run_id, timeout, expected_statuses, namespace, remote, project)
    969 if expected_statuses:
    970     if status not in expected_statuses:
--> 971         raise RuntimeError(
    972             f"Pipeline run status {status}{', ' + message if message else ''}"
    973         )
    975 logger.debug(
    976     f"Finished waiting for pipeline completion."
    977     f" run_id: {run_id},"
    980     f" namespace: {namespace}"
    981 )
    983 return resp

RuntimeError: Pipeline run status Failed

Viewing the results in the UI Monitor Workflows tab#