[SDK] Add pod labels for telemetry purpose. (#3578)

* add telemetry pod labels

* revert the id label

* update compiler tests

* update cli arg

* bypass tfx

* update docstring

sdk/python/kfp/compiler/_default_transformers.py

@@ -12,16 +12,31 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from typing import Callable, Dict, Optional, Text
 from ..dsl._container_op import BaseOp, ContainerOp
 
+# Pod label indicating the SDK type from which the pipeline is
+# generated. By default it's set to kfp.
+_SDK_ENV_LABEL = 'pipelines.kubeflow.org/pipeline-sdk-type'
+_SDK_ENV_DEFAULT = 'kfp'
+
+
+def get_default_telemetry_labels() -> Dict[Text, Text]:
+    """Returns the default pod labels for telemetry purpose."""
+    result = {
+        _SDK_ENV_LABEL: _SDK_ENV_DEFAULT,
+    }
+    return result
+
+
 def add_pod_env(op: BaseOp) -> BaseOp:
     """Adds pod environment info to ContainerOp.
     """
-    if isinstance(op, ContainerOp) and op.pod_labels and 'add-pod-env' in op.pod_labels and op.pod_labels['add-pod-env'] == 'true':
+    if isinstance(op, ContainerOp) and op.pod_labels and op.pod_labels.get('add-pod-env', None) == 'true':
         from kubernetes import client as k8s_client
         op.container.add_env_variable(
             k8s_client.V1EnvVar(
-                name='KFP_POD_NAME', 
+                name='KFP_POD_NAME',
                 value_from=k8s_client.V1EnvVarSource(
                     field_ref=k8s_client.V1ObjectFieldSelector(
                         field_path='metadata.name'
@@ -30,7 +45,7 @@ def add_pod_env(op: BaseOp) -> BaseOp:
             )
         ).add_env_variable(
             k8s_client.V1EnvVar(
-                name='KFP_NAMESPACE', 
+                name='KFP_NAMESPACE',
                 value_from=k8s_client.V1EnvVarSource(
                     field_ref=k8s_client.V1ObjectFieldSelector(
                         field_path='metadata.namespace'
@@ -38,4 +53,18 @@ def add_pod_env(op: BaseOp) -> BaseOp:
                 )
             )
         )
-    return op
+    return op
+
+
+def add_pod_labels(labels: Optional[Dict[Text, Text]] = None) -> Callable:
+    """Adds provided pod labels to each pod."""
+
+    def _add_pod_labels(task):
+        for k, v in labels.items():
+            # Only append but not update.
+            # This is needed to bypass TFX pipelines/components.
+            if k not in task.pod_labels:
+                task.add_pod_label(k, v)
+        return task
+
+    return _add_pod_labels

sdk/python/kfp/compiler/compiler.py

@@ -27,7 +27,7 @@ from kfp.dsl import _for_loop
 from .. import dsl
 from ._k8s_helper import convert_k8s_obj_to_json, sanitize_k8s_name
 from ._op_to_template import _op_to_template
-from ._default_transformers import add_pod_env
+from ._default_transformers import add_pod_env, add_pod_labels, get_default_telemetry_labels
 
 from ..components.structures import InputSpec
 from ..components._yaml_utils import dump_yaml
@@ -759,7 +759,8 @@ class Compiler(object):
       pipeline_description: Text=None,
       params_list: List[dsl.PipelineParam]=None,
       pipeline_conf: dsl.PipelineConf = None,
-      ) -> Dict[Text, Any]:
+      allow_telemetry: bool = True,
+  ) -> Dict[Text, Any]:
     """ Internal implementation of create_workflow."""
     params_list = params_list or []
     argspec = inspect.getfullargspec(pipeline_func)
@@ -821,6 +822,13 @@ class Compiler(object):
             default=param.value) for param in params_list]
 
     op_transformers = [add_pod_env]
