Skip to content

feat(controller): add wfThrottleQueue to accelerate event handling. Fixes: #14791 #14876

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
shuangkun wants to merge 6 commits into
base: main
Choose a base branch
from
+197 −10
Open

feat(controller): add wfThrottleQueue to accelerate event handling. Fixes: #14791 #14876

Changes from 1 commit
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
75f4708
feat: add throttle queue to speed up eventhandle processing.
shuangkun Sep 21, 2025
e9196cf
feat: add throttle queue to speed up eventhandle processing.
shuangkun Sep 21, 2025
ab61934
fix: docs
shuangkun Sep 21, 2025
e55f0de
fix: ut.
shuangkun Sep 21, 2025
71952cf
fix: ut.
shuangkun Sep 21, 2025
ab94e8f
fix: ut.
shuangkun Sep 28, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
fix: ut. 
Signed-off-by: shuangkun <[email protected]>
  • Loading branch information
@shuangkun
shuangkun committed Sep 28, 2025
commit ab94e8fb81dc22d895d22f108e5c639ad31cb948
14 changes: 11 additions & 3 deletions workflow/controller/controller_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -710,6 +710,7 @@ func TestParallelism(t *testing.T) {
wfv1.MustUnmarshalWorkflow(`
metadata:
name: my-wf-0
namespace: default
spec:
entrypoint: main
templates:
Expand All @@ -720,6 +721,7 @@ spec:
wfv1.MustUnmarshalWorkflow(`
metadata:
name: my-wf-1
namespace: default
spec:
entrypoint: main
templates:
Expand All @@ -730,6 +732,7 @@ spec:
wfv1.MustUnmarshalWorkflow(`
metadata:
name: my-wf-2
namespace: default
spec:
shutdown: Terminate
entrypoint: main
Expand All @@ -742,20 +745,25 @@ spec:
)
defer cancel()

// Process throttle items first to handle throttling
assert.True(t, controller.processNextThrottleItem(ctx))
assert.True(t, controller.processNextThrottleItem(ctx))
assert.True(t, controller.processNextThrottleItem(ctx))

assert.True(t, controller.processNextItem(ctx))
assert.True(t, controller.processNextItem(ctx))
assert.True(t, controller.processNextItem(ctx))

expectWorkflow(ctx, controller, "my-wf-0", func(wf *wfv1.Workflow) {
expectNamespacedWorkflow(ctx, controller, "default", "my-wf-0", func(wf *wfv1.Workflow) {
require.NotNil(t, wf)
assert.Equal(t, wfv1.WorkflowRunning, wf.Status.Phase)
})
expectWorkflow(ctx, controller, "my-wf-1", func(wf *wfv1.Workflow) {
expectNamespacedWorkflow(ctx, controller, "default", "my-wf-1", func(wf *wfv1.Workflow) {
require.NotNil(t, wf)
assert.Equal(t, wfv1.WorkflowPending, wf.Status.Phase)
assert.Equal(t, "Workflow processing has been postponed because too many workflows are already running", wf.Status.Message)
})
expectWorkflow(ctx, controller, "my-wf-2", func(wf *wfv1.Workflow) {
expectNamespacedWorkflow(ctx, controller, "default", "my-wf-2", func(wf *wfv1.Workflow) {
require.NotNil(t, wf)
assert.Equal(t, wfv1.WorkflowFailed, wf.Status.Phase)
})
Expand Down
Loading