feat(sdk): DSL - Added support for volatile components (#4104)

Volatile components do not reuse the cached results by default.
The pipeline authors can re-enable cache reuse if they want.

components/diagnostics/diagnose_me/component.yaml

@@ -23,6 +23,9 @@ description: |-
         Raises:
             RuntimeError: If configuration is not setup properly and
             HALT_ON_ERROR flag is set.
+metadata:
+  annotations:
+    volatile_component: "true"
 inputs:
 - name: bucket
   type: String

components/git/clone/component.yaml

@@ -1,5 +1,8 @@
 name: Git clone
 description: Creates a shallow clone of the specified repo branch
+metadata:
+  annotations:
+    volatile_component: "true"
 inputs:
 - {name: Repo URI, type: URI}
 - {name: Branch, type: String, default: master}

components/google-cloud/storage/list/component.yaml

@@ -3,6 +3,9 @@ inputs:
 - {name: GCS path, type: URI, description: 'GCS path for listing. For recursive listing use the "gs://bucket/path/**" syntax".'}
 outputs:
 - {name: Paths}
+metadata:
+    annotations:
+        volatile_component: 'true'
 implementation:
     container:
         image: google/cloud-sdk

sdk/python/kfp/dsl/_component_bridge.py

@@ -84,9 +84,13 @@ def _create_container_op_from_component_and_arguments(
             task.container.add_env_variable(k8s_client.V1EnvVar(name=name, value=value))
 
     if component_spec.metadata:
-        for key, value in (component_spec.metadata.annotations or {}).items():
+        annotations = component_spec.metadata.annotations or {}
+        for key, value in annotations.items():
             task.add_pod_annotation(key, value)
         for key, value in (component_spec.metadata.labels or {}).items():
             task.add_pod_label(key, value)
+        # Disabling the caching for the volatile components by default
+        if annotations.get('volatile_component', 'false') == 'true':
+            task.execution_options.caching_strategy.max_cache_staleness = 'P0D'
 
     return task

sdk/python/tests/dsl/component_bridge_tests.py

@@ -96,6 +96,21 @@ def test_passing_component_metadata_to_container_op(self):
         self.assertEqual(task1.pod_annotations['key1'], 'value1')
         self.assertEqual(task1.pod_labels['key1'], 'value1')
 
+    def test_volatile_components(self):
+        component_text = textwrap.dedent('''\
+            metadata:
+                annotations:
+                    volatile_component: "true"
+            implementation:
+                container:
+                    image: busybox
+            '''
+        )
+        task_factory1 = load_component_from_text(text=component_text)
+
+        task1 = task_factory1()
+        self.assertEqual(task1.execution_options.caching_strategy.max_cache_staleness, 'P0D')
+
     def test_type_compatibility_check_not_failing_when_disabled(self):
         component_a = textwrap.dedent('''\
             outputs: