Merge pull request #18075 from justinsb/aiconformance_schedulingOrchestration_pod_autoscaling

[aiconformance]: test for schedulingOrchestration pod autoscaling

Author: k8s-ci-robot

tests/e2e/scenarios/ai-conformance/run-test.sh

@@ -134,6 +134,36 @@ helm upgrade -i kube-prometheus-stack \
   --set grafana.enabled=false \
   --wait
 
+# Prometheus Adapter
+# Bridges Prometheus metrics to the Kubernetes custom metrics API (custom.metrics.k8s.io),
+# required for HPA to scale on custom metrics like vllm:num_requests_waiting.
+# Note: colons in metric names are incompatible with the Kubernetes custom metrics
+# API (they appear in URL paths), so the adapter renames the metric with underscores.
+echo "Installing prometheus-adapter..."
+cat >prometheus-adapter-values.yaml <<'ADAPTER_EOF'
+prometheus:
+  url: http://kube-prometheus-stack-prometheus.monitoring.svc
+  port: 9090
+rules:
+  custom:
+  - seriesQuery: '{__name__=~"^vllm:num_requests_waiting$"}'
+    resources:
+      overrides:
+        namespace:
+          resource: "namespace"
+        pod:
+          resource: "pod"
+    name:
+      matches: ""
+      as: "vllm_num_requests_waiting"
+    metricsQuery: sum by(namespace, pod) (<<.Series>>)
+ADAPTER_EOF
+helm upgrade -i prometheus-adapter prometheus-community/prometheus-adapter \
+  --namespace monitoring \
+  -f prometheus-adapter-values.yaml \
+  --wait
+rm -f prometheus-adapter-values.yaml
+
 # NVIDIA GPU Operator
 # Manages the full NVIDIA stack: kernel driver, container toolkit, device plugin, DCGM exporter.
 # The driver is installed into /run/nvidia/driver on each node.
@@ -239,6 +269,11 @@ echo "Verifying Prometheus Stack..."
 kubectl rollout status deployment -n monitoring kube-prometheus-stack-operator --timeout=5m || echo "Warning: Prometheus Operator not ready yet"
 kubectl rollout status statefulset -n monitoring prometheus-kube-prometheus-stack-prometheus --timeout=5m || echo "Warning: Prometheus not ready yet"
 
+echo "Verifying Prometheus Adapter..."
+kubectl rollout status deployment -n monitoring prometheus-adapter --timeout=5m || echo "Warning: Prometheus Adapter not ready yet"
+echo "Checking custom metrics API registration..."
+kubectl get apiservice v1beta1.custom.metrics.k8s.io || echo "Warning: custom metrics API not registered"
+
 echo "Verifying DCGM Exporter in gpu-operator namespace..."
 kubectl rollout status daemonset -n gpu-operator nvidia-dcgm-exporter --timeout=5m || echo "Warning: DCGM Exporter not ready yet"
 
@@ -352,7 +387,7 @@ echo "AI Conformance Environment Setup Complete."
 
 # Now run the actual AI conformance tests
 cd "${REPO_ROOT}/tests/e2e/scenarios/ai-conformance/validators"
-go test -v ./... -timeout=60m
+go test -v -p 1 ./... -timeout=60m
 
 # Compile and write the conformance report
 cd "${REPO_ROOT}/tests/e2e/scenarios/ai-conformance"

tests/e2e/scenarios/ai-conformance/validators/kube.go

