This is an automated email from the ASF dual-hosted git repository.
wilfred-s pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/yunikorn-k8shim.git
The following commit(s) were added to refs/heads/master by this push:
new 553b43ff [YUNIKORN-3431] Add regression test coverage for
KubernetesShim scheduling loop shutdown (#1085)
553b43ff is described below
commit 553b43ff1fe90b34dcbd64306d474ca1f8513843
Author: hedger9487 <[email protected]>
AuthorDate: Thu Sep 10 20:59:02 2026 +1000
[YUNIKORN-3431] Add regression test coverage for KubernetesShim scheduling
loop shutdown (#1085)
In YUNIKORN-3367, KubernetesShim.Stop() was updated to close stopChan
so that both wait.Until scheduling loops started by doScheduling()
terminate. However, existing unit tests only asserted that stopChan was
closed without starting doScheduling() or verifying that both loop
goroutines actually terminate. A regression to a single send on an
unbuffered channel or a buffered channel could satisfy existing channel
assertions while silently leaking one loop.
This change adds regression test coverage without modifying production code:
1. Implement CountGoroutines in pkg/common/utils using runtime.Stack
inspection to non-invasively track active goroutines matching a pattern,
accompanied by comprehensive unit tests.
2. Add TestSchedulingLoopsShutdownOnStop using utils.WaitForCondition and
utils.CountGoroutines to verify that stopping the cluster terminates
both scheduling loop goroutines cleanly under real cluster conditions.
Closes: #1085
Signed-off-by: Wilfred Spiegelenburg <[email protected]>
---
pkg/common/utils/utils.go | 16 +++++++++++++++
pkg/common/utils/utils_test.go | 44 ++++++++++++++++++++++++++++++++++++++++++
pkg/shim/scheduler_test.go | 20 +++++++++++++++++++
3 files changed, 80 insertions(+)
diff --git a/pkg/common/utils/utils.go b/pkg/common/utils/utils.go
index e42bfbce..4522da2e 100644
--- a/pkg/common/utils/utils.go
+++ b/pkg/common/utils/utils.go
@@ -23,6 +23,7 @@ import (
"errors"
"fmt"
"reflect"
+ "runtime"
"strconv"
"strings"
"time"
@@ -301,6 +302,21 @@ func WaitForCondition(eval func() bool, interval
time.Duration, timeout time.Dur
}
}
+// CountGoroutines returns the number of active goroutines whose stack trace
contains the given pattern.
+func CountGoroutines(pattern string) int {
+ if pattern == "" {
+ return 0
+ }
+ buf := make([]byte, 1024)
+ for {
+ n := runtime.Stack(buf, true)
+ if n < len(buf) {
+ return strings.Count(string(buf[:n]), pattern)
+ }
+ buf = make([]byte, 2*len(buf))
+ }
+}
+
// merge two string maps
// if the same key defined in the first and second maps
// the value will be set by the second map
diff --git a/pkg/common/utils/utils_test.go b/pkg/common/utils/utils_test.go
index e226102e..f01c3186 100644
--- a/pkg/common/utils/utils_test.go
+++ b/pkg/common/utils/utils_test.go
@@ -1264,3 +1264,47 @@ func TestWaitForCondition(t *testing.T) {
}
}
}
+
+func dummyGoroutineWorker(stop chan struct{}) {
+ <-stop
+}
+
+func TestCountGoroutines(t *testing.T) {
+ // Negative test: empty pattern should return 0
+ assert.Equal(t, CountGoroutines(""), 0, "empty pattern should return 0")
+
+ // Negative test: non-existent pattern should return 0
+ assert.Equal(t, CountGoroutines("nonExistentGoroutineWorker_XYZ"), 0,
"non-existent pattern should return 0")
+
+ // 0-1-N concurrency boundary test: 0 -> 3 -> 2 -> 0
+ assert.Equal(t, CountGoroutines("dummyGoroutineWorker"), 0)
+
+ stop1 := make(chan struct{})
+ stop2 := make(chan struct{})
+ stop3 := make(chan struct{})
+
+ go dummyGoroutineWorker(stop1)
+ go dummyGoroutineWorker(stop2)
+ go dummyGoroutineWorker(stop3)
+
+ // Verify count transitions from 0 to 3
+ err := WaitForCondition(func() bool {
+ return CountGoroutines("dummyGoroutineWorker") == 3
+ }, 10*time.Millisecond, time.Second)
+ assert.NilError(t, err, "expected exactly 3 goroutines")
+
+ // Verify count decreases from 3 to 2 when one goroutine finishes
+ close(stop1)
+ err = WaitForCondition(func() bool {
+ return CountGoroutines("dummyGoroutineWorker") == 2
+ }, 10*time.Millisecond, time.Second)
+ assert.NilError(t, err, "expected count to decrease to 2")
+
+ // Verify count decreases from 2 to 0 when remaining goroutines finish
+ close(stop2)
+ close(stop3)
+ err = WaitForCondition(func() bool {
+ return CountGoroutines("dummyGoroutineWorker") == 0
+ }, 10*time.Millisecond, time.Second)
+ assert.NilError(t, err, "expected all goroutines to terminate")
+}
diff --git a/pkg/shim/scheduler_test.go b/pkg/shim/scheduler_test.go
index d7a8cae5..57a8226a 100644
--- a/pkg/shim/scheduler_test.go
+++ b/pkg/shim/scheduler_test.go
@@ -203,6 +203,26 @@ func TestSchedulerStopStopsAPIFactory(t *testing.T) {
shim.Stop()
}
+func TestSchedulingLoopsShutdownOnStop(t *testing.T) {
+ cluster := MockScheduler{}
+ cluster.init()
+ assert.NilError(t, cluster.start(), "failed to start cluster")
+ assert.Check(t, cluster.started.Load(), "cluster should be started")
+
+ // Ensure both scheduling loops are active
+ err := utils.WaitForCondition(func() bool {
+ return utils.CountGoroutines("(*KubernetesShim).doScheduling")
== 2
+ }, 10*time.Millisecond, 5*time.Second)
+ assert.NilError(t, err, "both scheduling loops should be running after
cluster start")
+
+ // Verify stop terminates both scheduling loops
+ cluster.stop()
+ err = utils.WaitForCondition(func() bool {
+ return utils.CountGoroutines("(*KubernetesShim).doScheduling")
== 0
+ }, 10*time.Millisecond, 5*time.Second)
+ assert.NilError(t, err, "both scheduling loops should terminate when
scheduler is stopped")
+}
+
func TestNewCallbackWithCancel(t *testing.T) {
mockedAPIProvider := client.NewMockedAPIProvider(false)
shimCtx := cache.NewContext(mockedAPIProvider)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
