feat: Artifact GC Finalizer needs to be added if any Output Artifacts…

… have a strategy (#8856)

Signed-off-by: Julie Vogelman <julie_vogelman@intuit.com>

api/jsonschema/schema.json

@@ -8001,7 +8001,7 @@
         },
         "artifactGC": {
           "$ref": "#/definitions/io.argoproj.workflow.v1alpha1.ArtifactGC",
-          "description": "ArtifactGC describes the strategy to use when to deleting artifacts from completed or deleted workflows"
+          "description": "ArtifactGC describes the strategy to use when deleting artifacts from completed or deleted workflows (applies to all output Artifacts unless Artifact.ArtifactGC is specified, which overrides this)"
         },
         "artifactRepositoryRef": {
           "$ref": "#/definitions/io.argoproj.workflow.v1alpha1.ArtifactRepositoryRef",
@@ -8080,7 +8080,7 @@
         },
         "podGC": {
           "$ref": "#/definitions/io.argoproj.workflow.v1alpha1.PodGC",
-          "description": "PodGC describes the strategy to use when to deleting completed pods"
+          "description": "PodGC describes the strategy to use when deleting completed pods"
         },
         "podMetadata": {
           "$ref": "#/definitions/io.argoproj.workflow.v1alpha1.Metadata",
@@ -8158,7 +8158,7 @@
         },
         "volumeClaimGC": {
           "$ref": "#/definitions/io.argoproj.workflow.v1alpha1.VolumeClaimGC",
-          "description": "VolumeClaimGC describes the strategy to use when to deleting volumes from completed workflows"
+          "description": "VolumeClaimGC describes the strategy to use when deleting volumes from completed workflows"
         },
         "volumeClaimTemplates": {
           "description": "VolumeClaimTemplates is a list of claims that containers are allowed to reference. The Workflow controller will create the claims at the beginning of the workflow and delete the claims upon completion of the workflow",

api/openapi-spec/swagger.json

@@ -12436,7 +12436,7 @@
           "$ref": "#/definitions/io.argoproj.workflow.v1alpha1.Arguments"
         },
         "artifactGC": {
-          "description": "ArtifactGC describes the strategy to use when to deleting artifacts from completed or deleted workflows",
+          "description": "ArtifactGC describes the strategy to use when deleting artifacts from completed or deleted workflows (applies to all output Artifacts unless Artifact.ArtifactGC is specified, which overrides this)",
           "$ref": "#/definitions/io.argoproj.workflow.v1alpha1.ArtifactGC"
         },
         "artifactRepositoryRef": {
@@ -12515,7 +12515,7 @@
           "$ref": "#/definitions/io.k8s.api.policy.v1beta1.PodDisruptionBudgetSpec"
         },
         "podGC": {
-          "description": "PodGC describes the strategy to use when to deleting completed pods",
+          "description": "PodGC describes the strategy to use when deleting completed pods",
           "$ref": "#/definitions/io.argoproj.workflow.v1alpha1.PodGC"
         },
         "podMetadata": {
@@ -12593,7 +12593,7 @@
           "$ref": "#/definitions/io.argoproj.workflow.v1alpha1.TTLStrategy"
         },
         "volumeClaimGC": {
-          "description": "VolumeClaimGC describes the strategy to use when to deleting volumes from completed workflows",
+          "description": "VolumeClaimGC describes the strategy to use when deleting volumes from completed workflows",
           "$ref": "#/definitions/io.argoproj.workflow.v1alpha1.VolumeClaimGC"
         },
         "volumeClaimTemplates": {

docs/fields.md

@@ -777,7 +777,7 @@ WorkflowSpec is the specification of a Workflow.
 |`affinity`|[`Affinity`](#affinity)|Affinity sets the scheduling constraints for all pods in the io.argoproj.workflow.v1alpha1. Can be overridden by an affinity specified in the template|
 |`archiveLogs`|`boolean`|ArchiveLogs indicates if the container logs should be archived|
 |`arguments`|[`Arguments`](#arguments)|Arguments contain the parameters and artifacts sent to the workflow entrypoint Parameters are referencable globally using the 'workflow' variable prefix. e.g. {{io.argoproj.workflow.v1alpha1.parameters.myparam}}|
-|`artifactGC`|[`ArtifactGC`](#artifactgc)|ArtifactGC describes the strategy to use when to deleting artifacts from completed or deleted workflows|
+|`artifactGC`|[`ArtifactGC`](#artifactgc)|ArtifactGC describes the strategy to use when deleting artifacts from completed or deleted workflows (applies to all output Artifacts unless Artifact.ArtifactGC is specified, which overrides this)|
 |`artifactRepositoryRef`|[`ArtifactRepositoryRef`](#artifactrepositoryref)|ArtifactRepositoryRef specifies the configMap name and key containing the artifact repository config.|
 |`automountServiceAccountToken`|`boolean`|AutomountServiceAccountToken indicates whether a service account token should be automatically mounted in pods. ServiceAccountName of ExecutorConfig must be specified if this value is false.|
 |`dnsConfig`|[`PodDNSConfig`](#poddnsconfig)|PodDNSConfig defines the DNS parameters of a pod in addition to those generated from DNSPolicy.|
@@ -793,7 +793,7 @@ WorkflowSpec is the specification of a Workflow.
 |`onExit`|`string`|OnExit is a template reference which is invoked at the end of the workflow, irrespective of the success, failure, or error of the primary io.argoproj.workflow.v1alpha1.|
 |`parallelism`|`integer`|Parallelism limits the max total parallel pods that can execute at the same time in a workflow|
 |`podDisruptionBudget`|[`PodDisruptionBudgetSpec`](#poddisruptionbudgetspec)|PodDisruptionBudget holds the number of concurrent disruptions that you allow for Workflow's Pods. Controller will automatically add the selector with workflow name, if selector is empty. Optional: Defaults to empty.|
-|`podGC`|[`PodGC`](#podgc)|PodGC describes the strategy to use when to deleting completed pods|
+|`podGC`|[`PodGC`](#podgc)|PodGC describes the strategy to use when deleting completed pods|
 |`podMetadata`|[`Metadata`](#metadata)|PodMetadata defines additional metadata that should be applied to workflow pods|
 |`podPriority`|`integer`|Priority to apply to workflow pods.|
 |`podPriorityClassName`|`string`|PriorityClassName to apply to workflow pods.|
@@ -810,7 +810,7 @@ WorkflowSpec is the specification of a Workflow.
 |`templates`|`Array<`[`Template`](#template)`>`|Templates is a list of workflow templates used in a workflow|
 |`tolerations`|`Array<`[`Toleration`](#toleration)`>`|Tolerations to apply to workflow pods.|
 |`ttlStrategy`|[`TTLStrategy`](#ttlstrategy)|TTLStrategy limits the lifetime of a Workflow that has finished execution depending on if it Succeeded or Failed. If this struct is set, once the Workflow finishes, it will be deleted after the time to live expires. If this field is unset, the controller config map will hold the default values.|
-|`volumeClaimGC`|[`VolumeClaimGC`](#volumeclaimgc)|VolumeClaimGC describes the strategy to use when to deleting volumes from completed workflows|
+|`volumeClaimGC`|[`VolumeClaimGC`](#volumeclaimgc)|VolumeClaimGC describes the strategy to use when deleting volumes from completed workflows|
 |`volumeClaimTemplates`|`Array<`[`PersistentVolumeClaim`](#persistentvolumeclaim)`>`|VolumeClaimTemplates is a list of claims that containers are allowed to reference. The Workflow controller will create the claims at the beginning of the workflow and delete the claims upon completion of the workflow|
 |`volumes`|`Array<`[`Volume`](#volume)`>`|Volumes is a list of volumes that can be mounted by containers in a io.argoproj.workflow.v1alpha1.|
 |`workflowMetadata`|[`WorkflowMetadata`](#workflowmetadata)|WorkflowMetadata contains some metadata of the workflow to be refer|

pkg/apis/workflow/v1alpha1/workflow_types.go

@@ -196,6 +196,19 @@ func (w *Workflow) GetExecSpec() *WorkflowSpec {
 	return &w.Spec
 }
 
+func (w *Workflow) HasArtifactGC() bool {
+	// either it's defined by an Output Artifact or by the WorkflowSpec itself, or both
+	for _, template := range w.Spec.Templates {
+		for _, artifact := range template.Outputs.Artifacts {
+			if artifact.GetArtifactGC().Strategy != ArtifactGCNever {
+				return true
+			}
+		}
+	}
+
+	return (w.Spec.ArtifactGC != nil && w.Spec.ArtifactGC.Strategy != ArtifactGCNever)
+}
+
 var (
 	WorkflowCreatedAfter = func(t time.Time) WorkflowPredicate {
 		return func(wf Workflow) bool {
@@ -342,7 +355,7 @@ type WorkflowSpec struct {
 	// +optional
 	SchedulerName string `json:"schedulerName,omitempty" protobuf:"bytes,21,opt,name=schedulerName"`
 
-	// PodGC describes the strategy to use when to deleting completed pods
+	// PodGC describes the strategy to use when deleting completed pods
 	PodGC *PodGC `json:"podGC,omitempty" protobuf:"bytes,22,opt,name=podGC"`
 
 	// PriorityClassName to apply to workflow pods.
@@ -382,7 +395,7 @@ type WorkflowSpec struct {
 	// Synchronization holds synchronization lock configuration for this Workflow
 	Synchronization *Synchronization `json:"synchronization,omitempty" protobuf:"bytes,35,opt,name=synchronization,casttype=Synchronization"`
 
-	// VolumeClaimGC describes the strategy to use when to deleting volumes from completed workflows
+	// VolumeClaimGC describes the strategy to use when deleting volumes from completed workflows
 	VolumeClaimGC *VolumeClaimGC `json:"volumeClaimGC,omitempty" protobuf:"bytes,36,opt,name=volumeClaimGC,casttype=VolumeClaimGC"`
 
 	// RetryStrategy for all templates in the workflow.
@@ -404,7 +417,8 @@ type WorkflowSpec struct {
 	// WorkflowMetadata contains some metadata of the workflow to be refer
 	WorkflowMetadata *WorkflowMetadata `json:"workflowMetadata,omitempty" protobuf:"bytes,42,opt,name=workflowMetadata"`
 
-	// ArtifactGC describes the strategy to use when to deleting artifacts from completed or deleted workflows
+	// ArtifactGC describes the strategy to use when deleting artifacts from completed or deleted workflows (applies to all output Artifacts
+	// unless Artifact.ArtifactGC is specified, which overrides this)
 	ArtifactGC *ArtifactGC `json:"artifactGC,omitempty" protobuf:"bytes,43,opt,name=artifactGC"`
 }
 

pkg/apis/workflow/v1alpha1/workflow_types_test.go

@@ -1,6 +1,7 @@
 package v1alpha1
 
 import (
+	"fmt"
 	"sort"
 	"testing"
 	"time"
@@ -116,6 +117,96 @@ func TestWorkflowHappenedBetween(t *testing.T) {
 	}))
 }
 
+func TestWorkflowHasArtifactGC(t *testing.T) {
+	tests := []struct {
+		name                      string
+		workflowArtGCStrategySpec string
+		artifactGCStrategySpec    string
+		expectedResult            bool
+	}{
+		{
+			name: "WorkflowSpecGC_Completion",
+			workflowArtGCStrategySpec: `
+              artifactGC:
+                strategy: OnWorkflowCompletion`,
+			artifactGCStrategySpec: "",
+			expectedResult:         true,
+		},
+		{
+			name:                      "ArtifactSpecGC_Completion",
+			workflowArtGCStrategySpec: "",
+			artifactGCStrategySpec: `
+                      artifactGC:
+                        strategy: OnWorkflowCompletion`,
+			expectedResult: true,
+		},
+		{
+			name: "WorkflowSpecGC_Deletion",
+			workflowArtGCStrategySpec: `
+              artifactGC:
+                strategy: OnWorkflowDeletion`,
+			artifactGCStrategySpec: "",
+			expectedResult:         true,
+		},
+		{
+			name:                      "ArtifactSpecGC_Deletion",
+			workflowArtGCStrategySpec: "",
+			artifactGCStrategySpec: `
+                      artifactGC:
+                        strategy: OnWorkflowDeletion`,
+			expectedResult: true,
+		},
+		{
+			name:                      "NoGC",
+			workflowArtGCStrategySpec: "",
+			artifactGCStrategySpec:    "",
+			expectedResult:            false,
+		},
+		{
+			name: "WorkflowSpecGCNone",
+			workflowArtGCStrategySpec: `
+              artifactGC:
+                strategy: ""`,
+			artifactGCStrategySpec: "",
+			expectedResult:         false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+
+			workflowSpec := fmt.Sprintf(`
+            apiVersion: argoproj.io/v1alpha1
+            kind: Workflow
+            metadata:
+              generateName: artifact-passing-
+            spec:
+              entrypoint: whalesay
+              %s
+              templates:
+              - name: whalesay
+                container:
+                  image: docker/whalesay:latest
+                  command: [sh, -c]
+                  args: ["sleep 1; cowsay hello world | tee /tmp/hello_world.txt"]
+                outputs:
+                  artifacts:
+                    - name: out
+                      path: /out
+                      s3:
+                        key: out
+                        %s`, tt.workflowArtGCStrategySpec, tt.artifactGCStrategySpec)
+
+			wf := MustUnmarshalWorkflow(workflowSpec)
+
+			hasArtifact := wf.HasArtifactGC()
+
+			assert.Equal(t, hasArtifact, tt.expectedResult)
+		})
+	}
+
+}
+
 func TestArtifact_ValidatePath(t *testing.T) {
 	t.Run("empty path fails", func(t *testing.T) {
 		a1 := Artifact{Name: "a1", Path: ""}

workflow/controller/operator.go

@@ -3517,7 +3517,7 @@ func (woc *wfOperationCtx) addFinalizers() {
 }
 
 func (woc *wfOperationCtx) addArtifactGCFinalizer() {
-	if woc.execWf.Spec.ArtifactGC != nil && woc.execWf.Spec.ArtifactGC.Strategy != wfv1.ArtifactGCNever {
+	if woc.execWf.HasArtifactGC() {
 		finalizers := append(woc.wf.GetFinalizers(), common.FinalizerArtifactGC)
 		woc.wf.SetFinalizers(finalizers)
 	}

workflow/controller/operator_test.go

@@ -3778,37 +3778,6 @@ func TestNestedOptionalOutputArtifacts(t *testing.T) {
 	assert.Equal(t, wfv1.WorkflowSucceeded, woc.wf.Status.Phase)
 }
 
-var artifactGCOnWorkflowCompletion = `
-apiVersion: argoproj.io/v1alpha1
-kind: Workflow
-metadata:
-  generateName: artifact-passing-
-spec:
-  entrypoint: whalesay
-  artifactGC:
-    strategy: OnWorkflowCompletion
-  templates:
-  - name: whalesay
-    container:
-      image: docker/whalesay:latest
-      command: [sh, -c]
-      args: ["sleep 1; cowsay hello world | tee /tmp/hello_world.txt"]
-`
-
-func TestArtifactGCOnWorkflowCompletion(t *testing.T) {
-	cancel, controller := newController()
-	defer cancel()
-	ctx := context.Background()
-
-	wf := wfv1.MustUnmarshalWorkflow(artifactGCOnWorkflowCompletion)
-	woc := newWorkflowOperationCtx(wf, controller)
-	woc.operate(ctx)
-
-	assert.Equal(t, wfv1.WorkflowRunning, woc.wf.Status.Phase)
-	assert.Equal(t, 1, len(woc.wf.GetFinalizers()))
-	assert.Equal(t, "workflows.argoproj.io/artifact-gc", woc.wf.GetFinalizers()[0])
-}
-
 //  TestPodSpecLogForFailedPods tests PodSpec logging configuration
 func TestPodSpecLogForFailedPods(t *testing.T) {
 	wf := wfv1.MustUnmarshalWorkflow(helloWorldWf)