Skip to content

Commit 5ac265d

Browse files
Joibelntny
Joibel
and
ntny
authored
fix: WorkflowTaskSets size bloat for large workflows (cherry-pick #16075 for 4.0) (#16253)
Signed-off-by: arpechenin <arpechenin@avito.ru> Signed-off-by: Alan Clucas <alan@clucas.org> Co-authored-by: Anton Pechenin <ntny1986@gmail.com>
1 parent 03fbf9f commit 5ac265d

9 files changed

Lines changed: 179 additions & 16 deletions

File tree

‎test/e2e/executor_plugins_test.go‎

Lines changed: 32 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -49,16 +49,17 @@ func (s *ExecutorPluginsSuite) TestTemplateExecutor() {
4949
assert.Contains(t, spec.Volumes[2].Name, "kube-api-access-")
5050
assert.Equal(t, "argo-workflows-agent-ca-certificates", spec.Volumes[3].Name)
5151

52-
require.Len(t, spec.Containers, 2)
52+
require.Len(t, spec.Containers, 3)
5353
{
54-
plug := spec.Containers[0]
54+
plug := requireFindContainerByName(t, spec.Containers, "hello-executor-plugin")
55+
require.NotNil(t, plug)
5556
require.Equal(t, "hello-executor-plugin", plug.Name)
5657
require.Len(t, plug.VolumeMounts, 2)
5758
assert.Equal(t, "var-run-argo", plug.VolumeMounts[0].Name)
5859
assert.Contains(t, plug.VolumeMounts[1].Name, "kube-api-access-")
5960
}
6061
{
61-
agent := spec.Containers[1]
62+
agent := requireFindContainerByName(t, spec.Containers, "main")
6263
require.Equal(t, "main", agent.Name)
6364
require.Len(t, agent.VolumeMounts, 3)
6465
assert.Equal(t, "var-run-argo", agent.VolumeMounts[0].Name)
@@ -83,6 +84,34 @@ func (s *ExecutorPluginsSuite) TestTemplateExecutor() {
8384
})
8485
}
8586

87+
func (s *ExecutorPluginsSuite) TestCompressedTemplateExecutor_WorkflowTaskSetIsProperlyCleaned() {
88+
s.Given().
89+
Workflow("@testdata/plugins/executor/massive-executor-workflow.yaml").
90+
When().
91+
SubmitWorkflow().
92+
WaitForWorkflow(fixtures.ToBeSucceeded).
93+
Then().
94+
ExpectWorkflowCompressed().
95+
ExpectWorkflowTaskSet(func(t *testing.T, wfts *wfv1.WorkflowTaskSet) {
96+
assert.NotNil(t, wfts)
97+
assert.Empty(t, wfts.Status.Nodes)
98+
assert.Empty(t, wfts.Spec.Tasks)
99+
assert.Equal(t, "true", wfts.Labels[common.LabelKeyCompleted])
100+
})
101+
}
102+
86103
func TestExecutorPluginsSuite(t *testing.T) {
87104
suite.Run(t, new(ExecutorPluginsSuite))
88105
}
106+
107+
func requireFindContainerByName(t *testing.T, containers []apiv1.Container, name string) *apiv1.Container {
108+
var result *apiv1.Container
109+
for _, container := range containers {
110+
if container.Name == name {
111+
result = &container
112+
break
113+
}
114+
}
115+
require.NotNil(t, result, "could not find container %s", name)
116+
return result
117+
}

‎test/e2e/fixtures/then.go‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -83,6 +83,18 @@ func (t *Then) ExpectWorkflowDeleted() *Then {
8383
return t
8484
}
8585

86+
func (t *Then) ExpectWorkflowCompressed() *Then {
87+
ctx := logging.TestContext(t.t.Context())
88+
wf, err := t.client.Get(ctx, t.wf.Name, metav1.GetOptions{})
89+
if err != nil {
90+
t.t.Fatal(err)
91+
}
92+
if wf.Status.CompressedNodes == "" {
93+
t.t.Errorf("expected workflow to be compressed")
94+
}
95+
return t
96+
}
97+
8698
// Check on a specific node in the workflow.
8799
// If no node matches the selector, then the NodeStatus and Pod will be nil.
88100
// 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‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -30,16 +30,17 @@ data:
3030
self.end_headers()
3131
3232
def do_POST(self):
33-
if self.headers.get("Authorization") != "Bearer " + token:
34-
self.forbidden()
35-
elif self.path == '/api/v1/template.execute':
33+
if self.path == '/api/v1/template.execute':
3634
args = self.args()
3735
if 'hello' in args['template'].get('plugin', {}):
36+
if self.headers.get("Authorization") != "Bearer " + token:
37+
self.forbidden()
38+
return
3839
self.reply(
3940
{'node': {'phase': 'Succeeded', 'message': 'Hello template!',
4041
'outputs': {'parameters': [{'name': 'foo', 'value': 'bar'}]}}})
4142
else:
42-
self.reply({})
43+
self.reply(None)
4344
else:
4445
self.unsupported()
4546

‎test/e2e/manifests/plugins/kustomization.yaml‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,5 +6,6 @@ resources:
66
- hello-executor-plugin-serviceaccount.yaml
77
- hello-executor-plugin.service-account-token-secret.yaml
88
- hello-executor-plugin-configmap.yaml
9+
- massive-executor-plugin-configmap.yaml
910

1011
namespace: argo

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

Lines changed: 89 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,89 @@
1+
apiVersion: v1
2+
kind: ConfigMap
3+
metadata:
4+
name: massive-executor-plugin
5+
labels:
6+
workflows.argoproj.io/configmap-type: ExecutorPlugin
7+
annotations:
8+
workflows.argoproj.io/description: |
9+
This plugin returns a massive string (~100000 chars).
10+
data:
11+
sidecar.automountServiceAccountToken: "false"
12+
sidecar.container: |
13+
args:
14+
- |
15+
import json
16+
import random
17+
import string
18+
from http.server import BaseHTTPRequestHandler, HTTPServer
19+
20+
with open("/var/run/argo/token") as f:
21+
token = f.read().strip()
22+
23+
def gen_text():
24+
base = "HELLO_WORLD_123 "
25+
return base * (100_000 // len(base))
26+
27+
class Plugin(BaseHTTPRequestHandler):
28+
29+
def args(self):
30+
return json.loads(self.rfile.read(int(self.headers.get('Content-Length'))))
31+
32+
def reply(self, reply):
33+
self.send_response(200)
34+
self.end_headers()
35+
self.wfile.write(json.dumps(reply).encode("UTF-8"))
36+
37+
def unsupported(self):
38+
self.send_response(404)
39+
self.end_headers()
40+
41+
def do_POST(self):
42+
if self.path == '/api/v1/template.execute':
43+
args = self.args()
44+
45+
if 'massive' in args['template'].get('plugin', {}):
46+
text = gen_text()
47+
self.reply({
48+
'node': {
49+
'phase': 'Succeeded',
50+
'message': 'Massive payload generated',
51+
'outputs': {
52+
'parameters': [
53+
{
54+
'name': 'big-text',
55+
'value': text
56+
}
57+
]
58+
}
59+
}
60+
})
61+
else:
62+
self.reply(None)
63+
else:
64+
self.unsupported()
65+
66+
if __name__ == '__main__':
67+
httpd = HTTPServer(('', 4356), Plugin)
68+
httpd.serve_forever()
69+
command:
70+
- python
71+
- -c
72+
image: python:alpine3.23
73+
name: massive-executor-plugin
74+
ports:
75+
- containerPort: 4356
76+
resources:
77+
limits:
78+
cpu: 200m
79+
memory: 64Mi
80+
requests:
81+
cpu: 100m
82+
memory: 32Mi
83+
securityContext:
84+
allowPrivilegeEscalation: false
85+
capabilities:
86+
drop:
87+
- ALL
88+
readOnlyRootFilesystem: true
89+
runAsNonRoot: true

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

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,26 @@
1+
apiVersion: argoproj.io/v1alpha1
2+
kind: Workflow
3+
metadata:
4+
generateName: massive-executor-
5+
spec:
6+
entrypoint: main
7+
parallelism: 6
8+
9+
templates:
10+
- name: main
11+
steps:
12+
- - name: fanout
13+
template: massive-plugin
14+
withSequence:
15+
count: "15"
16+
arguments:
17+
parameters:
18+
- name: idx
19+
value: "{{item}}"
20+
21+
- name: massive-plugin
22+
inputs:
23+
parameters:
24+
- name: idx
25+
plugin:
26+
massive: {}

‎workflow/controller/operator.go‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -743,8 +743,10 @@ func (woc *wfOperationCtx) persistUpdates(ctx context.Context) {
743743
woc.log.WithPanic().Error(ctx, "cannot persist updates with mismatched resource versions")
744744
}
745745
wfClient := woc.controller.wfclientset.ArgoprojV1alpha1().Workflows(woc.wf.Namespace)
746-
// try and compress nodes if needed
746+
747747
nodes := woc.wf.Status.Nodes
748+
749+
// try and compress nodes if needed
748750
err := woc.controller.hydrator.Dehydrate(ctx, woc.wf)
749751
if err != nil {
750752
woc.log.WithError(err).Warn(ctx, "Failed to dehydrate")
@@ -759,7 +761,7 @@ func (woc *wfOperationCtx) persistUpdates(ctx context.Context) {
759761
}
760762

761763
// Remove completed taskset status before update workflow.
762-
err = woc.removeCompletedTaskSetStatus(ctx)
764+
err = woc.removeCompletedTaskSetStatus(ctx, nodes)
763765
if err != nil {
764766
woc.log.WithError(err).Warn(ctx, "error updating taskset")
765767
}

‎workflow/controller/taskset.go‎

Lines changed: 8 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -30,9 +30,9 @@ func (woc *wfOperationCtx) mergePatchTaskSet(ctx context.Context, patch interfac
3030
return nil
3131
}
3232

33-
func (woc *wfOperationCtx) getDeleteTaskAndNodePatch() (tasksPatch map[string]interface{}, nodesPatch map[string]interface{}) {
33+
func (woc *wfOperationCtx) getDeleteTaskAndNodePatch(nodes wfv1.Nodes) (tasksPatch map[string]interface{}, nodesPatch map[string]interface{}) {
3434
deletedNode := make(map[string]interface{})
35-
for _, node := range woc.wf.Status.Nodes {
35+
for _, node := range nodes {
3636
if node.IsTaskSetNode() && node.Fulfilled() {
3737
deletedNode[node.ID] = nil
3838
}
@@ -66,11 +66,14 @@ func (woc *wfOperationCtx) hasTaskSetNodes() bool {
6666
})
6767
}
6868

69-
func (woc *wfOperationCtx) removeCompletedTaskSetStatus(ctx context.Context) error {
70-
if !woc.hasTaskSetNodes() {
69+
func (woc *wfOperationCtx) removeCompletedTaskSetStatus(ctx context.Context, nodes wfv1.Nodes) error {
70+
// Avoid sending empty patches when there are no completed taskset nodes to remove.
71+
if !nodes.Any(func(node wfv1.NodeStatus) bool {
72+
return node.IsTaskSetNode() && node.Fulfilled()
73+
}) {
7174
return nil
7275
}
73-
tasksPatch, nodesPatch := woc.getDeleteTaskAndNodePatch()
76+
tasksPatch, nodesPatch := woc.getDeleteTaskAndNodePatch(nodes)
7477
if woc.wf.Status.Fulfilled() {
7578
tasksPatch["metadata"] = metav1.ObjectMeta{
7679
Labels: map[string]string{

‎workflow/controller/taskset_test.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -297,7 +297,7 @@ status:
297297
_, err := controller.wfclientset.ArgoprojV1alpha1().WorkflowTaskSets("default").Create(ctx, &ts, v1.CreateOptions{})
298298
require.NoError(t, err)
299299
woc := newWorkflowOperationCtx(ctx, wf, controller)
300-
err = woc.removeCompletedTaskSetStatus(ctx)
300+
err = woc.removeCompletedTaskSetStatus(ctx, woc.wf.Status.Nodes)
301301
require.NoError(t, err)
302302
tslist, err := woc.controller.wfclientset.ArgoprojV1alpha1().WorkflowTaskSets("default").List(ctx, v1.ListOptions{})
303303
require.NoError(t, err)
@@ -328,7 +328,7 @@ func TestNonHTTPTemplateScenario(t *testing.T) {
328328
})
329329
t.Run("removeCompletedTaskSetStatus", func(t *testing.T) {
330330
woc.operate(ctx)
331-
err := woc.removeCompletedTaskSetStatus(ctx)
331+
err := woc.removeCompletedTaskSetStatus(ctx, woc.wf.Status.Nodes)
332332
require.NoError(t, err)
333333
})
334334
}

0 commit comments

Comments
 (0)