[KFP] Obtain the Ray cluster run ID from the user for KFP v2.

kfp/doc/simple_transform_pipeline.md

@@ -57,11 +57,16 @@ Ray cluster. For each step we have to define a component that will execute them:
 ```python
     # components
     base_kfp_image = "quay.io/dataprep1/data-prep-kit/kfp-data-processing:0.0.2"
-    # compute execution parameters. Here different transforms might need different implementations. As
-    # a result, instead of creating a component we are creating it in place here.
-    compute_exec_params_op = comp.func_to_container_op(
-      func=ComponentUtils.default_compute_execution_params, base_image=base_kfp_image
-    )
+    # KFPv1 and KFP2 uses different methods to create a component from a function. KFPv1 uses the
+    # `create_component_from_func` function, but it is deprecated by KFPv2 and so has a different import path.
+    # KFPv2 recommends using the `@dsl.component` decorator, which doesn't exist in KFPv1. Therefore, here we use
+    # this if/else statement and explicitly call the decorator.
+    if os.getenv("KFPv2", "0") == "1":
+      compute_exec_params_op = dsl.component_decorator.component(
+                func=compute_exec_params_func, base_image=base_kfp_image
+                )
+    else:
+      compute_exec_params_op = comp.create_component_from_func(func=compute_exec_params_func, base_image=base_kfp_image)
     # create Ray cluster
     create_ray_op = comp.load_component_from_file("../../../kfp_ray_components/createRayComponent.yaml")
     # execute job
@@ -107,6 +112,7 @@ The input parameters section defines all the parameters required for the pipelin
 The parameters used here are as follows:
 
 * ray_name: name of the Ray cluster
+* ray_run_id_KFPv2: Ray cluster unique ID used only in KFP v2
 * ray_head_options: head node options, containing the following:
   * cpu - number of cpus
   * memory - memory
@@ -148,8 +154,18 @@ Now, when all components and input parameters are defined, we can implement pipe
 component execution and parameters submitted to every component. 
 
 ```python
+    # In KFPv2 dsl.RUN_ID_PLACEHOLDER is deprecated and cannot be used since SDK 2.5.0. On another hand we cannot create
+    # a unique string in a component (at runtime) and pass it to the `clean_up_task` of `ExitHandler`, due to
+    # https://github.com/kubeflow/pipelines/issues/10187. Therefore, meantime the user is requested to insert
+    # a unique string created at run creation time.
+    if os.getenv("KFPv2", "0") == "1":
+        print("WARNING: the ray cluster name can be non-unique at runtime, please do not execute simultaneous Runs of the "
+              "same version of the same pipeline !!!")
+        run_id = ray_run_id_KFPv2
+    else:
+        run_id = dsl.RUN_ID_PLACEHOLDER
     # create clean_up task
-    clean_up_task = cleanup_ray_op(ray_name=ray_name, run_id=dsl.RUN_ID_PLACEHOLDER, server_url=server_url, additional_params=additional_params)
+    clean_up_task = cleanup_ray_op(ray_name=ray_name, run_id=run_id, server_url=server_url, additional_params=additional_params)
     ComponentUtils.add_settings_to_component(clean_up_task, ONE_HOUR_SEC * 2)
     # pipeline definition
     with dsl.ExitHandler(clean_up_task):