+    # By default adds telemetry instruments. Users can opt out toggling
+    # allow_telemetry.
+    # Also, TFX pipelines will be bypassed for pipeline compiled by tfx>0.21.4.
+    if allow_telemetry:
+      pod_labels = get_default_telemetry_labels()
+      op_transformers.append(add_pod_labels(pod_labels))
+
     op_transformers.extend(pipeline_conf.op_transformers)
 
     workflow = self._create_pipeline_workflow(
@@ -879,7 +887,14 @@ class Compiler(object):
     """Compile the given pipeline function into workflow."""
     return self._create_workflow(pipeline_func=pipeline_func, pipeline_conf=pipeline_conf)
 
-  def compile(self, pipeline_func, package_path, type_check=True, pipeline_conf: dsl.PipelineConf = None):
+  def compile(
+      self,
+      pipeline_func,
+      package_path,
+      type_check=True,
+      pipeline_conf: dsl.PipelineConf = None,
+      allow_telemetry: bool = True,
+  ):
     """Compile the given pipeline function into workflow yaml.
 
     Args:
@@ -887,6 +902,9 @@ class Compiler(object):
       package_path: the output workflow tar.gz file path. for example, "~/a.tar.gz"
       type_check: whether to enable the type check or not, default: False.
       pipeline_conf: PipelineConf instance. Can specify op transforms, image pull secrets and other pipeline-level configuration options. Overrides any configuration that may be set by the pipeline.
+      allow_telemetry: If set to true, two pod labels will be attached to k8s
+        pods spawned by this pipeline: 1) pipeline SDK style, 2) pipeline random
+        ID.
     """
     import kfp
     type_check_old_value = kfp.TYPE_CHECK
@@ -895,7 +913,8 @@ class Compiler(object):
       self._create_and_write_workflow(
           pipeline_func=pipeline_func,
           pipeline_conf=pipeline_conf,
-          package_path=package_path)
+          package_path=package_path,
+          allow_telemetry=allow_telemetry)
     finally:
       kfp.TYPE_CHECK = type_check_old_value
 
@@ -942,7 +961,8 @@ class Compiler(object):
       pipeline_description: Text=None,
       params_list: List[dsl.PipelineParam]=None,
       pipeline_conf: dsl.PipelineConf=None,
-      package_path: Text=None
+      package_path: Text=None,
+      allow_telemetry: bool=True
   ) -> None:
     """Compile the given pipeline function and dump it to specified file format."""
     workflow = self._create_workflow(
@@ -950,7 +970,8 @@ class Compiler(object):
         pipeline_name,
         pipeline_description,
         params_list,
-        pipeline_conf)
+        pipeline_conf,
+        allow_telemetry)
     self._write_workflow(workflow, package_path)
     _validate_workflow(workflow)
 

sdk/python/kfp/compiler/main.py

@@ -23,6 +23,16 @@ import sys
 import tempfile
 from deprecated.sphinx import deprecated
 
+def _str2bool(v):
+  if isinstance(v, bool):
+    return v
+  if v.lower() in ('yes', 'true', 't', 'y', '1'):
+    return True
+  elif v.lower() in ('no', 'false', 'f', 'n', '0'):
+    return False
+  else:
+    raise argparse.ArgumentTypeError('Boolean value expected.')
+
 
 def parse_arguments():
   """Parse command line arguments."""
@@ -47,12 +57,16 @@ def parse_arguments():
   parser.add_argument('--disable-type-check',
                       action='store_true',
                       help='disable the type check, default is enabled.')
+  parser.add_argument('--disable-telemetry',
+                      action='store_true',
+                      help='disable adding telemetry labels, default is enabled.')
 
   args = parser.parse_args()
   return args
 
 
-def _compile_pipeline_function(pipeline_funcs, function_name, output_path, type_check):
+def _compile_pipeline_function(
+    pipeline_funcs, function_name, output_path, type_check, allow_telemetry):
   if len(pipeline_funcs) == 0:
     raise ValueError('A function with @dsl.pipeline decorator is required in the py file.')
 
@@ -68,7 +82,8 @@ def _compile_pipeline_function(pipeline_funcs, function_name, output_path, type_
   else:
     pipeline_func = pipeline_funcs[0]
 
-  kfp.compiler.Compiler().compile(pipeline_func, output_path, type_check)
+  kfp.compiler.Compiler().compile(
+      pipeline_func, output_path, type_check, allow_telemetry=allow_telemetry)
 
 
 class PipelineCollectorContext():
@@ -90,26 +105,31 @@ class PipelineCollectorContext():
     Please switch to compiling pipeline files or functions.
     If you use this feature please create an issue in https://github.com/kubeflow/pipelines/issues .'''
 )
-def compile_package(package_path, namespace, function_name, output_path, type_check):
+def compile_package(
+    package_path, namespace, function_name, output_path, type_check, allow_telemetry):
   tmpdir = tempfile.mkdtemp()
   sys.path.insert(0, tmpdir)
   try:
     subprocess.check_call(['python3', '-m', 'pip', 'install', package_path, '-t', tmpdir])
     with PipelineCollectorContext() as pipeline_funcs:
       __import__(namespace)
-    _compile_pipeline_function(pipeline_funcs, function_name, output_path, type_check)
+    _compile_pipeline_function(
+        pipeline_funcs, function_name, output_path, type_check,
+        allow_telemetry=allow_telemetry)
   finally:
     del sys.path[0]
     shutil.rmtree(tmpdir)
 
 
