This is an automated email from the ASF dual-hosted git repository. wilfred-s pushed a commit to branch branch-1.9 in repository https://gitbox.apache.org/repos/asf/yunikorn-k8shim.git
commit 148c1f808a29a76796fea7d7c2f4a691d5fe9e78 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]> (cherry picked from commit 9e39c17cf9727ed13d57fb999face47611ce4d24) --- .../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]
