test:

- add explicit failure test for compressed workflows in removeCompletedTaskSetStatus
- add e2e coverage for WorkflowTaskSet cleanup in completed compressed workflows

Signed-off-by: arpechenin <arpechenin@avito.ru>

Author: arpechenin

test/e2e/executor_plugins_test.go

@@ -1,5 +1,3 @@
-//go:build plugins
-
 package e2e
 
 import (
@@ -82,6 +80,22 @@ func (s *ExecutorPluginsSuite) TestTemplateExecutor() {
 		})
 }
 
+func (s *ExecutorPluginsSuite) TestCompressedTemplateExecutor_WorkflowTaskSetIsProperlyCleaned() {
+	s.Given().
+		Workflow("@testdata/plugins/executor/massive-executor-workflow.yaml").
+		When().
+		SubmitWorkflow().
+		WaitForWorkflow(fixtures.ToBeSucceeded).
+		Then().
+		ExpectWorkflowCompressed().
+		ExpectWorkflowTaskSet(func(t *testing.T, wfts *wfv1.WorkflowTaskSet) {
+			assert.NotNil(t, wfts)
+			assert.Empty(t, wfts.Spec.Tasks)
+			assert.Empty(t, wfts.Status.Nodes)
+			assert.Equal(t, "true", wfts.Labels[common.LabelKeyCompleted])
+		})
+}
+
 func TestExecutorPluginsSuite(t *testing.T) {
 	suite.Run(t, new(ExecutorPluginsSuite))
 }

test/e2e/fixtures/then.go

@@ -83,6 +83,18 @@ func (t *Then) ExpectWorkflowDeleted() *Then {
 	return t
 }
 
+func (t *Then) ExpectWorkflowCompressed() *Then {
+	ctx := logging.TestContext(t.t.Context())
+	wf, err := t.client.Get(ctx, t.wf.Name, metav1.GetOptions{})
+	if err != nil {
+		t.t.Fatal(err)
+	}
+	if wf.Status.CompressedNodes == "" {
+		t.t.Errorf("expected workflow to be compressed")
+	}
+	return t
+}
+
 // ExpectWorkflowNode checks on a specific node in the workflow.
 // If no node matches the selector, then the NodeStatus and Pod will be nil.
 // If the pod does not exist (e.g. because it was deleted) then the Pod will be nil too.

test/e2e/manifests/plugins/hello-executor-plugin-configmap.yaml

@@ -30,16 +30,17 @@ data:
               self.end_headers()
 
           def do_POST(self):
-              if self.headers.get("Authorization") != "Bearer " + token:
-                  self.forbidden()
-              elif self.path == '/api/v1/template.execute':
+              if self.path == '/api/v1/template.execute':
                   args = self.args()
                   if 'hello' in args['template'].get('plugin', {}):
+                      if self.headers.get("Authorization") != "Bearer " + token:
+                        self.forbidden()
+                        return
                       self.reply(
                           {'node': {'phase': 'Succeeded', 'message': 'Hello template!',
                                     'outputs': {'parameters': [{'name': 'foo', 'value': 'bar'}]}}})
                   else:
-                      self.reply({})
+                      self.reply(None)
               else:
                   self.unsupported()
 

test/e2e/manifests/plugins/kustomization.yaml

@@ -6,5 +6,6 @@ resources:
   - hello-executor-plugin-serviceaccount.yaml
   - hello-executor-plugin.service-account-token-secret.yaml
   - hello-executor-plugin-configmap.yaml
+  - massive-executor-plugin-configmap.yaml
 
 namespace: argo

test/e2e/manifests/plugins/massive-executor-plugin-configmap.yaml