-def compile_pyfile(pyfile, function_name, output_path, type_check):
+def compile_pyfile(pyfile, function_name, output_path, type_check, allow_telemetry):
   sys.path.insert(0, os.path.dirname(pyfile))
   try:
     filename = os.path.basename(pyfile)
     with PipelineCollectorContext() as pipeline_funcs:
       __import__(os.path.splitext(filename)[0])
-    _compile_pipeline_function(pipeline_funcs, function_name, output_path, type_check)
+    _compile_pipeline_function(
+        pipeline_funcs, function_name, output_path, type_check,
+        allow_telemetry=allow_telemetry)
   finally:
     del sys.path[0]
 
@@ -120,9 +140,22 @@ def main():
       (args.py is not None and args.package is not None)):
     raise ValueError('Either --py or --package is needed but not both.')
   if args.py:
-    compile_pyfile(args.py, args.function, args.output, not args.disable_type_check)
+    compile_pyfile(
+        args.py,
+        args.function,
+        args.output,
+        not args.disable_type_check,
+        not args.disable_telemetry
+    )
   else:
     if args.namespace is None:
       raise ValueError('--namespace is required for compiling packages.')
-    compile_package(args.package, args.namespace, args.function, args.output, not args.disable_type_check)
+    compile_package(
+        args.package,
+        args.namespace,
+        args.function,
+        args.output,
+        not args.disable_type_check,
+        not args.disable_telemetry
+    )
   

sdk/python/tests/compiler/testdata/add_pod_env.yaml

@@ -28,6 +28,7 @@ spec:
       image: library/bash
     metadata:
       labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
         add-pod-env: 'true'
     name: echo
   - dag:

sdk/python/tests/compiler/testdata/basic.yaml

@@ -57,6 +57,7 @@ spec:
       - -c
       image: python:3.5-jessie
     name: exiting
+    metadata: {'labels': {'pipelines.kubeflow.org/pipeline-sdk-type': 'kfp'}}
   - container:
       args:
       - python -c "from collections import Counter; words = Counter('{{inputs.parameters.message}}'.split());
@@ -72,6 +73,9 @@ spec:
       parameters:
       - name: message
     name: get-frequent
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     outputs:
       artifacts:
       - name: get-frequent-word
@@ -98,6 +102,8 @@ spec:
       - name: get-frequent-word
       - name: outputpath
     metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
       annotations:
         tf-version.cloud-tpus.google.com: "1.12"
     name: save

sdk/python/tests/compiler/testdata/basic_no_decorator.yaml

@@ -58,6 +58,9 @@ spec:
           - -c
         image: python:3.5-jessie
       name: exiting
+      metadata:
+        labels:
+          pipelines.kubeflow.org/pipeline-sdk-type: kfp
     - container:
         args:
           - python -c "from collections import Counter; words = Counter('{{inputs.parameters.message}}'.split());
@@ -73,6 +76,9 @@ spec:
         parameters:
           - name: message
       name: get-frequent
+      metadata:
+        labels:
+          pipelines.kubeflow.org/pipeline-sdk-type: kfp
       outputs:
         artifacts:
           - name: get-frequent-word
@@ -99,6 +105,8 @@ spec:
           - name: get-frequent-word
           - name: outputpath
       metadata:
+        labels:
+          pipelines.kubeflow.org/pipeline-sdk-type: kfp
         annotations:
           tf-version.cloud-tpus.google.com: "1.12"
       name: save

sdk/python/tests/compiler/testdata/coin.yaml

@@ -75,6 +75,9 @@ spec:
       - -c
       image: python:alpine3.6
     name: flip
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     outputs:
       artifacts:
       - name: flip-output
@@ -92,6 +95,9 @@ spec:
       - -c
       image: python:alpine3.6
     name: flip-again
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     outputs:
       artifacts:
       - name: flip-again-output
@@ -129,6 +135,9 @@ spec:
       parameters:
       - name: flip-again-output
     name: print1
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
   - container:
       command:
       - echo
@@ -138,3 +147,6 @@ spec:
       parameters:
       - name: flip-again-output
     name: print2
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp

sdk/python/tests/compiler/testdata/compose.yaml

@@ -36,6 +36,9 @@ spec:
       parameters:
       - name: url
     name: download
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     outputs:
       artifacts:
       - name: download-downloaded
@@ -87,6 +90,9 @@ spec:
       parameters:
       - name: download-downloaded
     name: get-frequent
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     outputs:
       artifacts:
       - name: get-frequent-word
@@ -108,3 +114,6 @@ spec:
       - name: get-frequent-word
       - name: outputpath
     name: save
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp

sdk/python/tests/compiler/testdata/default_value.yaml

@@ -56,6 +56,9 @@ spec:
       parameters:
       - name: url
     name: download
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     outputs:
       artifacts:
       - name: download-downloaded
@@ -75,3 +78,6 @@ spec:
       parameters:
       - name: download-downloaded
     name: echo
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp

sdk/python/tests/compiler/testdata/imagepullsecrets.yaml

@@ -25,6 +25,9 @@ spec:
       parameters:
       - name: message
     name: get-frequent
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     outputs:
       artifacts:
       - name: get-frequent-word

sdk/python/tests/compiler/testdata/input_artifact_raw_value.yaml

@@ -22,6 +22,9 @@ spec:
         raw:
           data: Constant artifact value
     name: component-with-inline-input-artifact
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
   - container:
       command:
       - cat
@@ -34,6 +37,9 @@ spec:
         raw:
           data: Constant artifact value
     name: component-with-input-artifact
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
   - container:
       command:
       - cat
@@ -46,6 +52,9 @@ spec:
         raw:
           data: hard-coded artifact value
     name: component-with-input-artifact-2
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
   - container:
       command:
       - cat
@@ -58,6 +67,9 @@ spec:
         raw:
           data: Text from a file with hard-coded artifact value
     name: component-with-input-artifact-3
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
   - dag:
       tasks:
       - name: component-with-inline-input-artifact

sdk/python/tests/compiler/testdata/loop_over_lightweight_output.yaml

@@ -44,6 +44,8 @@
           - "name": |-
               produce-list-data_list-loop-item
       "metadata":
+        labels:
+          pipelines.kubeflow.org/pipeline-sdk-type: kfp
         "annotations":
           "pipelines.kubeflow.org/component_spec": |-
             {"inputs": [{"name": "data"}], "name": "Consume data"}
@@ -104,6 +106,8 @@
         "image": |-
           busybox
       "metadata":
+        labels:
+          pipelines.kubeflow.org/pipeline-sdk-type: kfp
         "annotations":
           "pipelines.kubeflow.org/component_spec": |-
             {"name": "Produce list", "outputs": [{"name": "data_list"}]}

sdk/python/tests/compiler/testdata/parallelfor_item_argument_resolving.yaml

@@ -48,7 +48,9 @@ spec:
         parameters: 
           - 
             name: produce-list-of-strings-output
-      metadata: 
+      metadata:
+        labels:
+          pipelines.kubeflow.org/pipeline-sdk-type: kfp
         annotations: 
           pipelines.kubeflow.org/component_spec: "{\"inputs\": [{\"name\": \"param1\"}], \"name\": \"Consume\"}"
       name: consume
@@ -90,7 +92,9 @@ spec:
         parameters: 
           - 
             name: produce-list-of-strings-output-loop-item
-      metadata: 
+      metadata:
+        labels:
+          pipelines.kubeflow.org/pipeline-sdk-type: kfp
         annotations: 
           pipelines.kubeflow.org/component_spec: "{\"inputs\": [{\"name\": \"param1\"}], \"name\": \"Consume\"}"
       name: consume-2
@@ -132,7 +136,9 @@ spec:
         parameters: 
           - 
             name: produce-str-output
-      metadata: 
+      metadata:
+        labels:
+          pipelines.kubeflow.org/pipeline-sdk-type: kfp
         annotations: 
           pipelines.kubeflow.org/component_spec: "{\"inputs\": [{\"name\": \"param1\"}], \"name\": \"Consume\"}"
       name: consume-3
@@ -174,7 +180,9 @@ spec:
         parameters: 
           - 
             name: produce-list-of-ints-output
-      metadata: 
+      metadata:
+        labels:
+          pipelines.kubeflow.org/pipeline-sdk-type: kfp
         annotations: 
           pipelines.kubeflow.org/component_spec: "{\"inputs\": [{\"name\": \"param1\"}], \"name\": \"Consume\"}"
       name: consume-4
@@ -216,7 +224,9 @@ spec:
         parameters: 
           - 
             name: produce-list-of-ints-output-loop-item
-      metadata: 
+      metadata:
+        labels:
+          pipelines.kubeflow.org/pipeline-sdk-type: kfp
         annotations: 
           pipelines.kubeflow.org/component_spec: "{\"inputs\": [{\"name\": \"param1\"}], \"name\": \"Consume\"}"
       name: consume-5
@@ -258,7 +268,9 @@ spec:
         parameters: 
           - 
             name: produce-list-of-dicts-output
