We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[v2compat] re-evaluate execution custom properties schema
* [ ] rename task execution `task_name` to `display_name`?

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `sdk/python/kfp/compiler/_default_transformers.py`
Content:
```
1 # Copyright 2019 The Kubeflow Authors
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #      http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 
15 import warnings
16 from kubernetes import client as k8s_client
17 from typing import Callable, Dict, Optional, Text
18 from kfp.dsl._container_op import BaseOp, ContainerOp
19 
20 def add_pod_env(op: BaseOp) -> BaseOp:
21   """Adds environment info if the Pod has the label `add-pod-env = true`.
22     """
23   if isinstance(
24       op, ContainerOp
25   ) and op.pod_labels and 'add-pod-env' in op.pod_labels and op.pod_labels[
26       'add-pod-env'] == 'true':
27     return add_kfp_pod_env(op)
28 
29 
30 def add_kfp_pod_env(op: BaseOp) -> BaseOp:
31   """Adds KFP pod environment info to the specified ContainerOp.
32     """
33   if not isinstance(op, ContainerOp):
34     warnings.warn(
35         'Trying to add default KFP environment variables to an Op that is '
36         'not a ContainerOp. Ignoring request.')
37     return op
38 
39   op.container.add_env_variable(
40       k8s_client.V1EnvVar(name='KFP_POD_NAME',
41                           value_from=k8s_client.V1EnvVarSource(
42                               field_ref=k8s_client.V1ObjectFieldSelector(
43                                   field_path='metadata.name')))
44   ).add_env_variable(
45       k8s_client.V1EnvVar(name='KFP_NAMESPACE',
46                           value_from=k8s_client.V1EnvVarSource(
47                               field_ref=k8s_client.V1ObjectFieldSelector(
48                                   field_path='metadata.namespace')))
49   ).add_env_variable(
50       k8s_client.V1EnvVar(
51           name='WORKFLOW_ID',
52           value_from=k8s_client.
53           V1EnvVarSource(field_ref=k8s_client.V1ObjectFieldSelector(
54               field_path="metadata.labels['workflows.argoproj.io/workflow']")))
55   ).add_env_variable(
56       k8s_client.V1EnvVar(
57           name='ENABLE_CACHING',
58           value_from=k8s_client.
59           V1EnvVarSource(field_ref=k8s_client.V1ObjectFieldSelector(
60               field_path="metadata.labels['pipelines.kubeflow.org/enable_caching']")))
61   )
62   return op
63 
64 
65 def add_pod_labels(labels: Optional[Dict] = None) -> Callable:
66     """Adds provided pod labels to each pod."""
67 
68     def _add_pod_labels(task):
69         for k, v in labels.items():
70             # Only append but not update.
71             # This is needed to bypass TFX pipelines/components.
72             if k not in task.pod_labels:
73                 task.add_pod_label(k, v)
74         return task
75 
76     return _add_pod_labels
77
```

