AIP-8089 use unique {{workflow.uid}}

Author: talebzeghmi

kfp/dsl/_volume_op.py

@@ -105,7 +105,7 @@ def __init__(self,
         if not match_serialized_pipelineparam(str(resource_name)):
             resource_name = sanitize_k8s_name(resource_name)
         pvc_metadata = V1ObjectMeta(
-            name="{{workflow.name}}-%s" % resource_name if generate_unique_name else resource_name,
+            name="{{workflow.uid}}-%s" % resource_name if generate_unique_name else resource_name,
             annotations=annotations)
         requested_resources = V1ResourceRequirements(requests={"storage": size})
         pvc_spec = V1PersistentVolumeClaimSpec(

kfp/dsl/_volume_snapshot_op.py

@@ -96,7 +96,7 @@ def __init__(self,
         if not match_serialized_pipelineparam(str(resource_name)):
             resource_name = sanitize_k8s_name(resource_name)
         snapshot_metadata = V1ObjectMeta(
-            name="{{workflow.name}}-%s" % resource_name,
+            name="{{workflow.uid}}-%s" % resource_name,
             annotations=annotations)
         k8s_resource = {
             "apiVersion": api_version,

metaflow/plugins/aip/aip.py

@@ -94,7 +94,7 @@ class StepVariables:
     user_code_retries: int
 
 
-METAFLOW_RUN_ID = "argo-{{workflow.name}}"
+METAFLOW_RUN_ID = "argo-{{workflow.uid}}"
 FLOW_PARAMETERS_JSON = "{{workflow.parameters}}"
 
 
@@ -963,12 +963,12 @@ def _create_volume(
             api_version="argoproj.io/v1alpha1",
             controller=True,
             kind="Workflow",
-            name="{{workflow.name}}",
+            name="{{workflow.uid}}",
             uid=workflow_uid,
         )
         owner_references = [owner_reference]
         pvc_metadata = V1ObjectMeta(
-            name=f"{{{{workflow.name}}}}-{volume_name}-pvc",
+            name=f"{{{{workflow.uid}}}}-pvc",
             owner_references=owner_references,
         )
         k8s_resource = V1PersistentVolumeClaim(
@@ -1485,7 +1485,7 @@ def _create_metaflow_step_op(
                 if aip_component.interruptible_decorator
                 else " --not-interruptible "
             )
-            + " --workflow_name {{workflow.name}}"
+            + " --workflow_name {{workflow.uid}}"
         )
 
         if node.name == "start":
@@ -1561,7 +1561,7 @@ def _create_workflow_uid_op(
                     f"{package_commands}"
                     " && python -m metaflow.plugins.aip.aip_get_workflow_uid"
                     f" --s3_sensor_path '{s3_sensor_path}'"
-                    " --workflow_name {{workflow.name}}"
+                    " --workflow_name {{workflow.uid}}"
                 ),
             ]
             workflow_uid_op: ContainerOp = dsl.ContainerOp(
@@ -1627,7 +1627,7 @@ def _create_s3_sensor_op(
             (
                 f"{package_commands}"
                 " && python -m metaflow.plugins.aip.aip_s3_sensor"
-                " --run_id argo-{{workflow.name}}"
+                " --run_id argo-{{workflow.uid}}"
                 f" --flow_name {self.name}"
                 f" --flow_parameters_json '{FLOW_PARAMETERS_JSON}'"
                 f" --path {path}"
@@ -1762,7 +1762,7 @@ def _get_aip_exit_handler_op(
                 f"{package_commands}"
                 " && python -m metaflow.plugins.aip.aip_exit_handler"
                 f" --flow_name {self.name}"
-                " --run_id {{workflow.name}}"
+                " --run_id {{workflow.uid}}"
                 f" --env_variables_json {json.dumps(json.dumps(env_variables))}"
                 f" --flow_parameters_json {flow_parameters_json if flow_parameters else '{}'}"
                 "  --status {{workflow.status}}"
@@ -1813,7 +1813,7 @@ def _get_user_defined_exit_handler_op(
                 f"{package_commands}"
                 f" && METAFLOW_USER=aip-user python {os.path.basename(sys.argv[0])} {top_level} aip user-defined-exit-handler"
                 f" --flow_name {self.name}"
-                " --run_id {{workflow.name}}"
+                " --run_id {{workflow.uid}}"
                 f" --env_variables_json {json.dumps(json.dumps(env_variables))}"
                 f" --flow_parameters_json {flow_parameters_json if flow_parameters else '{}'}"
                 "  --status {{workflow.status}}"