-      metadata: 
+      metadata:
+        labels:
+          pipelines.kubeflow.org/pipeline-sdk-type: kfp
         annotations: 
           pipelines.kubeflow.org/component_spec: "{\"inputs\": [{\"name\": \"param1\"}], \"name\": \"Consume\"}"
       name: consume-6
@@ -300,7 +312,9 @@ spec:
         parameters: 
           - 
             name: produce-list-of-dicts-output-loop-item-subvar-aaa
-      metadata: 
+      metadata:
+        labels:
+          pipelines.kubeflow.org/pipeline-sdk-type: kfp
         annotations: 
           pipelines.kubeflow.org/component_spec: "{\"inputs\": [{\"name\": \"param1\"}], \"name\": \"Consume\"}"
       name: consume-7
@@ -502,7 +516,9 @@ spec:
                   with open(output_file, 'w') as f:
                       f.write(_output_serializers[idx](_outputs[idx]))
         image: "tensorflow/tensorflow:1.13.2-py3"
-      metadata: 
+      metadata:
+        labels:
+          pipelines.kubeflow.org/pipeline-sdk-type: kfp
         annotations: 
           pipelines.kubeflow.org/component_spec: "{\"name\": \"Produce list of dicts\", \"outputs\": [{\"name\": \"Output\", \"type\": \"JsonArray\"}]}"
       name: produce-list-of-dicts
@@ -564,7 +580,9 @@ spec:
                   with open(output_file, 'w') as f:
                       f.write(_output_serializers[idx](_outputs[idx]))
         image: "tensorflow/tensorflow:1.13.2-py3"
-      metadata: 
+      metadata:
+        labels:
+          pipelines.kubeflow.org/pipeline-sdk-type: kfp
         annotations: 
           pipelines.kubeflow.org/component_spec: "{\"name\": \"Produce list of ints\", \"outputs\": [{\"name\": \"Output\", \"type\": \"JsonArray\"}]}"
       name: produce-list-of-ints
@@ -626,7 +644,9 @@ spec:
                   with open(output_file, 'w') as f:
                       f.write(_output_serializers[idx](_outputs[idx]))
         image: "tensorflow/tensorflow:1.13.2-py3"
-      metadata: 
+      metadata:
+        labels:
+          pipelines.kubeflow.org/pipeline-sdk-type: kfp
         annotations: 
           pipelines.kubeflow.org/component_spec: "{\"name\": \"Produce list of strings\", \"outputs\": [{\"name\": \"Output\", \"type\": \"JsonArray\"}]}"
       name: produce-list-of-strings
@@ -682,7 +702,9 @@ spec:
                   with open(output_file, 'w') as f:
                       f.write(_output_serializers[idx](_outputs[idx]))
         image: "tensorflow/tensorflow:1.13.2-py3"
-      metadata: 
+      metadata:
+        labels:
+          pipelines.kubeflow.org/pipeline-sdk-type: kfp
         annotations: 
           pipelines.kubeflow.org/component_spec: "{\"name\": \"Produce str\", \"outputs\": [{\"name\": \"Output\", \"type\": \"String\"}]}"
       name: produce-str

sdk/python/tests/compiler/testdata/param_op_transform.yaml

@@ -21,6 +21,7 @@ spec:
       - name: param
     metadata:
       labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
         param: '{{inputs.parameters.param}}'
     name: cop
   - dag:

sdk/python/tests/compiler/testdata/param_substitutions.yaml

@@ -23,11 +23,17 @@ spec:
       parameters:
       - name: create-volume-name
     name: cop
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     volumes:
     - name: create-volume
       persistentVolumeClaim:
         claimName: '{{inputs.parameters.create-volume-name}}'
   - name: create-volume
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     outputs:
       parameters:
       - name: create-volume-manifest

sdk/python/tests/compiler/testdata/pipelineparams.yaml

@@ -29,6 +29,9 @@ spec:
       value: '10'
   templates:
   - name: download
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     inputs:
       parameters:
       - name: sleep_ms
@@ -54,6 +57,9 @@ spec:
       args:
       - -text="hello world"
   - name: echo
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     inputs:
       parameters:
       - name: download-downloaded

sdk/python/tests/compiler/testdata/preemptible_tpu_gpu.yaml

@@ -32,6 +32,9 @@ spec:
           limits:
             nvidia.com/gpu: 1
       name: flip
+      metadata:
+        labels:
+          pipelines.kubeflow.org/pipeline-sdk-type: kfp
       outputs:
         artifacts:
           - name: flip-output

sdk/python/tests/compiler/testdata/recursive_do_while.yaml

@@ -32,6 +32,9 @@ spec:
       - -c
       image: python:alpine3.6
     name: flip
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     outputs:
       artifacts:
       - name: flip-output