@@ -27,8 +27,10 @@ import (
 	"strings"
 	"time"
 
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/dynamic"
 	"k8s.io/kops/tests/e2e/scenarios/ai-conformance/testartifacts"
 
@@ -193,12 +195,29 @@ func (h *ValidatorHarness) TestNamespace() string {
 
 		h.t.Cleanup(func() {
 			ctx := context.WithoutCancel(h.Context())
-			h.dumpNamespaceResources(ctx, ns)
+			h.dumpNamespaceResources(ctx, ns, "cluster-info")
+
+			startTime := time.Now()
 
 			h.Logf("Deleting test namespace %q", ns)
-			err := h.DynamicClient().Resource(namespaceGVR).Delete(ctx, ns, metav1.DeleteOptions{})
-			if err != nil {
-				h.Logf("failed to delete test namespace: %v", err)
+			if err := h.DynamicClient().Resource(namespaceGVR).Delete(ctx, ns, metav1.DeleteOptions{}); err != nil {
+				h.Errorf("failed to delete test namespace: %v", err)
+			}
+
+			// Wait for namespace deletion to complete so that we don't have leftover namespaces consuming resources.
+			if err := wait.PollUntilContextTimeout(ctx, 2*time.Second, 5*time.Minute, false, func(ctx context.Context) (done bool, err error) {
+				if _, err := h.DynamicClient().Resource(namespaceGVR).Get(ctx, ns, metav1.GetOptions{}); err != nil {
+					if apierrors.IsNotFound(err) {
+						return true, nil
+					}
+					return false, fmt.Errorf("error checking for namespace deletion: %w", err)
+				}
+				return false, nil
+			}); err != nil {
+				h.Errorf("error waiting for namespace deletion: %v", err)
+				h.dumpNamespaceResources(ctx, ns, "namespace-deletion-failure-info")
+			} else {
+				h.Logf("Namespace deletion took %s", time.Since(startTime).Round(time.Second))
 			}
 		})
 	}
@@ -225,8 +244,8 @@ func (h *ValidatorHarness) ApplyManifest(defaultNamespace string, manifestPath s
 }
 
 // dumpNamespaceResources dumps key resources from the namespace to the artifacts directory for debugging.
-func (h *ValidatorHarness) dumpNamespaceResources(ctx context.Context, ns string) {
-	clusterInfoDir := testartifacts.PathForTestArtifact(h.t, "cluster-info")
+func (h *ValidatorHarness) dumpNamespaceResources(ctx context.Context, ns string, outputDir string) {
+	clusterInfoDir := testartifacts.PathForTestArtifact(h.t, outputDir)
 	clusterInfoDir = filepath.Join(clusterInfoDir, ns)
 
 	if err := os.MkdirAll(clusterInfoDir, 0o755); err != nil {
@@ -244,6 +263,7 @@ func (h *ValidatorHarness) dumpNamespaceResources(ctx context.Context, ns string
 	// Always include Events, Pods: they are usually not in the manifest, but are often critical for understanding failures.
 	resourceTypes["Events"] = true
 	resourceTypes["Pods"] = true
+	resourceTypes["Nodes"] = true
 
 	for resourceType := range resourceTypes {
 		filename := strings.ToLower(resourceType) + ".yaml"
@@ -252,13 +272,20 @@ func (h *ValidatorHarness) dumpNamespaceResources(ctx context.Context, ns string
 		}
 	}
 
+	describeResourcesTypes := []string{"nodes", "pods"}
+	for _, describeResourcesType := range describeResourcesTypes {
+		filename := strings.ToLower(describeResourcesType) + ".txt"
+		if err := h.kubectlDescribeResource(ctx, ns, describeResourcesType, filepath.Join(clusterInfoDir, filename)); err != nil {
+			h.Logf("failed to kubectl describe resource %s: %v", describeResourcesType, err)
+		}
+	}
+
 	if err := h.dumpPodLogs(ctx, ns, clusterInfoDir); err != nil {
 		h.Logf("failed to dump pod logs: %v", err)
 	}
 }
 
 // dumpResource runs kubectl get for a resource type and writes the output to a file.
-// Errors are logged but do not fail the test.
 func (h *ValidatorHarness) dumpResource(ctx context.Context, ns string, resourceType string, outputPath string) error {
 	args := []string{"get", resourceType}
 	if ns != "" {
@@ -282,6 +309,29 @@ func (h *ValidatorHarness) dumpResource(ctx context.Context, ns string, resource
 	return nil
 }
 
+// dumpResource runs kubectl describe for a resource type and writes the output to a file.
+func (h *ValidatorHarness) kubectlDescribeResource(ctx context.Context, ns string, resourceType string, outputPath string) error {
+	args := []string{"describe", resourceType}
+	if ns != "" {
+		args = append(args, "-n", ns)
+	}
+	cmd := exec.CommandContext(ctx, "kubectl", args...)
+	var stdout bytes.Buffer
+	var stderr bytes.Buffer
+	cmd.Stdout = &stdout
+	cmd.Stderr = &stderr
+
+	if err := cmd.Run(); err != nil {
+		return fmt.Errorf("failed to kubectl describe %s in namespace %s: %v (stderr: %s)", resourceType, ns, err, stderr.String())
+	}
+
+	if err := os.WriteFile(outputPath, stdout.Bytes(), 0o644); err != nil {
+		return fmt.Errorf("failed to write %s describe output to %s: %w", resourceType, outputPath, err)
+	}
+
+	return nil
+}
+
 // dumpPodLogs collects logs from all pods in the namespace and writes them to individual files.
 func (h *ValidatorHarness) dumpPodLogs(ctx context.Context, ns string, clusterInfoDir string) error {
 	podLogsDir := filepath.Join(clusterInfoDir, "pod-logs")

tests/e2e/scenarios/ai-conformance/validators/markdown.go

@@ -110,10 +110,19 @@ func (o *MarkdownOutput) BeforeShellExec(command string) {
 
 // AfterShellExec writes the result of the executed shell command to the markdown file in a formatted code block.
 func (o *MarkdownOutput) AfterShellExec(command string, results *CommandResult) {
-	o.printf("```bash")
-	o.printf("%s", results.Stdout())
-	o.printf("%s", results.Stderr())
 	o.printf("```\n")
+	stdout := strings.TrimSpace(results.Stdout())
+	stderr := strings.TrimSpace(results.Stderr())
+	if stdout != "" {
+		o.printf("%s", stdout)
+	}
+	if stderr != "" {
+		if stdout != "" {
+			o.printf("\n")
+		}
+		o.printf("%s", stderr)
+	}
+	o.printf("\n```\n")
 
 	if results.Err() != nil {
 		o.printf("Error:\n```\n%v\n```\n", results.Err())

tests/e2e/scenarios/ai-conformance/validators/operator/robust_controller/testdata/rayjob-sample.yaml

@@ -47,6 +47,7 @@ spec:
       #pod template
       template:
         spec:
+          terminationGracePeriodSeconds: 5
           containers:
           - name: ray-head
             image: rayproject/ray:2.46.0

…ing-orchestration/gangscheduling_test.go …on/gangscheduling/gangscheduling_test.gotests/e2e/scenarios/ai-conformance/validators/scheduling-orchestration/gangscheduling_test.go renamed to tests/e2e/scenarios/ai-conformance/validators/scheduling-orchestration/gangscheduling/gangscheduling_test.go tests/e2e/scenarios/ai-conformance/validators/scheduling-orchestration/gangscheduling_test.go renamed to tests/e2e/scenarios/ai-conformance/validators/scheduling-orchestration/gangscheduling/gangscheduling_test.go

@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package dra_support
+package gangscheduling
 
 import (
 	"fmt"
@@ -23,9 +23,9 @@ import (
 	"k8s.io/kops/tests/e2e/scenarios/ai-conformance/validators"
 )
 
-// TestGangScheduling_ViaKueue corresponds to the schedulingOrchestration/gang_scheduling scenario,
+// TestSchedulingOrchestration_GangScheduling_ViaKueue corresponds to the schedulingOrchestration/gang_scheduling scenario,
 // for the case that the vendor chooses to demonstrate gang scheduling support via Kueue.
-func TestGangScheduling_ViaKueue(t *testing.T) {
+func TestSchedulingOrchestration_GangScheduling_ViaKueue(t *testing.T) {
 	// Description:
 	//   The platform must allow for the installation and successful operation of at least one gang scheduling solution that ensures all-or-nothing scheduling for distributed AI workloads (e.g. Kueue, Volcano, etc.) To be conformant, the vendor must demonstrate that their platform can successfully run at least one such solution.
 
@@ -44,7 +44,7 @@ func TestGangScheduling_ViaKueue(t *testing.T) {
 
 		h.Logf("Creating a Kueue Job that requires gang scheduling")
 		ns := h.TestNamespace()
-		h.ApplyManifest(ns, "testdata/gangscheduling/gangscheduling-kueue.yaml")
+		h.ApplyManifest(ns, "testdata/gangscheduling-kueue.yaml")
 
 		h.Logf("Waiting for Job to complete")
 		h.ShellExec(fmt.Sprintf("kubectl wait --namespace %s --for=condition=complete job/%s --timeout=300s", ns, jobName))