Path: `sdk/python/kfp/compiler/v2_compat.py`
Content:
```
1 # Copyright 2021 The Kubeflow Authors
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #      http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 """Utility functions for enabling v2-compatible pipelines in v1."""
15 import collections
16 import json
17 from typing import Optional
18 
19 from kfp import dsl
20 from kfp.compiler import _default_transformers
21 from kfp.pipeline_spec import pipeline_spec_pb2
22 from kfp.v2 import compiler
23 
24 from kubernetes import client as k8s_client
25 
26 _DEFAULT_LAUNCHER_IMAGE = "gcr.io/ml-pipeline/kfp-launcher:1.6.4"
27 
28 
29 def update_op(op: dsl.ContainerOp,
30               pipeline_name: dsl.PipelineParam,
31               pipeline_root: dsl.PipelineParam,
32               launcher_image: Optional[str] = None) -> None:
33   """Updates the passed in Op for running in v2-compatible mode.
34 
35     Args:
36       op: The Op to update.
37       pipeline_spec: The PipelineSpec for the pipeline under which `op`
38         runs.
39       pipeline_root: The root output directory for pipeline artifacts.
40       launcher_image: An optional launcher image. Useful for tests.
41     """
42   op.is_v2 = True
43   # Inject the launcher binary and overwrite the entrypoint.
44   image_name = launcher_image or _DEFAULT_LAUNCHER_IMAGE
45   launcher_container = dsl.UserContainer(name="kfp-launcher",
46                                          image=image_name,
47                                          command="/bin/mount_launcher.sh",
48                                          mirror_volume_mounts=True)
49 
50   op.add_init_container(launcher_container)
51   op.add_volume(k8s_client.V1Volume(name='kfp-launcher'))
52   op.add_volume_mount(
53       k8s_client.V1VolumeMount(name='kfp-launcher', mount_path='/kfp-launcher'))
54 
55   # op.command + op.args will have the following sections:
56   # 1. args passed to kfp-launcher
57   # 2. a separator "--"
58   # 3. parameters in format "key1=value1", "key2=value2", ...
59   # 4. a separator "--" as end of arguments passed to launcher
60   # 5. (start of op.args) arguments of the original user program command + args
61   #
62   # example:
63   # - command:
64   # - /kfp-launcher/launch
65   # - '--mlmd_server_address'
66   # - $(METADATA_GRPC_SERVICE_HOST)
67   # - '--mlmd_server_port'
68   # - $(METADATA_GRPC_SERVICE_PORT)
69   # - ... # more launcher params
70   # - '--pipeline_task_id'
71   # - $(KFP_POD_NAME)
72   # - '--pipeline_root'
73   # - ''
74   # - '--' # start of parameter values
75   # - first=first
76   # - second=second
77   # - '--' # start of user command and args
78   # args:
79   # - sh
80   # - '-ec'
81   # - |
82   #     program_path=$(mktemp)
83   #     printf "%s" "$0" > "$program_path"
84   #     python3 -u "$program_path" "$@"
85   # - >
86   #     import json
87   #     import xxx
88   #     ...
89   op.command = [
90       "/kfp-launcher/launch",
91       "--mlmd_server_address",
92       "$(METADATA_GRPC_SERVICE_HOST)",
93       "--mlmd_server_port",
94       "$(METADATA_GRPC_SERVICE_PORT)",
95       "--runtime_info_json",
96       "$(KFP_V2_RUNTIME_INFO)",
97       "--container_image",
98       "$(KFP_V2_IMAGE)",
99       "--task_name",
100       op.name,
101       "--pipeline_name",
102       pipeline_name,
103       "--pipeline_run_id",
104       "$(WORKFLOW_ID)",
105       "--pipeline_task_id",
106       "$(KFP_POD_NAME)",
107       "--pipeline_root",
108       pipeline_root,
109       "--enable_caching",
110       "$(ENABLE_CACHING)",
111   ]
112 
113   # Mount necessary environment variables.
114   op.apply(_default_transformers.add_kfp_pod_env)
115   op.container.add_env_variable(
116       k8s_client.V1EnvVar(name="KFP_V2_IMAGE", value=op.container.image))
117 
118   config_map_ref = k8s_client.V1ConfigMapEnvSource(
119       name='metadata-grpc-configmap', optional=True)
120   op.container.add_env_from(
121       k8s_client.V1EnvFromSource(config_map_ref=config_map_ref))
122 
123   op.arguments = list(op.container_spec.command) + list(op.container_spec.args)
124 
125   runtime_info = {
126       "inputParameters": collections.OrderedDict(),
127       "inputArtifacts": collections.OrderedDict(),
128       "outputParameters": collections.OrderedDict(),
129       "outputArtifacts": collections.OrderedDict(),
130   }
131 
132   op.command += ["--"]
133   component_spec = op.component_spec
134   for parameter, spec in sorted(
135       component_spec.input_definitions.parameters.items()):
136     parameter_info = {
137         "type":
138             pipeline_spec_pb2.PrimitiveType.PrimitiveTypeEnum.Name(spec.type),
139     }
140     op.command += [f"{parameter}={op._parameter_arguments[parameter]}"]
141     runtime_info["inputParameters"][parameter] = parameter_info
142   op.command += ["--"]
143 
144   for artifact_name, spec in sorted(
145       component_spec.input_definitions.artifacts.items()):
146     artifact_info = {
147         "metadataPath": op.input_artifact_paths[artifact_name],
148         "schemaTitle": spec.artifact_type.schema_title,
149         "instanceSchema": spec.artifact_type.instance_schema,
150     }
151     runtime_info["inputArtifacts"][artifact_name] = artifact_info
152 
153   for parameter, spec in sorted(
154       component_spec.output_definitions.parameters.items()):
155     parameter_info = {
156         "type":
157             pipeline_spec_pb2.PrimitiveType.PrimitiveTypeEnum.Name(spec.type),
158         "path":
159             op.file_outputs[parameter],
160     }
161     runtime_info["outputParameters"][parameter] = parameter_info
162 
163   for artifact_name, spec in sorted(
164       component_spec.output_definitions.artifacts.items()):
165     # TODO: Assert instance_schema.
166     artifact_info = {
167         # Type used to register output artifacts.
168         "schemaTitle": spec.artifact_type.schema_title,
169         "instanceSchema": spec.artifact_type.instance_schema,
170         # File used to write out the registered artifact ID.
171         "metadataPath": op.file_outputs[artifact_name],
172     }
173     runtime_info["outputArtifacts"][artifact_name] = artifact_info
174 
175   op.container.add_env_variable(
176       k8s_client.V1EnvVar(name="KFP_V2_RUNTIME_INFO",
177                           value=json.dumps(runtime_info)))
178 
179   op.pod_annotations['pipelines.kubeflow.org/v2_component'] = "true"
180   op.pod_labels['pipelines.kubeflow.org/v2_component']= "true"
181
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/sdk/python/kfp/compiler/_default_transformers.py b/sdk/python/kfp/compiler/_default_transformers.py
--- a/sdk/python/kfp/compiler/_default_transformers.py
+++ b/sdk/python/kfp/compiler/_default_transformers.py
@@ -41,6 +41,11 @@
                           value_from=k8s_client.V1EnvVarSource(
                               field_ref=k8s_client.V1ObjectFieldSelector(
                                   field_path='metadata.name')))
+  ).add_env_variable(
+      k8s_client.V1EnvVar(name='KFP_POD_UID',
+                          value_from=k8s_client.V1EnvVarSource(
+                              field_ref=k8s_client.V1ObjectFieldSelector(
+                                  field_path='metadata.uid')))
   ).add_env_variable(
       k8s_client.V1EnvVar(name='KFP_NAMESPACE',
                           value_from=k8s_client.V1EnvVarSource(
@@ -52,6 +57,12 @@
           value_from=k8s_client.
           V1EnvVarSource(field_ref=k8s_client.V1ObjectFieldSelector(
               field_path="metadata.labels['workflows.argoproj.io/workflow']")))
+  ).add_env_variable(
+      k8s_client.V1EnvVar(
+          name='KFP_RUN_ID',
+          value_from=k8s_client.
+          V1EnvVarSource(field_ref=k8s_client.V1ObjectFieldSelector(
+              field_path="metadata.labels['pipeline/runid']")))
   ).add_env_variable(
       k8s_client.V1EnvVar(
           name='ENABLE_CACHING',
diff --git a/sdk/python/kfp/compiler/v2_compat.py b/sdk/python/kfp/compiler/v2_compat.py
--- a/sdk/python/kfp/compiler/v2_compat.py
+++ b/sdk/python/kfp/compiler/v2_compat.py
@@ -100,10 +100,16 @@
       op.name,
       "--pipeline_name",
       pipeline_name,
-      "--pipeline_run_id",
-      "$(WORKFLOW_ID)",
-      "--pipeline_task_id",
+      "--run_id",
+      "$(KFP_RUN_ID)",
+      "--run_resource",
+      "workflows.argoproj.io/$(WORKFLOW_ID)",
+      "--namespace",
+      "$(KFP_NAMESPACE)",
+      "--pod_name",
       "$(KFP_POD_NAME)",
+      "--pod_uid",
+      "$(KFP_POD_UID)",
       "--pipeline_root",
       pipeline_root,
       "--enable_caching",

{"golden_diff": "diff --git a/sdk/python/kfp/compiler/_default_transformers.py b/sdk/python/kfp/compiler/_default_transformers.py\n--- a/sdk/python/kfp/compiler/_default_transformers.py\n+++ b/sdk/python/kfp/compiler/_default_transformers.py\n@@ -41,6 +41,11 @@\n                           value_from=k8s_client.V1EnvVarSource(\n                               field_ref=k8s_client.V1ObjectFieldSelector(\n                                   field_path='metadata.name')))\n+  ).add_env_variable(\n+      k8s_client.V1EnvVar(name='KFP_POD_UID',\n+                          value_from=k8s_client.V1EnvVarSource(\n+                              field_ref=k8s_client.V1ObjectFieldSelector(\n+                                  field_path='metadata.uid')))\n   ).add_env_variable(\n       k8s_client.V1EnvVar(name='KFP_NAMESPACE',\n                           value_from=k8s_client.V1EnvVarSource(\n@@ -52,6 +57,12 @@\n           value_from=k8s_client.\n           V1EnvVarSource(field_ref=k8s_client.V1ObjectFieldSelector(\n               field_path=\"metadata.labels['workflows.argoproj.io/workflow']\")))\n+  ).add_env_variable(\n+      k8s_client.V1EnvVar(\n+          name='KFP_RUN_ID',\n+          value_from=k8s_client.\n+          V1EnvVarSource(field_ref=k8s_client.V1ObjectFieldSelector(\n+              field_path=\"metadata.labels['pipeline/runid']\")))\n   ).add_env_variable(\n       k8s_client.V1EnvVar(\n           name='ENABLE_CACHING',\ndiff --git a/sdk/python/kfp/compiler/v2_compat.py b/sdk/python/kfp/compiler/v2_compat.py\n--- a/sdk/python/kfp/compiler/v2_compat.py\n+++ b/sdk/python/kfp/compiler/v2_compat.py\n@@ -100,10 +100,16 @@\n       op.name,\n       \"--pipeline_name\",\n       pipeline_name,\n-      \"--pipeline_run_id\",\n-      \"$(WORKFLOW_ID)\",\n-      \"--pipeline_task_id\",\n+      \"--run_id\",\n+      \"$(KFP_RUN_ID)\",\n+      \"--run_resource\",\n+      \"workflows.argoproj.io/$(WORKFLOW_ID)\",\n+      \"--namespace\",\n+      \"$(KFP_NAMESPACE)\",\n+      \"--pod_name\",\n       \"$(KFP_POD_NAME)\",\n+      \"--pod_uid\",\n+      \"$(KFP_POD_UID)\",\n       \"--pipeline_root\",\n       pipeline_root,\n       \"--enable_caching\",\n", "issue": "[v2compat] re-evaluate execution custom properties schema\n* [ ] rename task execution `task_name` to `display_name`?\n", "before_files": [{"content": "# Copyright 2019 The Kubeflow Authors\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#      http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport warnings\nfrom kubernetes import client as k8s_client\nfrom typing import Callable, Dict, Optional, Text\nfrom kfp.dsl._container_op import BaseOp, ContainerOp\n\ndef add_pod_env(op: BaseOp) -> BaseOp:\n  \"\"\"Adds environment info if the Pod has the label `add-pod-env = true`.\n    \"\"\"\n  if isinstance(\n      op, ContainerOp\n  ) and op.pod_labels and 'add-pod-env' in op.pod_labels and op.pod_labels[\n      'add-pod-env'] == 'true':\n    return add_kfp_pod_env(op)\n\n\ndef add_kfp_pod_env(op: BaseOp) -> BaseOp:\n  \"\"\"Adds KFP pod environment info to the specified ContainerOp.\n    \"\"\"\n  if not isinstance(op, ContainerOp):\n    warnings.warn(\n        'Trying to add default KFP environment variables to an Op that is '\n        'not a ContainerOp. Ignoring request.')\n    return op\n\n  op.container.add_env_variable(\n      k8s_client.V1EnvVar(name='KFP_POD_NAME',\n                          value_from=k8s_client.V1EnvVarSource(\n                              field_ref=k8s_client.V1ObjectFieldSelector(\n                                  field_path='metadata.name')))\n  ).add_env_variable(\n      k8s_client.V1EnvVar(name='KFP_NAMESPACE',\n                          value_from=k8s_client.V1EnvVarSource(\n                              field_ref=k8s_client.V1ObjectFieldSelector(\n                                  field_path='metadata.namespace')))\n  ).add_env_variable(\n      k8s_client.V1EnvVar(\n          name='WORKFLOW_ID',\n          value_from=k8s_client.\n          V1EnvVarSource(field_ref=k8s_client.V1ObjectFieldSelector(\n              field_path=\"metadata.labels['workflows.argoproj.io/workflow']\")))\n  ).add_env_variable(\n      k8s_client.V1EnvVar(\n          name='ENABLE_CACHING',\n          value_from=k8s_client.\n          V1EnvVarSource(field_ref=k8s_client.V1ObjectFieldSelector(\n              field_path=\"metadata.labels['pipelines.kubeflow.org/enable_caching']\")))\n  )\n  return op\n\n\ndef add_pod_labels(labels: Optional[Dict] = None) -> Callable:\n    \"\"\"Adds provided pod labels to each pod.\"\"\"\n\n    def _add_pod_labels(task):\n        for k, v in labels.items():\n            # Only append but not update.\n            # This is needed to bypass TFX pipelines/components.\n            if k not in task.pod_labels:\n                task.add_pod_label(k, v)\n        return task\n\n    return _add_pod_labels\n", "path": "sdk/python/kfp/compiler/_default_transformers.py"}, {"content": "# Copyright 2021 The Kubeflow Authors\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#      http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\"\"\"Utility functions for enabling v2-compatible pipelines in v1.\"\"\"\nimport collections\nimport json\nfrom typing import Optional\n\nfrom kfp import dsl\nfrom kfp.compiler import _default_transformers\nfrom kfp.pipeline_spec import pipeline_spec_pb2\nfrom kfp.v2 import compiler\n\nfrom kubernetes import client as k8s_client\n\n_DEFAULT_LAUNCHER_IMAGE = \"gcr.io/ml-pipeline/kfp-launcher:1.6.4\"\n\n\ndef update_op(op: dsl.ContainerOp,\n              pipeline_name: dsl.PipelineParam,\n              pipeline_root: dsl.PipelineParam,\n              launcher_image: Optional[str] = None) -> None:\n  \"\"\"Updates the passed in Op for running in v2-compatible mode.\n\n    Args:\n      op: The Op to update.\n      pipeline_spec: The PipelineSpec for the pipeline under which `op`\n        runs.\n      pipeline_root: The root output directory for pipeline artifacts.\n      launcher_image: An optional launcher image. Useful for tests.\n    \"\"\"\n  op.is_v2 = True\n  # Inject the launcher binary and overwrite the entrypoint.\n  image_name = launcher_image or _DEFAULT_LAUNCHER_IMAGE\n  launcher_container = dsl.UserContainer(name=\"kfp-launcher\",\n                                         image=image_name,\n                                         command=\"/bin/mount_launcher.sh\",\n                                         mirror_volume_mounts=True)\n\n  op.add_init_container(launcher_container)\n  op.add_volume(k8s_client.V1Volume(name='kfp-launcher'))\n  op.add_volume_mount(\n      k8s_client.V1VolumeMount(name='kfp-launcher', mount_path='/kfp-launcher'))\n\n  # op.command + op.args will have the following sections:\n  # 1. args passed to kfp-launcher\n  # 2. a separator \"--\"\n  # 3. parameters in format \"key1=value1\", \"key2=value2\", ...\n  # 4. a separator \"--\" as end of arguments passed to launcher\n  # 5. (start of op.args) arguments of the original user program command + args\n  #\n  # example:\n  # - command:\n  # - /kfp-launcher/launch\n  # - '--mlmd_server_address'\n  # - $(METADATA_GRPC_SERVICE_HOST)\n  # - '--mlmd_server_port'\n  # - $(METADATA_GRPC_SERVICE_PORT)\n  # - ... # more launcher params\n  # - '--pipeline_task_id'\n  # - $(KFP_POD_NAME)\n  # - '--pipeline_root'\n  # - ''\n  # - '--' # start of parameter values\n  # - first=first\n  # - second=second\n  # - '--' # start of user command and args\n  # args:\n  # - sh\n  # - '-ec'\n  # - |\n  #     program_path=$(mktemp)\n  #     printf \"%s\" \"$0\" > \"$program_path\"\n  #     python3 -u \"$program_path\" \"$@\"\n  # - >\n  #     import json\n  #     import xxx\n  #     ...\n  op.command = [\n      \"/kfp-launcher/launch\",\n      \"--mlmd_server_address\",\n      \"$(METADATA_GRPC_SERVICE_HOST)\",\n      \"--mlmd_server_port\",\n      \"$(METADATA_GRPC_SERVICE_PORT)\",\n      \"--runtime_info_json\",\n      \"$(KFP_V2_RUNTIME_INFO)\",\n      \"--container_image\",\n      \"$(KFP_V2_IMAGE)\",\n      \"--task_name\",\n      op.name,\n      \"--pipeline_name\",\n      pipeline_name,\n      \"--pipeline_run_id\",\n      \"$(WORKFLOW_ID)\",\n      \"--pipeline_task_id\",\n      \"$(KFP_POD_NAME)\",\n      \"--pipeline_root\",\n      pipeline_root,\n      \"--enable_caching\",\n      \"$(ENABLE_CACHING)\",\n  ]\n\n  # Mount necessary environment variables.\n  op.apply(_default_transformers.add_kfp_pod_env)\n  op.container.add_env_variable(\n      k8s_client.V1EnvVar(name=\"KFP_V2_IMAGE\", value=op.container.image))\n\n  config_map_ref = k8s_client.V1ConfigMapEnvSource(\n      name='metadata-grpc-configmap', optional=True)\n  op.container.add_env_from(\n      k8s_client.V1EnvFromSource(config_map_ref=config_map_ref))\n\n  op.arguments = list(op.container_spec.command) + list(op.container_spec.args)\n\n  runtime_info = {\n      \"inputParameters\": collections.OrderedDict(),\n      \"inputArtifacts\": collections.OrderedDict(),\n      \"outputParameters\": collections.OrderedDict(),\n      \"outputArtifacts\": collections.OrderedDict(),\n  }\n\n  op.command += [\"--\"]\n  component_spec = op.component_spec\n  for parameter, spec in sorted(\n      component_spec.input_definitions.parameters.items()):\n    parameter_info = {\n        \"type\":\n            pipeline_spec_pb2.PrimitiveType.PrimitiveTypeEnum.Name(spec.type),\n    }\n    op.command += [f\"{parameter}={op._parameter_arguments[parameter]}\"]\n    runtime_info[\"inputParameters\"][parameter] = parameter_info\n  op.command += [\"--\"]\n\n  for artifact_name, spec in sorted(\n      component_spec.input_definitions.artifacts.items()):\n    artifact_info = {\n        \"metadataPath\": op.input_artifact_paths[artifact_name],\n        \"schemaTitle\": spec.artifact_type.schema_title,\n        \"instanceSchema\": spec.artifact_type.instance_schema,\n    }\n    runtime_info[\"inputArtifacts\"][artifact_name] = artifact_info\n\n  for parameter, spec in sorted(\n      component_spec.output_definitions.parameters.items()):\n    parameter_info = {\n        \"type\":\n            pipeline_spec_pb2.PrimitiveType.PrimitiveTypeEnum.Name(spec.type),\n        \"path\":\n            op.file_outputs[parameter],\n    }\n    runtime_info[\"outputParameters\"][parameter] = parameter_info\n\n  for artifact_name, spec in sorted(\n      component_spec.output_definitions.artifacts.items()):\n    # TODO: Assert instance_schema.\n    artifact_info = {\n        # Type used to register output artifacts.\n        \"schemaTitle\": spec.artifact_type.schema_title,\n        \"instanceSchema\": spec.artifact_type.instance_schema,\n        # File used to write out the registered artifact ID.\n        \"metadataPath\": op.file_outputs[artifact_name],\n    }\n    runtime_info[\"outputArtifacts\"][artifact_name] = artifact_info\n\n  op.container.add_env_variable(\n      k8s_client.V1EnvVar(name=\"KFP_V2_RUNTIME_INFO\",\n                          value=json.dumps(runtime_info)))\n\n  op.pod_annotations['pipelines.kubeflow.org/v2_component'] = \"true\"\n  op.pod_labels['pipelines.kubeflow.org/v2_component']= \"true\"\n", "path": "sdk/python/kfp/compiler/v2_compat.py"}], "after_files": [{"content": "# Copyright 2019 The Kubeflow Authors\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#      http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport warnings\nfrom kubernetes import client as k8s_client\nfrom typing import Callable, Dict, Optional, Text\nfrom kfp.dsl._container_op import BaseOp, ContainerOp\n\ndef add_pod_env(op: BaseOp) -> BaseOp:\n  \"\"\"Adds environment info if the Pod has the label `add-pod-env = true`.\n    \"\"\"\n  if isinstance(\n      op, ContainerOp\n  ) and op.pod_labels and 'add-pod-env' in op.pod_labels and op.pod_labels[\n      'add-pod-env'] == 'true':\n    return add_kfp_pod_env(op)\n\n\ndef add_kfp_pod_env(op: BaseOp) -> BaseOp:\n  \"\"\"Adds KFP pod environment info to the specified ContainerOp.\n    \"\"\"\n  if not isinstance(op, ContainerOp):\n    warnings.warn(\n        'Trying to add default KFP environment variables to an Op that is '\n        'not a ContainerOp. Ignoring request.')\n    return op\n\n  op.container.add_env_variable(\n      k8s_client.V1EnvVar(name='KFP_POD_NAME',\n                          value_from=k8s_client.V1EnvVarSource(\n                              field_ref=k8s_client.V1ObjectFieldSelector(\n                                  field_path='metadata.name')))\n  ).add_env_variable(\n      k8s_client.V1EnvVar(name='KFP_POD_UID',\n                          value_from=k8s_client.V1EnvVarSource(\n                              field_ref=k8s_client.V1ObjectFieldSelector(\n                                  field_path='metadata.uid')))\n  ).add_env_variable(\n      k8s_client.V1EnvVar(name='KFP_NAMESPACE',\n                          value_from=k8s_client.V1EnvVarSource(\n                              field_ref=k8s_client.V1ObjectFieldSelector(\n                                  field_path='metadata.namespace')))\n  ).add_env_variable(\n      k8s_client.V1EnvVar(\n          name='WORKFLOW_ID',\n          value_from=k8s_client.\n          V1EnvVarSource(field_ref=k8s_client.V1ObjectFieldSelector(\n              field_path=\"metadata.labels['workflows.argoproj.io/workflow']\")))\n  ).add_env_variable(\n      k8s_client.V1EnvVar(\n          name='KFP_RUN_ID',\n          value_from=k8s_client.\n          V1EnvVarSource(field_ref=k8s_client.V1ObjectFieldSelector(\n              field_path=\"metadata.labels['pipeline/runid']\")))\n  ).add_env_variable(\n      k8s_client.V1EnvVar(\n          name='ENABLE_CACHING',\n          value_from=k8s_client.\n          V1EnvVarSource(field_ref=k8s_client.V1ObjectFieldSelector(\n              field_path=\"metadata.labels['pipelines.kubeflow.org/enable_caching']\")))\n  )\n  return op\n\n\ndef add_pod_labels(labels: Optional[Dict] = None) -> Callable:\n    \"\"\"Adds provided pod labels to each pod.\"\"\"\n\n    def _add_pod_labels(task):\n        for k, v in labels.items():\n            # Only append but not update.\n            # This is needed to bypass TFX pipelines/components.\n            if k not in task.pod_labels:\n                task.add_pod_label(k, v)\n        return task\n\n    return _add_pod_labels\n", "path": "sdk/python/kfp/compiler/_default_transformers.py"}, {"content": "# Copyright 2021 The Kubeflow Authors\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#      http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\"\"\"Utility functions for enabling v2-compatible pipelines in v1.\"\"\"\nimport collections\nimport json\nfrom typing import Optional\n\nfrom kfp import dsl\nfrom kfp.compiler import _default_transformers\nfrom kfp.pipeline_spec import pipeline_spec_pb2\nfrom kfp.v2 import compiler\n\nfrom kubernetes import client as k8s_client\n\n_DEFAULT_LAUNCHER_IMAGE = \"gcr.io/ml-pipeline/kfp-launcher:1.6.4\"\n\n\ndef update_op(op: dsl.ContainerOp,\n              pipeline_name: dsl.PipelineParam,\n              pipeline_root: dsl.PipelineParam,\n              launcher_image: Optional[str] = None) -> None:\n  \"\"\"Updates the passed in Op for running in v2-compatible mode.\n\n    Args:\n      op: The Op to update.\n      pipeline_spec: The PipelineSpec for the pipeline under which `op`\n        runs.\n      pipeline_root: The root output directory for pipeline artifacts.\n      launcher_image: An optional launcher image. Useful for tests.\n    \"\"\"\n  op.is_v2 = True\n  # Inject the launcher binary and overwrite the entrypoint.\n  image_name = launcher_image or _DEFAULT_LAUNCHER_IMAGE\n  launcher_container = dsl.UserContainer(name=\"kfp-launcher\",\n                                         image=image_name,\n                                         command=\"/bin/mount_launcher.sh\",\n                                         mirror_volume_mounts=True)\n\n  op.add_init_container(launcher_container)\n  op.add_volume(k8s_client.V1Volume(name='kfp-launcher'))\n  op.add_volume_mount(\n      k8s_client.V1VolumeMount(name='kfp-launcher', mount_path='/kfp-launcher'))\n\n  # op.command + op.args will have the following sections:\n  # 1. args passed to kfp-launcher\n  # 2. a separator \"--\"\n  # 3. parameters in format \"key1=value1\", \"key2=value2\", ...\n  # 4. a separator \"--\" as end of arguments passed to launcher\n  # 5. (start of op.args) arguments of the original user program command + args\n  #\n  # example:\n  # - command:\n  # - /kfp-launcher/launch\n  # - '--mlmd_server_address'\n  # - $(METADATA_GRPC_SERVICE_HOST)\n  # - '--mlmd_server_port'\n  # - $(METADATA_GRPC_SERVICE_PORT)\n  # - ... # more launcher params\n  # - '--pipeline_task_id'\n  # - $(KFP_POD_NAME)\n  # - '--pipeline_root'\n  # - ''\n  # - '--' # start of parameter values\n  # - first=first\n  # - second=second\n  # - '--' # start of user command and args\n  # args:\n  # - sh\n  # - '-ec'\n  # - |\n  #     program_path=$(mktemp)\n  #     printf \"%s\" \"$0\" > \"$program_path\"\n  #     python3 -u \"$program_path\" \"$@\"\n  # - >\n  #     import json\n  #     import xxx\n  #     ...\n  op.command = [\n      \"/kfp-launcher/launch\",\n      \"--mlmd_server_address\",\n      \"$(METADATA_GRPC_SERVICE_HOST)\",\n      \"--mlmd_server_port\",\n      \"$(METADATA_GRPC_SERVICE_PORT)\",\n      \"--runtime_info_json\",\n      \"$(KFP_V2_RUNTIME_INFO)\",\n      \"--container_image\",\n      \"$(KFP_V2_IMAGE)\",\n      \"--task_name\",\n      op.name,\n      \"--pipeline_name\",\n      pipeline_name,\n      \"--run_id\",\n      \"$(KFP_RUN_ID)\",\n      \"--run_resource\",\n      \"workflows.argoproj.io/$(WORKFLOW_ID)\",\n      \"--namespace\",\n      \"$(KFP_NAMESPACE)\",\n      \"--pod_name\",\n      \"$(KFP_POD_NAME)\",\n      \"--pod_uid\",\n      \"$(KFP_POD_UID)\",\n      \"--pipeline_root\",\n      pipeline_root,\n      \"--enable_caching\",\n      \"$(ENABLE_CACHING)\",\n  ]\n\n  # Mount necessary environment variables.\n  op.apply(_default_transformers.add_kfp_pod_env)\n  op.container.add_env_variable(\n      k8s_client.V1EnvVar(name=\"KFP_V2_IMAGE\", value=op.container.image))\n\n  config_map_ref = k8s_client.V1ConfigMapEnvSource(\n      name='metadata-grpc-configmap', optional=True)\n  op.container.add_env_from(\n      k8s_client.V1EnvFromSource(config_map_ref=config_map_ref))\n\n  op.arguments = list(op.container_spec.command) + list(op.container_spec.args)\n\n  runtime_info = {\n      \"inputParameters\": collections.OrderedDict(),\n      \"inputArtifacts\": collections.OrderedDict(),\n      \"outputParameters\": collections.OrderedDict(),\n      \"outputArtifacts\": collections.OrderedDict(),\n  }\n\n  op.command += [\"--\"]\n  component_spec = op.component_spec\n  for parameter, spec in sorted(\n      component_spec.input_definitions.parameters.items()):\n    parameter_info = {\n        \"type\":\n            pipeline_spec_pb2.PrimitiveType.PrimitiveTypeEnum.Name(spec.type),\n    }\n    op.command += [f\"{parameter}={op._parameter_arguments[parameter]}\"]\n    runtime_info[\"inputParameters\"][parameter] = parameter_info\n  op.command += [\"--\"]\n\n  for artifact_name, spec in sorted(\n      component_spec.input_definitions.artifacts.items()):\n    artifact_info = {\n        \"metadataPath\": op.input_artifact_paths[artifact_name],\n        \"schemaTitle\": spec.artifact_type.schema_title,\n        \"instanceSchema\": spec.artifact_type.instance_schema,\n    }\n    runtime_info[\"inputArtifacts\"][artifact_name] = artifact_info\n\n  for parameter, spec in sorted(\n      component_spec.output_definitions.parameters.items()):\n    parameter_info = {\n        \"type\":\n            pipeline_spec_pb2.PrimitiveType.PrimitiveTypeEnum.Name(spec.type),\n        \"path\":\n            op.file_outputs[parameter],\n    }\n    runtime_info[\"outputParameters\"][parameter] = parameter_info\n\n  for artifact_name, spec in sorted(\n      component_spec.output_definitions.artifacts.items()):\n    # TODO: Assert instance_schema.\n    artifact_info = {\n        # Type used to register output artifacts.\n        \"schemaTitle\": spec.artifact_type.schema_title,\n        \"instanceSchema\": spec.artifact_type.instance_schema,\n        # File used to write out the registered artifact ID.\n        \"metadataPath\": op.file_outputs[artifact_name],\n    }\n    runtime_info[\"outputArtifacts\"][artifact_name] = artifact_info\n\n  op.container.add_env_variable(\n      k8s_client.V1EnvVar(name=\"KFP_V2_RUNTIME_INFO\",\n                          value=json.dumps(runtime_info)))\n\n  op.pod_annotations['pipelines.kubeflow.org/v2_component'] = \"true\"\n  op.pod_labels['pipelines.kubeflow.org/v2_component']= \"true\"\n", "path": "sdk/python/kfp/compiler/v2_compat.py"}]}