@@ -49,6 +52,9 @@ spec:
       - -c
       image: python:alpine3.6
     name: flip-2
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     outputs:
       artifacts:
       - name: flip-2-output
@@ -62,6 +68,9 @@ spec:
       - -c
       image: python:alpine3.6
     name: flip-3
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     outputs:
       artifacts:
       - name: flip-3-output
@@ -129,6 +138,9 @@ spec:
       parameters:
       - name: flip-output
     name: print
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
   - container:
       command:
       - echo
@@ -138,3 +150,6 @@ spec:
       parameters:
       - name: flip-output
     name: print-2
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp

sdk/python/tests/compiler/testdata/recursive_while.yaml

@@ -46,6 +46,9 @@ spec:
       - -c
       image: python:alpine3.6
     name: flip
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     outputs:
       artifacts:
       - name: flip-output
@@ -63,6 +66,9 @@ spec:
       - -c
       image: python:alpine3.6
     name: flip-2
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     outputs:
       artifacts:
       - name: flip-2-output
@@ -76,6 +82,9 @@ spec:
       - -c
       image: python:alpine3.6
     name: flip-3
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     outputs:
       artifacts:
       - name: flip-3-output
@@ -131,6 +140,9 @@ spec:
       parameters:
       - name: flip-output
     name: print
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
   - container:
       command:
       - echo
@@ -140,3 +152,6 @@ spec:
       parameters:
       - name: flip-output
     name: print-2
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp

sdk/python/tests/compiler/testdata/resourceop_basic.yaml

@@ -27,6 +27,9 @@ spec:
       parameters:
       - name: create-my-secret-name
     name: cop
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     volumes:
     - name: my-secret
       secret:
@@ -36,6 +39,9 @@ spec:
       - name: password
       - name: username
     name: create-my-secret
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     outputs:
       parameters:
       - name: create-my-secret-manifest

sdk/python/tests/compiler/testdata/sidecar.yaml

@@ -43,7 +43,13 @@ spec:
       command:
       - sh
       - "-c"
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
   - name: echo
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     inputs:
       parameters:
       - name: download-downloaded

sdk/python/tests/compiler/testdata/timeout.yaml

@@ -30,6 +30,9 @@ spec:
       - -c
       image: python:alpine3.6
     name: random-failure
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
   - container:
       args:
       - import random; import sys; exit_code = random.choice([0,1]); print(exit_code);
@@ -39,3 +42,6 @@ spec:
       - -c
       image: python:alpine3.6
     name: random-failure-2
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp

sdk/python/tests/compiler/testdata/volume.yaml

@@ -40,6 +40,9 @@ spec:
       - mountPath: /secret/gcp-credentials
         name: gcp-credentials
     name: download
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     outputs:
       artifacts:
       - name: download-downloaded
@@ -63,6 +66,9 @@ spec:
       parameters:
       - name: download-downloaded
     name: echo
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
   - dag:
       tasks:
       - name: download

sdk/python/tests/compiler/testdata/volume_snapshotop_rokurl.yaml

@@ -16,6 +16,9 @@ spec:
       parameters:
       - name: create-volume-1-name
     name: create-snapshot-1
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     outputs:
       parameters:
       - name: create-snapshot-1-manifest
@@ -37,6 +40,9 @@ spec:
       parameters:
       - name: create-volume-2-name
     name: create-snapshot-2
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     outputs:
       parameters:
       - name: create-snapshot-2-manifest
@@ -58,6 +64,9 @@ spec:
       parameters:
       - name: rok_url
     name: create-volume-1
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     outputs:
       parameters:
       - name: create-volume-1-manifest
@@ -80,6 +89,9 @@ spec:
       - name: create-snapshot-1-name
       - name: create-snapshot-1-size
     name: create-volume-2
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     outputs:
       parameters:
       - name: create-volume-2-manifest
@@ -102,6 +114,9 @@ spec:
       - name: create-snapshot-2-name
       - name: create-snapshot-2-size
     name: create-volume-3
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     outputs:
       parameters:
       - name: create-volume-3-manifest
@@ -133,6 +148,9 @@ spec:
       parameters:
       - name: create-volume-1-name
     name: step1-concat
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     volumes:
     - name: create-volume-1
       persistentVolumeClaim:
@@ -150,6 +168,9 @@ spec:
       parameters:
       - name: create-volume-2-name
     name: step2-gunzip
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     volumes:
     - name: create-volume-2
       persistentVolumeClaim:
@@ -166,6 +187,9 @@ spec:
       parameters:
       - name: create-volume-3-name
     name: step3-output
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     volumes:
     - name: create-volume-3
       persistentVolumeClaim:

