This is an automated email from the ASF dual-hosted git repository.
manirajv06 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 9e39c17c [YUNIKORN-3298] Stabilize predicates cleanup health checks
(#1040)
9e39c17c is described below
commit 9e39c17cf9727ed13d57fb999face47611ce4d24
Author: Venkateshwaran Shanmugham <[email protected]>
AuthorDate: Tue Jun 16 15:40:51 2026 +0530
[YUNIKORN-3298] Stabilize predicates cleanup health checks (#1040)
Wait for helper pods to fully terminate after deletion.
Replace the immediate post-cleanup health check with polling that waits for:
- non-negative node resources, and
- scheduler health checks to pass
Continue to fail if health does not recover within 60 seconds.
Closes: #1040
Signed-off-by: Manikandan R <[email protected]>
---
.../framework/helpers/yunikorn/rest_api_utils.go | 94 ++++++-
.../helpers/yunikorn/rest_api_utils_test.go | 279 +++++++++++++++++++++
test/e2e/predicates/predicates_test.go | 15 +-
3 files changed, 373 insertions(+), 15 deletions(-)
diff --git a/test/e2e/framework/helpers/yunikorn/rest_api_utils.go
b/test/e2e/framework/helpers/yunikorn/rest_api_utils.go
index ee582658..f7e9c669 100644
--- a/test/e2e/framework/helpers/yunikorn/rest_api_utils.go
+++ b/test/e2e/framework/helpers/yunikorn/rest_api_utils.go
@@ -27,6 +27,7 @@ import (
"io"
"net/http"
"net/url"
+ "strings"
"time"
v1 "k8s.io/api/core/v1"
@@ -36,7 +37,12 @@ import (
"github.com/apache/yunikorn-k8shim/test/e2e/framework/configmanager"
)
-const DefaultPartition = "default"
+const (
+ DefaultPartition = "default"
+ // DefaultSchedulerHealthTimeout allows the periodic health checker
(default 30s interval)
+ // to refresh cached results after transient negative resources during
pod cleanup.
+ DefaultSchedulerHealthTimeout = 60
+)
type RClient struct {
BaseURL *url.URL
@@ -404,13 +410,39 @@ func (c *RClient) ValidateSchedulerConfig(cm
v1.ConfigMap) (*dao.ValidateConfRes
return validateConfResponse, err
}
-func GetFailedHealthChecks() (string, error) {
- restClient := RClient{}
- var failCheck string
- healthCheck, err := restClient.GetHealthCheck()
- if err != nil {
- return "", fmt.Errorf("failed to get scheduler health check
from API")
+func hasNegativeResourceValues(values map[string]int64) bool {
+ for _, value := range values {
+ if value < 0 {
+ return true
+ }
}
+ return false
+}
+
+// HasNegativeNodeResources reports whether any node has negative resource
values in the
+// fields checked by the scheduler's node-level "Negative resources" health
check
+// (yunikorn-core/pkg/scheduler/health_checker.go checkSchedulingContext):
+// - node.GetAllocatedResource().HasNegativeValue() -> NodeDAOInfo.Allocated
+// - node.GetAvailableResource().HasNegativeValue() -> NodeDAOInfo.Available
+// - node.GetCapacity().HasNegativeValue() -> NodeDAOInfo.Capacity
+// - node.GetOccupiedResource().HasNegativeValue() -> NodeDAOInfo.Occupied
+//
+// This is a subset of the full health check. Partition-level negative
resources,
+// consistency checks, and orphan allocations are covered only by
GetHealthCheck().
+func HasNegativeNodeResources(nodes []dao.NodeDAOInfo) bool {
+ for _, node := range nodes {
+ if hasNegativeResourceValues(node.Capacity) ||
+ hasNegativeResourceValues(node.Allocated) ||
+ hasNegativeResourceValues(node.Occupied) ||
+ hasNegativeResourceValues(node.Available) {
+ return true
+ }
+ }
+ return false
+}
+
+func formatFailedHealthChecks(healthCheck dao.SchedulerHealthDAOInfo) string {
+ var failCheck string
if !healthCheck.Healthy {
for _, check := range healthCheck.HealthChecks {
if !check.Succeeded {
@@ -418,7 +450,53 @@ func GetFailedHealthChecks() (string, error) {
}
}
}
- return failCheck, nil
+ return failCheck
+}
+
+func GetFailedHealthChecks() (string, error) {
+ restClient := RClient{}
+ healthCheck, err := restClient.GetHealthCheck()
+ if err != nil {
+ return "", fmt.Errorf("failed to get scheduler health check
from API")
+ }
+ return formatFailedHealthChecks(healthCheck), nil
+}
+
+func (c *RClient) WaitForSchedulerHealth(partition string, timeout int) error {
+ // Poll until both live node state and the cached health-check result
are clean.
+ // The health endpoint returns the last periodic check (default
interval: 30s), so the
+ // timeout should exceed one full interval. A persistent scheduler bug
still fails once
+ // the timeout elapses.
+ var lastFailure string
+ err := wait.PollUntilContextTimeout(context.Background(), time.Second,
time.Duration(timeout)*time.Second, false,
+ func(ctx context.Context) (bool, error) {
+ nodes, err := c.GetNodes(partition)
+ if err != nil {
+ return false, err
+ }
+ if nodes != nil && HasNegativeNodeResources(*nodes) {
+ lastFailure = "nodes API reported negative
resources"
+ return false, nil
+ }
+
+ healthCheck, err := c.GetHealthCheck()
+ if err != nil {
+ return false, err
+ }
+ if failedChecks :=
formatFailedHealthChecks(healthCheck); failedChecks != "" {
+ lastFailure = failedChecks
+ return false, nil
+ }
+ return true, nil
+ })
+ if err != nil {
+ return fmt.Errorf("scheduler did not become healthy within %ds:
%s", timeout, strings.TrimSpace(lastFailure))
+ }
+ return nil
+}
+
+func WaitForSchedulerHealth(timeout int) error {
+ return (&RClient{}).WaitForSchedulerHealth(DefaultPartition, timeout)
}
func (c *RClient) GetQueue(partition string, queueName string, withChildren
bool) (*dao.PartitionQueueDAOInfo, error) {
diff --git a/test/e2e/framework/helpers/yunikorn/rest_api_utils_test.go
b/test/e2e/framework/helpers/yunikorn/rest_api_utils_test.go
new file mode 100644
index 00000000..a2b7e1e0
--- /dev/null
+++ b/test/e2e/framework/helpers/yunikorn/rest_api_utils_test.go
@@ -0,0 +1,279 @@
+/*
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+package yunikorn
+
+import (
+ "encoding/json"
+ "net/http"
+ "net/http/httptest"
+ "net/url"
+ "strings"
+ "sync/atomic"
+ "testing"
+
+ "github.com/apache/yunikorn-core/pkg/webservice/dao"
+ "github.com/apache/yunikorn-k8shim/test/e2e/framework/configmanager"
+)
+
+func TestHasNegativeNodeResources(t *testing.T) {
+ healthyNode := dao.NodeDAOInfo{
+ NodeID: "node-1",
+ Capacity: map[string]int64{"memory": 1000, "vcore": 4},
+ Allocated: map[string]int64{"memory": 100, "vcore": 1},
+ Occupied: map[string]int64{"memory": 0, "vcore": 0},
+ Available: map[string]int64{"memory": 900, "vcore": 3},
+ }
+
+ tests := []struct {
+ name string
+ nodes []dao.NodeDAOInfo
+ want bool
+ }{
+ {
+ name: "healthy node",
+ nodes: []dao.NodeDAOInfo{healthyNode},
+ want: false,
+ },
+ {
+ name: "negative available resources",
+ nodes: []dao.NodeDAOInfo{{
+ NodeID: "yk8s-worker",
+ Capacity: map[string]int64{"memory": 1000,
"vcore": 4},
+ Allocated: map[string]int64{"memory": 100,
"vcore": 1},
+ Occupied: map[string]int64{"memory": 0,
"vcore": 0},
+ Available: map[string]int64{"memory": -1,
"vcore": 3},
+ }},
+ want: true,
+ },
+ {
+ name: "negative allocated resources",
+ nodes: []dao.NodeDAOInfo{func() dao.NodeDAOInfo {
+ node := healthyNode
+ node.Allocated = map[string]int64{"memory": -1,
"vcore": 1}
+ return node
+ }()},
+ want: true,
+ },
+ {
+ name: "negative capacity resources",
+ nodes: []dao.NodeDAOInfo{func() dao.NodeDAOInfo {
+ node := healthyNode
+ node.Capacity = map[string]int64{"memory": -1,
"vcore": 4}
+ return node
+ }()},
+ want: true,
+ },
+ {
+ name: "negative occupied resources",
+ nodes: []dao.NodeDAOInfo{func() dao.NodeDAOInfo {
+ node := healthyNode
+ node.Occupied = map[string]int64{"memory": -1,
"vcore": 0}
+ return node
+ }()},
+ want: true,
+ },
+ {
+ name: "empty node list",
+ nodes: []dao.NodeDAOInfo{},
+ want: false,
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ if got := HasNegativeNodeResources(tt.nodes); got !=
tt.want {
+ t.Fatalf("HasNegativeNodeResources() = %v, want
%v", got, tt.want)
+ }
+ })
+ }
+}
+
+func TestWaitForSchedulerHealth(t *testing.T) {
+ healthyNodes := []dao.NodeDAOInfo{{
+ NodeID: "node-1",
+ Capacity: map[string]int64{"memory": 1000, "vcore": 4},
+ Allocated: map[string]int64{"memory": 0, "vcore": 0},
+ Occupied: map[string]int64{"memory": 0, "vcore": 0},
+ Available: map[string]int64{"memory": 1000, "vcore": 4},
+ }}
+ healthyHealthCheck := dao.SchedulerHealthDAOInfo{Healthy: true}
+ unhealthyHealthCheck := dao.SchedulerHealthDAOInfo{
+ Healthy: false,
+ HealthChecks: []dao.HealthCheckInfo{{
+ Name: "Negative resources",
+ Succeeded: false,
+ DiagnosisMessage: `Nodes with negative resources:
["yk8s-worker"]`,
+ }},
+ }
+
+ var healthCheckCalls atomic.Int32
+ server := httptest.NewServer(http.HandlerFunc(func(w
http.ResponseWriter, r *http.Request) {
+ switch r.URL.Path {
+ case "/ws/v1/partition/default/nodes":
+ w.Header().Set("Content-Type", "application/json")
+ if err := json.NewEncoder(w).Encode(healthyNodes); err
!= nil {
+ t.Fatalf("failed to encode nodes response: %v",
err)
+ }
+ case "/ws/v1/scheduler/healthcheck":
+ w.Header().Set("Content-Type", "application/json")
+ call := healthCheckCalls.Add(1)
+ response := unhealthyHealthCheck
+ if call >= 2 {
+ response = healthyHealthCheck
+ }
+ if err := json.NewEncoder(w).Encode(response); err !=
nil {
+ t.Fatalf("failed to encode health check
response: %v", err)
+ }
+ default:
+ http.NotFound(w, r)
+ }
+ }))
+ defer server.Close()
+
+ serverURL, err := url.Parse(server.URL)
+ if err != nil {
+ t.Fatalf("failed to parse test server URL: %v", err)
+ }
+
+ client := &RClient{
+ BaseURL: serverURL,
+ httpClient: server.Client(),
+ }
+
+ if err := client.WaitForSchedulerHealth(configmanager.DefaultPartition,
5); err != nil {
+ t.Fatalf("WaitForSchedulerHealth() error = %v", err)
+ }
+ if healthCheckCalls.Load() < 2 {
+ t.Fatalf("expected health check to be polled until healthy, got
%d calls", healthCheckCalls.Load())
+ }
+}
+
+func TestWaitForSchedulerHealthNegativeNodes(t *testing.T) {
+ var nodeResponses atomic.Int32
+ healthyNodes := []dao.NodeDAOInfo{{
+ NodeID: "node-1",
+ Capacity: map[string]int64{"memory": 1000, "vcore": 4},
+ Allocated: map[string]int64{"memory": 0, "vcore": 0},
+ Occupied: map[string]int64{"memory": 0, "vcore": 0},
+ Available: map[string]int64{"memory": 1000, "vcore": 4},
+ }}
+ negativeNodes := []dao.NodeDAOInfo{{
+ NodeID: "yk8s-worker",
+ Capacity: map[string]int64{"memory": 1000, "vcore": 4},
+ Allocated: map[string]int64{"memory": 100, "vcore": 1},
+ Occupied: map[string]int64{"memory": 0, "vcore": 0},
+ Available: map[string]int64{"memory": -100, "vcore": 3},
+ }}
+
+ server := httptest.NewServer(http.HandlerFunc(func(w
http.ResponseWriter, r *http.Request) {
+ switch r.URL.Path {
+ case "/ws/v1/partition/default/nodes":
+ w.Header().Set("Content-Type", "application/json")
+ response := negativeNodes
+ if nodeResponses.Add(1) >= 2 {
+ response = healthyNodes
+ }
+ if err := json.NewEncoder(w).Encode(response); err !=
nil {
+ t.Fatalf("failed to encode nodes response: %v",
err)
+ }
+ case "/ws/v1/scheduler/healthcheck":
+ w.Header().Set("Content-Type", "application/json")
+ if err :=
json.NewEncoder(w).Encode(dao.SchedulerHealthDAOInfo{Healthy: true}); err !=
nil {
+ t.Fatalf("failed to encode health check
response: %v", err)
+ }
+ default:
+ http.NotFound(w, r)
+ }
+ }))
+ defer server.Close()
+
+ serverURL, err := url.Parse(server.URL)
+ if err != nil {
+ t.Fatalf("failed to parse test server URL: %v", err)
+ }
+
+ client := &RClient{
+ BaseURL: serverURL,
+ httpClient: server.Client(),
+ }
+
+ if err := client.WaitForSchedulerHealth(configmanager.DefaultPartition,
5); err != nil {
+ t.Fatalf("WaitForSchedulerHealth() error = %v", err)
+ }
+ if nodeResponses.Load() < 2 {
+ t.Fatalf("expected nodes API to be polled until resources
recovered, got %d calls", nodeResponses.Load())
+ }
+}
+
+func TestWaitForSchedulerHealthTimeout(t *testing.T) {
+ negativeNodes := []dao.NodeDAOInfo{{
+ NodeID: "yk8s-worker",
+ Capacity: map[string]int64{"memory": 1000, "vcore": 4},
+ Allocated: map[string]int64{"memory": 100, "vcore": 1},
+ Occupied: map[string]int64{"memory": 0, "vcore": 0},
+ Available: map[string]int64{"memory": -100, "vcore": 3},
+ }}
+ unhealthyHealthCheck := dao.SchedulerHealthDAOInfo{
+ Healthy: false,
+ HealthChecks: []dao.HealthCheckInfo{{
+ Name: "Negative resources",
+ Succeeded: false,
+ DiagnosisMessage: `Nodes with negative resources:
["yk8s-worker"]`,
+ }},
+ }
+
+ server := httptest.NewServer(http.HandlerFunc(func(w
http.ResponseWriter, r *http.Request) {
+ switch r.URL.Path {
+ case "/ws/v1/partition/default/nodes":
+ w.Header().Set("Content-Type", "application/json")
+ if err := json.NewEncoder(w).Encode(negativeNodes); err
!= nil {
+ t.Fatalf("failed to encode nodes response: %v",
err)
+ }
+ case "/ws/v1/scheduler/healthcheck":
+ w.Header().Set("Content-Type", "application/json")
+ if err :=
json.NewEncoder(w).Encode(unhealthyHealthCheck); err != nil {
+ t.Fatalf("failed to encode health check
response: %v", err)
+ }
+ default:
+ http.NotFound(w, r)
+ }
+ }))
+ defer server.Close()
+
+ serverURL, err := url.Parse(server.URL)
+ if err != nil {
+ t.Fatalf("failed to parse test server URL: %v", err)
+ }
+
+ client := &RClient{
+ BaseURL: serverURL,
+ httpClient: server.Client(),
+ }
+
+ err = client.WaitForSchedulerHealth(configmanager.DefaultPartition, 2)
+ if err == nil {
+ t.Fatal("WaitForSchedulerHealth() expected error for persistent
unhealthy state")
+ }
+ if !strings.Contains(err.Error(), "scheduler did not become healthy
within 2s") {
+ t.Fatalf("WaitForSchedulerHealth() error = %v, want timeout
message", err)
+ }
+ if !strings.Contains(err.Error(), "nodes API reported negative
resources") {
+ t.Fatalf("WaitForSchedulerHealth() error = %v, want nodes API
failure detail", err)
+ }
+}
diff --git a/test/e2e/predicates/predicates_test.go
b/test/e2e/predicates/predicates_test.go
index 941bcc5e..aa6f2a1b 100644
--- a/test/e2e/predicates/predicates_test.go
+++ b/test/e2e/predicates/predicates_test.go
@@ -58,6 +58,7 @@ func runPodAndGetNodeName(k *k8s.KubeCtl, conf
k8s.SleepPodConfig) string {
By("Explicitly delete pod here to free the resource it takes.")
err := k.DeletePod(pod.Name, pod.Namespace)
Ω(err).NotTo(HaveOccurred())
+ Ω(k.WaitForPodTerminated(pod.Namespace, pod.Name,
60*time.Second)).NotTo(HaveOccurred())
return pod.Spec.NodeName
}
@@ -116,6 +117,13 @@ var _ = Describe("Predicates", func() {
err = kClient.TearDownNamespace(n)
Ω(err).NotTo(HaveOccurred())
}
+
+ By("Wait for Yunikorn's health to recover after cleanup")
+ // Poll until live node state and cached health check are both
clean. Exits
+ // immediately once healthy; DefaultSchedulerHealthTimeout is a
ceiling (2× the
+ // default 30s health-check interval), not a fixed post-test
delay.
+ err =
yunikorn.WaitForSchedulerHealth(yunikorn.DefaultSchedulerHealthTimeout)
+ Ω(err).NotTo(HaveOccurred())
})
// Test Nodes does not have any label, hence it should be impossible to
schedule Pod with
@@ -1089,11 +1097,4 @@ var _ = Describe("Predicates", func() {
logEntries := yunikorn.AllocLogToStrings(log)
Ω(logEntries).To(ContainElement(MatchRegexp(".*free ports.*")),
"Log entry message mismatch")
})
-
- AfterEach(func() {
- By("Check Yunikorn's health")
- checks, err := yunikorn.GetFailedHealthChecks()
- Ω(err).NotTo(HaveOccurred())
- Ω(checks).To(Equal(""), checks)
- })
})
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