@@ -0,0 +1,89 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: massive-executor-plugin
+  labels:
+    workflows.argoproj.io/configmap-type: ExecutorPlugin
+  annotations:
+    workflows.argoproj.io/description: |
+      This plugin returns a massive string (~1000 chars).
+data:
+  sidecar.automountServiceAccountToken: "false"
+  sidecar.container: |
+    args:
+    - |
+      import json
+      import random
+      import string
+      from http.server import BaseHTTPRequestHandler, HTTPServer
+
+      with open("/var/run/argo/token") as f:
+          token = f.read().strip()
+
+      def gen_text():
+        base = "HELLO_WORLD_123 "
+        return base * (100_000 // len(base))
+
+      class Plugin(BaseHTTPRequestHandler):
+
+          def args(self):
+              return json.loads(self.rfile.read(int(self.headers.get('Content-Length'))))
+
+          def reply(self, reply):
+              self.send_response(200)
+              self.end_headers()
+              self.wfile.write(json.dumps(reply).encode("UTF-8"))
+
+          def unsupported(self):
+              self.send_response(404)
+              self.end_headers()
+
+          def do_POST(self):
+              if self.path == '/api/v1/template.execute':
+                  args = self.args()
+
+                  if 'massive' in args['template'].get('plugin', {}):
+                      text = gen_text()
+                      self.reply({
+                          'node': {
+                              'phase': 'Succeeded',
+                              'message': 'Massive payload generated',
+                              'outputs': {
+                                  'parameters': [
+                                      {
+                                          'name': 'big-text',
+                                          'value': text
+                                      }
+                                  ]
+                              }
+                          }
+                      })
+                  else:
+                      self.reply(None)
+              else:
+                  self.unsupported()
+
+      if __name__ == '__main__':
+          httpd = HTTPServer(('', 4355), Plugin)
+          httpd.serve_forever()
+    command:
+    - python
+    - -c
+    image: python:alpine3.23
+    name: massive-executor-plugin
+    ports:
+    - containerPort: 4356
+    resources:
+      limits:
+        cpu: 200m
+        memory: 64Mi
+      requests:
+        cpu: 100m
+        memory: 32Mi
+    securityContext:
+      allowPrivilegeEscalation: false
+      capabilities:
+        drop:
+        - ALL
+      readOnlyRootFilesystem: true
+      runAsNonRoot: true

test/e2e/testdata/plugins/executor/massive-executor-workflow.yaml

@@ -0,0 +1,26 @@
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  generateName: massive-executor-
+spec:
+  entrypoint: main
+  parallelism: 6
+
+  templates:
+    - name: main
+      steps:
+        - - name: fanout
+            template: massive-plugin
+            withSequence:
+              count: "15"
+            arguments:
+              parameters:
+                - name: idx
+                  value: "{{item}}"
+
+    - name: massive-plugin
+      inputs:
+        parameters:
+          - name: idx
+      plugin:
+        massive: {}

workflow/controller/taskset_test.go

@@ -312,6 +312,35 @@ status:
 			assert.Empty(t, ts.Status.Nodes)
 		}
 	})
+	t.Run("RemoveCompletedTaskSetStatusFailsForCompressedWorkflow", func(t *testing.T) {
+		cancel, controller := newController(ctx, wf, ts)
+		defer cancel()
+
+		_, err := controller.wfclientset.ArgoprojV1alpha1().
+			WorkflowTaskSets("default").
+			Create(ctx, &ts, v1.CreateOptions{})
+		require.NoError(t, err)
+
+		compressedWf := wf.DeepCopy()
+		compressedWf.Status.CompressedNodes = "compressed"
+
+		woc := newWorkflowOperationCtx(ctx, compressedWf, controller)
+
+		err = woc.removeCompletedTaskSetStatus(ctx)
+		require.Error(t, err)
+		require.ErrorContains(t, err, "workflow must be decompressed")
+
+		tslist, err := woc.controller.wfclientset.ArgoprojV1alpha1().
+			WorkflowTaskSets("default").
+			List(ctx, v1.ListOptions{})
+		require.NoError(t, err)
+
+		require.Len(t, tslist.Items, 1)
+
+		// Ensure tasksets were not modified.
+		assert.NotEmpty(t, tslist.Items[0].Spec.Tasks)
+		assert.NotEmpty(t, tslist.Items[0].Status.Nodes)
+	})
 }
 
 func TestNonHTTPTemplateScenario(t *testing.T) {