sdk/python/tests/compiler/testdata/volume_snapshotop_sequential.yaml

@@ -13,6 +13,9 @@ spec:
   serviceAccountName: pipeline-runner
   templates:
   - name: create-volume
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     outputs:
       parameters:
       - name: create-volume-manifest
@@ -44,6 +47,9 @@ spec:
       - name: create-volume-name
       - name: url
     name: step1-ingest
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     volumes:
     - name: create-volume
       persistentVolumeClaim:
@@ -52,6 +58,9 @@ spec:
       parameters:
       - name: create-volume-name
     name: step1-snap
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     outputs:
       parameters:
       - name: step1-snap-manifest
@@ -83,6 +92,9 @@ spec:
       parameters:
       - name: create-volume-name
     name: step2-gunzip
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     volumes:
     - name: create-volume
       persistentVolumeClaim:
@@ -91,6 +103,9 @@ spec:
       parameters:
       - name: create-volume-name
     name: step2-snap
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     outputs:
       parameters:
       - name: step2-snap-manifest
@@ -122,6 +137,9 @@ spec:
       parameters:
       - name: create-volume-name
     name: step3-copy
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     volumes:
     - name: create-volume
       persistentVolumeClaim:
@@ -130,6 +148,9 @@ spec:
       parameters:
       - name: create-volume-name
     name: step3-snap
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     outputs:
       parameters:
       - name: step3-snap-manifest
@@ -160,6 +181,9 @@ spec:
       parameters:
       - name: create-volume-name
     name: step4-output
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     volumes:
     - name: create-volume
       persistentVolumeClaim:

sdk/python/tests/compiler/testdata/volumeop_basic.yaml

@@ -26,6 +26,9 @@ spec:
       parameters:
       - name: create-pvc-name
     name: cop
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     volumes:
     - name: create-pvc
       persistentVolumeClaim:
@@ -34,6 +37,9 @@ spec:
       parameters:
       - name: size
     name: create-pvc
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     outputs:
       parameters:
       - name: create-pvc-manifest

sdk/python/tests/compiler/testdata/volumeop_dag.yaml

@@ -12,6 +12,9 @@ spec:
   serviceAccountName: pipeline-runner
   templates:
   - name: create-pvc
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     outputs:
       parameters:
       - name: create-pvc-manifest
@@ -42,6 +45,9 @@ spec:
       parameters:
       - name: create-pvc-name
     name: step1
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     volumes:
     - name: create-pvc
       persistentVolumeClaim:
@@ -60,6 +66,9 @@ spec:
       parameters:
       - name: create-pvc-name
     name: step2
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     volumes:
     - name: create-pvc
       persistentVolumeClaim:
@@ -78,6 +87,9 @@ spec:
       parameters:
       - name: create-pvc-name
     name: step3
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     volumes:
     - name: create-pvc
       persistentVolumeClaim:

sdk/python/tests/compiler/testdata/volumeop_parallel.yaml

@@ -12,6 +12,9 @@ spec:
   serviceAccountName: pipeline-runner
   templates:
   - name: create-pvc
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     outputs:
       parameters:
       - name: create-pvc-manifest
@@ -42,6 +45,9 @@ spec:
       parameters:
       - name: create-pvc-name
     name: step1
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     volumes:
     - name: create-pvc
       persistentVolumeClaim:
@@ -60,6 +66,9 @@ spec:
       parameters:
       - name: create-pvc-name
     name: step2
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     volumes:
     - name: create-pvc
       persistentVolumeClaim:
@@ -78,6 +87,9 @@ spec:
       parameters:
       - name: create-pvc-name
     name: step3
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     volumes:
     - name: create-pvc
       persistentVolumeClaim:

sdk/python/tests/compiler/testdata/volumeop_sequential.yaml

@@ -12,6 +12,9 @@ spec:
   serviceAccountName: pipeline-runner
   templates:
   - name: mypvc
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     outputs:
       parameters:
       - name: mypvc-manifest
@@ -42,6 +45,9 @@ spec:
       parameters:
       - name: mypvc-name
     name: step1
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     volumes:
     - name: mypvc
       persistentVolumeClaim:
@@ -60,6 +66,9 @@ spec:
       parameters:
       - name: mypvc-name
     name: step2
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     volumes:
     - name: mypvc
       persistentVolumeClaim:
@@ -77,6 +86,9 @@ spec:
       parameters:
       - name: mypvc-name
     name: step3
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     volumes:
     - name: mypvc
       persistentVolumeClaim:

sdk/python/tests/compiler/testdata/withitem_basic.yaml