@@ -162,7 +178,7 @@ component execution and parameters submitted to every component.
       # start Ray cluster
       ray_cluster = create_ray_op(
         ray_name=ray_name,
-        run_id=dsl.RUN_ID_PLACEHOLDER,
+        run_id=run_id,
         ray_head_options=ray_head_options,
         ray_worker_options=ray_worker_options,
         server_url=server_url,
@@ -173,7 +189,7 @@ component execution and parameters submitted to every component.
       # Execute job
       execute_job = execute_ray_jobs_op(
         ray_name=ray_name,
-        run_id=dsl.RUN_ID_PLACEHOLDER,
+        run_id=run_id,
         additional_params=additional_params,
         # note that the parameters below are specific for NOOP transform
         exec_params={
@@ -183,7 +199,7 @@ component execution and parameters submitted to every component.
           "num_workers": compute_exec_params.output,
           "worker_options": actor_options,
           "pipeline_id": pipeline_id,
-          "job_id": dsl.RUN_ID_PLACEHOLDER,
+          "job_id": run_id,
           "code_location": code_location,
           "noop_sleep_sec": noop_sleep_sec,
         },

kfp/kfp_ray_components/Dockerfile

@@ -30,7 +30,7 @@ RUN pip install --no-cache-dir pydantic==2.6.3
 # remove credentials-containing file
 RUN rm requirements.txt
 # components
-COPY ./src /pipelines/component/src
+COPY --chmod=775 --chown=ray:root ./src /pipelines/component/src
 
 # Set environment
 ENV KFP_v2=$KFP_v2

kfp/kfp_support_lib/kfp_v2_workflow_support/src/workflow_support/pipeline_utils/pipelines_tests_utils.py

@@ -52,6 +52,27 @@ def run_test(pipeline_package_path: str, endpoint: str = "http://localhost:8080/
     logger.info(f"Pipeline {pipeline_name} successfully completed")
     return pipeline_name
 
+def _set_run_id(pipeline_package_path: str):
+    """
+    Assign a dummy run ID value for testing purposes. By default, this value
+    is empty and is set by the user during runtime.
+
+    :param pipeline_package_path: Local path to the pipeline package.
+    """
+    import yaml
+    import uuid
+
+    try:
+        stream = open(pipeline_package_path, "r")
+        docs = list(yaml.load_all(stream, yaml.FullLoader))
+        for doc in docs:
+            if "root" in doc:
+                doc["root"]["inputDefinitions"]["parameters"]["ray_run_id_KFPv2"]["defaultValue"] = uuid.uuid4().hex
+        with open(pipeline_package_path, "w") as outfile:
+            yaml.dump_all(docs, outfile)
+    except Exception as e:
+        logger.error(f"Failed to update run id value, exception {e}")
+        sys.exit(1)
 
 if __name__ == "__main__":
     import argparse
@@ -74,6 +95,7 @@ def run_test(pipeline_package_path: str, endpoint: str = "http://localhost:8080/
             if pipeline is None:
                 sys.exit(1)
         case "sanity-test":
+            _set_run_id(args.pipeline_package_path)
             run = run_test(
                 endpoint=args.endpoint,
                 pipeline_package_path=args.pipeline_package_path,

kfp/kfp_support_lib/shared_workflow_support/src/runtime_utils/kfp_utils.py

@@ -81,9 +81,10 @@ def runtime_name(ray_name: str = "", run_id: str = "") -> str:
         # the return value plus namespace name will be the name of the Ray Route,
         # which length is restricted to 64 characters,
         # therefore we restrict the return name by 15 character.
-        if run_id != "":
-            return f"{ray_name[:9]}-{run_id[:5]}"
-        return ray_name[:15]
+        if run_id == "":
+            logger.error("Run ID must not be provided")
+            sys.exit(1)
+        return f"{ray_name[:9]}-{run_id[:5]}"
 
     @staticmethod
     def dict_to_req(d: dict[str, Any], executor: str = "transformer_launcher.py") -> str:

kfp/pipeline_generator/single-pipeline/templates/simple_pipeline.py

@@ -73,23 +73,11 @@ def compute_exec_params_func(
 # KFPv2 recommends using the `@dsl.component` decorator, which doesn't exist in KFPv1. Therefore, here we use
 # this if/else statement and explicitly call the decorator.
 if os.getenv("KFPv2", "0") == "1":
-    # In KFPv2 dsl.RUN_ID_PLACEHOLDER is deprecated and cannot be used since SDK 2.5.0. On another hand we cannot create
-    # a unique string in a component (at runtime) and pass it to the `clean_up_task` of `ExitHandler`, due to
-    # https://github.com/kubeflow/pipelines/issues/10187. Therefore, meantime we use a unique string created at
-    # compilation time.
-    import uuid
-
     compute_exec_params_op = dsl.component_decorator.component(
         func=compute_exec_params_func, base_image=base_kfp_image
     )
-    print(
-        "WARNING: the ray cluster name can be non-unique at runtime, please do not execute simultaneous Runs of the "
-        + "same version of the same pipeline !!!"
-    )
-    run_id = uuid.uuid4().hex
 else:
     compute_exec_params_op = comp.create_component_from_func(func=compute_exec_params_func, base_image=base_kfp_image)
-    run_id = dsl.RUN_ID_PLACEHOLDER
 
 # create Ray cluster
 create_ray_op = comp.load_component_from_file(component_spec_path + "createRayClusterComponent.yaml")
@@ -111,9 +99,11 @@ def {{ pipeline_name }}(
     ray_name: str = "{{ pipeline_name }}-kfp-ray",  # name of Ray cluster
     # Add image_pull_secret and image_pull_policy to ray workers if needed
     {%- if image_pull_secret != "" %}
+    ray_run_id_KFPv2: str = "",
     ray_head_options: dict = {"cpu": 1, "memory": 4, "image_pull_secret": "{{ image_pull_secret }}", "image": task_image},
     ray_worker_options: dict = {"replicas": 2, "max_replicas": 2, "min_replicas": 2, "cpu": 2, "memory": 4, "image_pull_secret": "{{ image_pull_secret }}", "image": task_image},
     {%- else %}
+    ray_run_id_KFPv2: str = "",
     ray_head_options: dict = {"cpu": 1, "memory": 4, "image": task_image},
     ray_worker_options: dict = {"replicas": 2, "max_replicas": 2, "min_replicas": 2, "cpu": 2, "memory": 4, "image": task_image},
     {%- endif %}
@@ -142,6 +132,7 @@ def {{ pipeline_name }}(
     """
     Pipeline to execute {{ pipeline_name }} transform
     :param ray_name: name of the Ray cluster
+    :param ray_run_id_KFPv2: string holding the id used for the Ray cluster used only in KFP v2
     :param ray_head_options: head node options, containing the following:
         cpu - number of cpus
         memory - memory
@@ -177,6 +168,16 @@ def {{ pipeline_name }}(
     {%- endfor %}
     :return: None
     """
+    # In KFPv2 dsl.RUN_ID_PLACEHOLDER is deprecated and cannot be used since SDK 2.5.0. On another hand we cannot create
+    # a unique string in a component (at runtime) and pass it to the `clean_up_task` of `ExitHandler`, due to
+    # https://github.com/kubeflow/pipelines/issues/10187. Therefore, meantime the user is requested to insert
+    # a unique string created at run creation time.
+    if os.getenv("KFPv2", "0") == "1":
+        print("WARNING: the ray cluster name can be non-unique at runtime, please do not execute simultaneous Runs of the "
+              "same version of the same pipeline !!!")
+        run_id = ray_run_id_KFPv2
+    else:
+        run_id = dsl.RUN_ID_PLACEHOLDER
     # create clean_up task
     clean_up_task = cleanup_ray_op(ray_name=ray_name, run_id=run_id, server_url=server_url, additional_params=additional_params)
     ComponentUtils.add_settings_to_component(clean_up_task, ONE_HOUR_SEC * 2)

transforms/code/code2parquet/kfp_ray/code2parquet_wf.py

@@ -77,23 +77,11 @@ def compute_exec_params_func(
 # KFPv2 recommends using the `@dsl.component` decorator, which doesn't exist in KFPv1. Therefore, here we use
 # this if/else statement and explicitly call the decorator.
 if os.getenv("KFPv2", "0") == "1":
-    # In KFPv2 dsl.RUN_ID_PLACEHOLDER is deprecated and cannot be used since SDK 2.5.0. On another hand we cannot create
-    # a unique string in a component (at runtime) and pass it to the `clean_up_task` of `ExitHandler`, due to
-    # https://github.com/kubeflow/pipelines/issues/10187. Therefore, meantime we use a unique string created at
-    # compilation time.
-    import uuid
-
     compute_exec_params_op = dsl.component_decorator.component(
         func=compute_exec_params_func, base_image=base_kfp_image
     )
-    print(
-        "WARNING: the ray cluster name can be non-unique at runtime, please do not execute simultaneous Runs of the "
-        + "same version of the same pipeline !!!"
-    )
-    run_id = uuid.uuid4().hex
 else:
     compute_exec_params_op = comp.create_component_from_func(func=compute_exec_params_func, base_image=base_kfp_image)
-    run_id = dsl.RUN_ID_PLACEHOLDER
 
 
 # create Ray cluster
@@ -113,6 +101,7 @@ def compute_exec_params_func(
 )
 def code2parquet(
     ray_name: str = "code2parquet-kfp-ray",  # name of Ray cluster
+    ray_run_id_KFPv2: str = "",   # Ray cluster unique ID used only in KFP v2
     # Add image_pull_secret and image_pull_policy to ray workers if needed
     ray_head_options: dict = {"cpu": 1, "memory": 4, "image": task_image},
     ray_worker_options: dict = {"replicas": 2, "max_replicas": 2, "min_replicas": 2, "cpu": 2, "memory": 4, "image": task_image},
@@ -139,6 +128,7 @@ def code2parquet(
     """
     Pipeline to execute NOOP transform
     :param ray_name: name of the Ray cluster
+    :param ray_run_id_KFPv2: string holding the id used for the Ray cluster used only in KFP v2
     :param ray_head_options: head node options, containing the following:
         cpu - number of cpus
         memory - memory
@@ -178,6 +168,16 @@ def code2parquet(
                     (here we are assuming that select language info is in S3, but potentially in the different bucket)
     :return: None
     """
+    # In KFPv2 dsl.RUN_ID_PLACEHOLDER is deprecated and cannot be used since SDK 2.5.0. On another hand we cannot create
+    # a unique string in a component (at runtime) and pass it to the `clean_up_task` of `ExitHandler`, due to
+    # https://github.com/kubeflow/pipelines/issues/10187. Therefore, meantime the user is requested to insert
+    # a unique string created at run creation time.
+    if os.getenv("KFPv2", "0") == "1":
+        print("WARNING: the ray cluster name can be non-unique at runtime, please do not execute simultaneous Runs of the "
+              "same version of the same pipeline !!!")
+        run_id = ray_run_id_KFPv2
+    else:
+        run_id = dsl.RUN_ID_PLACEHOLDER
     # create clean_up task
     clean_up_task = cleanup_ray_op(ray_name=ray_name, run_id=run_id, server_url=server_url, additional_params=additional_params)
     ComponentUtils.add_settings_to_component(clean_up_task, ONE_HOUR_SEC * 2)

transforms/code/code2parquet/ray/Dockerfile

@@ -28,10 +28,10 @@ COPY --chmod=775 --chown=ray:root pyproject.toml pyproject.toml
 RUN pip install --no-cache-dir -e .
 
 # copy the main() entry point to the image 
-COPY src/code2parquet_transform_ray.py .
+COPY --chmod=775 --chown=ray:root src/code2parquet_transform_ray.py .
 
 # copy some of the samples in
-COPY src/code2parquet_local_ray.py local/
+COPY --chmod=775 --chown=ray:root src/code2parquet_local_ray.py local/
 
 # copy test
 COPY test/ test/

transforms/code/code_quality/kfp_ray/code_quality_wf.py

@@ -74,23 +74,11 @@ def compute_exec_params_func(
 # KFPv2 recommends using the `@dsl.component` decorator, which doesn't exist in KFPv1. Therefore, here we use
 # this if/else statement and explicitly call the decorator.
 if os.getenv("KFPv2", "0") == "1":
-    # In KFPv2 dsl.RUN_ID_PLACEHOLDER is deprecated and cannot be used since SDK 2.5.0. On another hand we cannot create
-    # a unique string in a component (at runtime) and pass it to the `clean_up_task` of `ExitHandler`, due to
-    # https://github.com/kubeflow/pipelines/issues/10187. Therefore, meantime we use a unique string created at
-    # compilation time.
-    import uuid
-
     compute_exec_params_op = dsl.component_decorator.component(
         func=compute_exec_params_func, base_image=base_kfp_image
     )
-    print(
-        "WARNING: the ray cluster name can be non-unique at runtime, please do not execute simultaneous Runs of the "
-        + "same version of the same pipeline !!!"
-    )
-    run_id = uuid.uuid4().hex
 else:
     compute_exec_params_op = comp.create_component_from_func(func=compute_exec_params_func, base_image=base_kfp_image)
-    run_id = dsl.RUN_ID_PLACEHOLDER
 
 
 # create Ray cluster
@@ -112,6 +100,7 @@ def compute_exec_params_func(
 def code_quality(
     # Ray cluster
     ray_name: str = "code_quality-kfp-ray",  # name of Ray cluster
+    ray_run_id_KFPv2: str = "",   # Ray cluster unique ID used only in KFP v2
     # Add image_pull_secret and image_pull_policy to ray workers if needed
     ray_head_options: dict = {"cpu": 1, "memory": 4, "image": task_image},
     ray_worker_options: dict = {"replicas": 2, "max_replicas": 2, "min_replicas": 2, "cpu": 2, "memory": 4, "image": task_image},
@@ -136,6 +125,7 @@ def code_quality(
     """
     Pipeline to execute Code Quality transform
     :param ray_name: name of the Ray cluster
+    :param ray_run_id_KFPv2: string holding the id used for the Ray cluster used only in KFP v2
     :param ray_head_options: head node options, containing the following:
         cpu - number of cpus
         memory - memory
@@ -171,6 +161,16 @@ def code_quality(
     :param cq_hf_token - Huggingface auth token to download and use the tokenizer
     :return: None
     """
+    # In KFPv2 dsl.RUN_ID_PLACEHOLDER is deprecated and cannot be used since SDK 2.5.0. On another hand we cannot create
+    # a unique string in a component (at runtime) and pass it to the `clean_up_task` of `ExitHandler`, due to
+    # https://github.com/kubeflow/pipelines/issues/10187. Therefore, meantime the user is requested to insert
+    # a unique string created at run creation time.
+    if os.getenv("KFPv2", "0") == "1":
+        print("WARNING: the ray cluster name can be non-unique at runtime, please do not execute simultaneous Runs of the "
+              "same version of the same pipeline !!!")
+        run_id = ray_run_id_KFPv2
+    else:
+        run_id = dsl.RUN_ID_PLACEHOLDER
     # create clean_up task
     clean_up_task = cleanup_ray_op(ray_name=ray_name, run_id=run_id, server_url=server_url, additional_params=additional_params)
     ComponentUtils.add_settings_to_component(clean_up_task, ONE_HOUR_SEC * 2)

transforms/code/code_quality/ray/Dockerfile

@@ -33,14 +33,14 @@ COPY --chmod=775 --chown=ray:root pyproject.toml pyproject.toml
 RUN pip install --no-cache-dir -e .
 
 # copy the main() entry point to the image 
-COPY ./src/code_quality_transform_ray.py .
+COPY --chmod=775 --chown=ray:root ./src/code_quality_transform_ray.py .
 
 # copy some of the samples in
-COPY ./src/code_quality_local_ray.py local/
+COPY --chmod=775 --chown=ray:root ./src/code_quality_local_ray.py local/
 
 # copy test
-COPY test/ test/
-COPY test-data/ test-data/
+COPY --chmod=775 --chown=ray:root test/ test/
+COPY --chmod=775 --chown=ray:root test-data/ test-data/
 
 # Set environment
 ENV PYTHONPATH /home/ray