feat(sdk): Add alpha env flag to opt out artifact tracking annotations (kubeflow#1032)

* add alpha env flag to opt out artifact tracking annotations

* reduce annotations further

Author: Tomcli

sdk/python/kfp_tekton/compiler/_data_passing_rewriter.py

@@ -550,11 +550,12 @@ def big_data_passing_tasks(prname: str, task: dict, pipelinerun_template: dict,
                     appended_taskrun_name = True
             task['taskSpec'] = replace_big_data_placeholder(
                 task.get("taskSpec", {}), placeholder, workspaces_parameter)
-            artifact_items = pipelinerun_template['metadata']['annotations']['tekton.dev/artifact_items']
-            artifact_items[task['name']] = replace_big_data_placeholder(
-                artifact_items[task['name']], placeholder, workspaces_parameter)
-            pipelinerun_template['metadata']['annotations']['tekton.dev/artifact_items'] = \
-                artifact_items
+            artifact_items = pipelinerun_template['metadata']['annotations'].get('tekton.dev/artifact_items', '{}')
+            if artifact_items != '{}':
+                artifact_items[task['name']] = replace_big_data_placeholder(
+                    artifact_items[task['name']], placeholder, workspaces_parameter)
+                pipelinerun_template['metadata']['annotations']['tekton.dev/artifact_items'] = \
+                    artifact_items
     if appended_taskrun_path_step:
         task['taskSpec']['steps'].append(appended_taskrun_path_step)
         _append_original_pr_name_env(task)
@@ -651,7 +652,7 @@ def append_taskrun_params(task_name_append: str, task_path_name: str):
                         task = input_artifacts_tasks_pr_params(task, task_artifact)
 
     # If a task produces a result and artifact, add a step to copy artifact to results.
-    artifact_items = pipelinerun_template['metadata']['annotations']['tekton.dev/artifact_items']
+    artifact_items = pipelinerun_template['metadata']['annotations'].get('tekton.dev/artifact_items', '{}')
     add_copy_results_artifacts_step = False
     if task.get("taskSpec", {}):
         if task_spec.get('results', []):
@@ -797,16 +798,18 @@ def clean_up_empty_workflow_structures(workflow: list):
 
 
 def load_annotations(template: dict):
-    artifact_items = json.loads(
-        str(template['metadata']['annotations']['tekton.dev/artifact_items']))
-    template['metadata']['annotations']['tekton.dev/artifact_items'] = \
-        artifact_items
+    if env.get('DISABLE_ARTIFACT_TRACKING', 'false').lower() != 'true':
+        artifact_items = json.loads(
+            str(template['metadata']['annotations'].get('tekton.dev/artifact_items', '{}')))
+        template['metadata']['annotations']['tekton.dev/artifact_items'] = \
+            artifact_items
     return template
 
 
 def jsonify_annotations(template: dict):
-    template['metadata']['annotations']['tekton.dev/artifact_items'] = \
-        json.dumps(template['metadata']['annotations']['tekton.dev/artifact_items'])
+    if env.get('DISABLE_ARTIFACT_TRACKING', 'false').lower() != 'true':
+        template['metadata']['annotations']['tekton.dev/artifact_items'] = \
+            json.dumps(template['metadata']['annotations'].get('tekton.dev/artifact_items', '{}'))
     return template
 
 

sdk/python/kfp_tekton/compiler/_op_to_template.py

@@ -20,6 +20,7 @@
 
 from collections import OrderedDict
 from typing import List, Text, Dict, Any
+from os import environ as env
 
 from kfp import dsl
 from kfp_tekton.compiler._k8s_helper import convert_k8s_obj_to_json, sanitize_k8s_name
@@ -564,6 +565,8 @@ def _op_to_template(op: BaseOp,
         }
         template.setdefault('metadata', {}).setdefault('annotations', {})['pipelines.kubeflow.org/component_spec_digest'] = \
             json.dumps(digested_component_spec_dict, sort_keys=True)
+        if env.get('DISABLE_ARTIFACT_TRACKING', 'false').lower() == 'true':
+            template['metadata']['annotations'].pop('pipelines.kubeflow.org/component_spec_digest', None)
 
     if isinstance(op, dsl.ContainerOp) and op.execution_options:
         if op.execution_options.caching_strategy.max_cache_staleness:

sdk/python/kfp_tekton/compiler/compiler.py

@@ -1330,6 +1330,11 @@ def get_when_task(input_task_when, depended_conditions):
       }
     }
 
+    if env.get('DISABLE_ARTIFACT_TRACKING', 'false').lower() == 'true':
+      pipeline_run['metadata']['annotations'].pop('tekton.dev/output_artifacts', None)
+      pipeline_run['metadata']['annotations'].pop('tekton.dev/input_artifacts', None)
+      pipeline_run['metadata']['annotations'].pop('tekton.dev/artifact_items', None)
+
     if self.pipeline_labels:
       pipeline_run['metadata']['labels'] = pipeline_run['metadata'].setdefault('labels', {})
       pipeline_run['metadata']['labels'].update(self.pipeline_labels)

sdk/python/tests/compiler/compiler_tests.py

@@ -939,10 +939,11 @@ def test_artifacts_of_ops_with_long_names(self):
           obj = yaml.safe_load(f)
         text = yaml.safe_dump(obj)
         self.assertNotRegex(text, "stepTemplate")
-        artifact_items = obj['metadata']['annotations']['tekton.dev/artifact_items']
-        items_for_op0 = list(json.loads(artifact_items).values())[0]
-        names_of_items_for_op0 = set([item[0] for item in items_for_op0])
-        self.assertSetEqual(names_of_items_for_op0, {"incr_i", "sq_i"})
+        artifact_items = obj['metadata']['annotations'].get('tekton.dev/artifact_items', {})
+        if artifact_items:
+          items_for_op0 = list(json.loads(artifact_items).values())[0]
+          names_of_items_for_op0 = set([item[0] for item in items_for_op0])
+          self.assertSetEqual(names_of_items_for_op0, {"incr_i", "sq_i"})
     finally:
       shutil.rmtree(temp_dir)