@@ -47,6 +47,9 @@ spec:
       - name: loop-item-param-00000001-subvar-a
       - name: my_pipe_param
     name: my-in-coop1
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
   - container:
       args:
       - echo op2 {{inputs.parameters.loop-item-param-00000001-subvar-b}}
@@ -58,6 +61,9 @@ spec:
       parameters:
       - name: loop-item-param-00000001-subvar-b
     name: my-in-coop2
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
   - container:
       args:
       - echo {{inputs.parameters.my_pipe_param}}
@@ -69,6 +75,9 @@ spec:
       parameters:
       - name: my_pipe_param
     name: my-out-cop
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
   - dag:
       tasks:
       - arguments:

sdk/python/tests/compiler/testdata/withitem_nested.yaml

@@ -79,6 +79,9 @@ spec:
       - name: loop-item-param-00000001-subvar-a
       - name: my_pipe_param
     name: my-in-coop1
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
   - container:
       args:
       - echo op2 {{inputs.parameters.loop-item-param-00000001-subvar-b}}
@@ -90,6 +93,9 @@ spec:
       parameters:
       - name: loop-item-param-00000001-subvar-b
     name: my-in-coop2
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
   - container:
       args:
       - echo op1 {{inputs.parameters.loop-item-param-00000001-subvar-a}} {{inputs.parameters.loop-item-param-00000002}}
@@ -104,6 +110,9 @@ spec:
       - name: loop-item-param-00000002
       - name: my_pipe_param
     name: my-inner-inner-coop
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
   - container:
       args:
       - echo {{inputs.parameters.my_pipe_param}}
@@ -115,6 +124,9 @@ spec:
       parameters:
       - name: my_pipe_param
     name: my-out-cop
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
   - dag:
       tasks:
       - arguments:

sdk/python/tests/compiler/testdata/withparam_global.yaml

@@ -36,6 +36,9 @@ spec:
       parameters:
       - name: loopidy_doop-loop-item
     name: my-in-cop1
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
   - container:
       args:
       - python -c "import json; import sys; json.dump([i for i in range(20, 31)],
@@ -45,6 +48,9 @@ spec:
       - -c
       image: python:alpine3.6
     name: my-out-cop0
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     outputs:
       artifacts:
       - name: my-out-cop0-out
@@ -64,6 +70,9 @@ spec:
       parameters:
       - name: my-out-cop0-out
     name: my-out-cop2
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
   - dag:
       tasks:
       - arguments:

sdk/python/tests/compiler/testdata/withparam_global_dict.yaml

@@ -36,6 +36,9 @@ spec:
       parameters:
       - name: loopidy_doop-loop-item-subvar-a
     name: my-in-cop1
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
   - container:
       args:
       - python -c "import json; import sys; json.dump([i for i in range(20, 31)],
@@ -45,6 +48,9 @@ spec:
       - -c
       image: python:alpine3.6
     name: my-out-cop0
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     outputs:
       artifacts:
       - name: my-out-cop0-out
@@ -64,6 +70,9 @@ spec:
       parameters:
       - name: my-out-cop0-out
     name: my-out-cop2
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
   - dag:
       tasks:
       - arguments:

sdk/python/tests/compiler/testdata/withparam_output.yaml

@@ -33,6 +33,9 @@ spec:
       parameters:
       - name: my-out-cop0-out-loop-item
     name: my-in-cop1
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
   - container:
       args:
       - python -c "import json; import sys; json.dump([i for i in range(20, 31)],
@@ -42,6 +45,9 @@ spec:
       - -c
       image: python:alpine3.6
     name: my-out-cop0
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     outputs:
       artifacts:
       - name: my-out-cop0-out
@@ -61,6 +67,9 @@ spec:
       parameters:
       - name: my-out-cop0-out
     name: my-out-cop2
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
   - dag:
       tasks:
       - arguments:

sdk/python/tests/compiler/testdata/withparam_output_dict.yaml

@@ -33,6 +33,9 @@ spec:
       parameters:
       - name: my-out-cop0-out-loop-item-subvar-a
     name: my-in-cop1
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
   - container:
       args:
       - 'python -c "import json; import sys; json.dump([{''a'': 1, ''b'': 2}, {''a'':
@@ -42,6 +45,9 @@ spec:
       - -c
       image: python:alpine3.6
     name: my-out-cop0
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
     outputs:
       artifacts:
       - name: my-out-cop0-out
@@ -61,6 +67,9 @@ spec:
       parameters:
       - name: my-out-cop0-out
     name: my-out-cop2
+    metadata:
+      labels:
+        pipelines.kubeflow.org/pipeline-sdk-type: kfp
   - dag:
       tasks:
       - arguments: