feat(backend): add new annotation flags to enable artifact tracking at pipeline and task level (kubeflow#1065)

* add new annotation flags to enable artifact tracking at pipeline and task level

* fix artifact metrics bug for new tekton version

* fix path bug

* fix path bug

* fix path bug

* fix path

* fix feature flag

* add feature doc

Author: Tomcli

backend/src/apiserver/common/const.go

@@ -80,6 +80,8 @@ const (
 	ArtifactBucketAnnotation         string = "tekton.dev/artifact_bucket"
 	ArtifactEndpointAnnotation       string = "tekton.dev/artifact_endpoint"
 	ArtifactEndpointSchemeAnnotation string = "tekton.dev/artifact_endpoint_scheme"
+	TrackArtifactAnnotation          string = "tekton.dev/track_artifact"
+	TrackStepArtifactAnnotation      string = "tekton.dev/track_step_artifact"
 )
 
 // For backward compatibility. Remove after 0.3 release

backend/src/apiserver/template/tekton_template.go

@@ -262,8 +262,17 @@ func (t *Tekton) injectArchivalStep(workflow util.Workflow, artifactItemsJSON ma
 		injectDefaultScript := common.IsInjectDefaultScript()
 		copyStepTemplate := common.GetCopyStepTemplate()
 
+		artifactAnnotation, exists := workflow.ObjectMeta.Annotations[common.TrackArtifactAnnotation]
+		if exists && strings.ToLower(artifactAnnotation) == "true" {
+			trackArtifacts = true
+		}
 		if task.TaskSpec != nil && task.TaskSpec.Steps != nil {
-			if (hasArtifacts && len(artifacts) > 0 && trackArtifacts) || archiveLogs || (hasArtifacts && len(artifacts) > 0 && stripEOF) {
+			injectStepArtifacts := false
+			stepArtifactAnnotation, exists := task.TaskSpec.Metadata.Annotations[common.TrackStepArtifactAnnotation]
+			if trackArtifacts || (exists && strings.ToLower(stepArtifactAnnotation) == "true") {
+				injectStepArtifacts = true
+			}
+			if (hasArtifacts && len(artifacts) > 0 && injectStepArtifacts) || archiveLogs || (hasArtifacts && len(artifacts) > 0 && stripEOF) {
 				artifactScript := common.GetArtifactScript()
 				if archiveLogs {
 					// Logging volumes
@@ -311,7 +320,7 @@ func (t *Tekton) injectArchivalStep(workflow util.Workflow, artifactItemsJSON ma
 
 				// Process the artifacts into minimum sh commands if running with minimum linux kernel
 				if injectDefaultScript {
-					artifactScript = t.injectDefaultScript(workflow, artifactScript, artifacts, hasArtifacts, archiveLogs, trackArtifacts, stripEOF, moveStep)
+					artifactScript = t.injectDefaultScript(workflow, artifactScript, artifacts, hasArtifacts, archiveLogs, injectStepArtifacts, stripEOF, moveStep)
 				}
 
 				// Define post-processing step
@@ -357,9 +366,14 @@ func (t *Tekton) injectDefaultScript(workflow util.Workflow, artifactScript stri
 	// Upload Artifacts if the artifact is enabled and the annoations are present
 	if hasArtifacts && len(artifacts) > 0 && trackArtifacts {
 		for _, artifact := range artifacts {
+			artifactBaseName := fmt.Sprintf("\"$(basename \"%s\")", artifact[1])
+			artifactPathName := fmt.Sprintf("\"%s/%s", common.GetPath4InternalResults(), artifactBaseName)
+			if artifact[0] == "mlpipeline-ui-metadata" || artifact[0] == "mlpipeline-metrics" {
+				artifactPathName = fmt.Sprintf("\"%s\"", artifact[1])
+			}
 			if len(artifact) == 2 {
-				artifactScript += fmt.Sprintf("push_artifact \"%s\" \"%s/\"$(basename \"%s\")\n",
-					artifact[0], common.GetPath4InternalResults(), artifact[1])
+				artifactScript += fmt.Sprintf("push_artifact \"%s\" %s\n",
+					artifact[0], artifactPathName)
 			} else {
 				glog.Warningf("Artifact annotations are missing for run %v.", workflow.Name)
 			}

samples/lightweight-component/calc_pipeline.py

@@ -72,6 +72,7 @@ def calc_pipeline(
     #Passing a task output reference as operation arguments
     #For an operation with a single return value, the output reference can be accessed using `task.output` or `task.outputs['output_name']` syntax
     divmod_task = divmod_op(add_task.output, b)
+    divmod_task.add_pod_annotation("tekton.dev/track_step_artifact", "true")
 
     #For an operation with a multiple return values, the output references can be accessed using `task.outputs['output_name']` syntax
     result_task = add_op(divmod_task.outputs['quotient'], c)

sdk/FEATURES.md

@@ -31,6 +31,7 @@ and test pipelines found in the KFP repository.
     - [Input Artifacts](#input-artifacts)
     - [Output Artifacts](#output-artifacts)
     - [Caching](#caching)
+    - [Enforce Artifact Tracking](#enforce-artifact-tracking)
   - [Features with Limitations](#features-with-limitations)
     - [Variable Substitutions](#variable-substitutions)
 
@@ -256,6 +257,24 @@ By default compiling a pipeline will add metadata annotations and labels so that
 
 The specific annotations and labels that are added to the task spec metadata to enable caching are: `annotations={'tekton.dev/template': ""}` and `labels={'pipelines.kubeflow.org/cache_enabled': 'true', 'pipelines.kubeflow.org/pipelinename': '', 'pipelines.kubeflow.org/generation': ''}`.
 
+### Enforce Artifact Tracking
+
+If you pipelines require artifact tracking in order to run, enforce artfact tracking to be always on for your pipelines using one of the two ways.
+
+Enable by adding a new pipeline annotation that enforce artifact tracking for the whole pipeline:
+```python
+import kfp_tekton
+from kfp_tekton.compiler import TektonCompiler
+pipeline_conf = kfp_tekton.compiler.pipeline_utils.TektonPipelineConf()
+pipeline_conf.add_pipeline_annotation("tekton.dev/track_artifact", 'true')
+TektonCompiler().compile(echo_pipeline, 'echo_pipeline.yaml', tekton_pipeline_conf=pipeline_conf)
+```
+
+Enable by adding a new task annotation that enforce artifact tracking for a specific task:
+```python
+task.add_pod_annotation("tekton.dev/track_step_artifact", "true")
+```
+
 ## Features with Limitations
 
 Below are the features that have certain limitation in the current Tekton release.