Merge pull request #18053 from justinsb/gang_scheduling_with_kueue

[aiconformance] Add test for gang scheduling with kueue

Author: k8s-ci-robot

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

@@ -20,6 +20,7 @@ import (
 	"context"
 	"os"
 	"path/filepath"
+	"sync"
 	"testing"
 
 	"k8s.io/client-go/dynamic"
@@ -35,6 +36,12 @@ type ValidatorHarness struct {
 
 	dynamicClient dynamic.Interface
 	restConfig    *rest.Config
+
+	// mutex guards our mutable state
+	mutex sync.Mutex
+
+	// testNamespace is a per-test namespace to use for creating resources. It is lazily initialized when TestNamespace() is called.
+	testNamespace string
 }
 
 // NewValidatorHarness creates a new ValidatorHarness.
@@ -70,3 +77,9 @@ func NewValidatorHarness(t *testing.T) *ValidatorHarness {
 func (h *ValidatorHarness) Context() context.Context {
 	return h.t.Context()
 }
+
+// Skip allows the test to be skipped with a message, and ensures that the skip is recorded in the output.
+func (h *ValidatorHarness) Skip(message string) {
+	h.output.Skip(message)
+	h.t.Skip(message)
+}

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

@@ -17,6 +17,10 @@ limitations under the License.
 package validators
 
 import (
+	"fmt"
+	"strings"
+	"time"
+
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/client-go/dynamic"
@@ -103,3 +107,88 @@ func (h *ValidatorHarness) ListResourceSlices() []*ResourceSlice {
 	}
 	return out
 }
+
+// CRD is a wrapper around the CustomResourceDefinition type.
+type CRD struct {
+	u *unstructured.Unstructured
+}
+
+// Name returns the name of the CRD.
+func (d *CRD) Name() string {
+	return d.u.GetName()
+}
+
+var crdGVR = schema.GroupVersionResource{
+	Group:    "apiextensions.k8s.io",
+	Version:  "v1",
+	Resource: "customresourcedefinitions",
+}
+
+// ListCRDs lists all CRDs in the cluster.
+func (h *ValidatorHarness) ListCRDs() []*CRD {
+	objectList, err := h.DynamicClient().Resource(crdGVR).List(h.Context(), metav1.ListOptions{})
+	if err != nil {
+		h.Fatalf("failed to list CRDs: %v", err)
+	}
+	var out []*CRD
+	for i := range objectList.Items {
+		out = append(out, &CRD{u: &objectList.Items[i]})
+	}
+	return out
+}
+
+// HasCRD returns true if a CRD with the given name exists.
+func (h *ValidatorHarness) HasCRD(name string) bool {
+	for _, crd := range h.ListCRDs() {
+		if crd.Name() == name {
+			return true
+		}
+	}
+	return false
+}
+
+var namespaceGVR = schema.GroupVersionResource{
+	Group:    "",
+	Version:  "v1",
+	Resource: "namespaces",
+}
+
+var namespaceGVK = schema.GroupVersionKind{
+	Group:   "",
+	Version: "v1",
+	Kind:    "Namespace",
+}
+
+func (h *ValidatorHarness) TestNamespace() string {
+	h.mutex.Lock()
+	defer h.mutex.Unlock()
+
+	if h.testNamespace == "" {
+		prefix := strings.ToLower(h.t.Name())
+		prefix = strings.ReplaceAll(prefix, "/", "-")
+		prefix = strings.ReplaceAll(prefix, "_", "-")
+		ns := fmt.Sprintf("%s-%d", prefix, time.Now().Unix())
+
+		nsObj := &unstructured.Unstructured{}
+		nsObj.SetGroupVersionKind(namespaceGVK)
+		nsObj.SetName(ns)
+
+		h.Logf("Creating test namespace %q", ns)
+
+		if _, err := h.DynamicClient().Resource(namespaceGVR).Create(h.Context(), nsObj, metav1.CreateOptions{}); err != nil {
+			h.Fatalf("failed to create test namespace: %v", err)
+		}
+
+		h.testNamespace = ns
+
+		h.t.Cleanup(func() {
+			h.Logf("Deleting test namespace %q", ns)
+			err := h.DynamicClient().Resource(namespaceGVR).Delete(h.Context(), ns, metav1.DeleteOptions{})
+			if err != nil {
+				h.Logf("failed to delete test namespace: %v", err)
+			}
+		})
+	}
+
+	return h.testNamespace
+}

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

@@ -40,6 +40,11 @@ func (o *MarkdownOutput) Success(text string) {
 	o.printf("✓ %s\n", text)
 }
 
+// Skip writes a skip message to the markdown file, prefixed with a warning symbol.
+func (o *MarkdownOutput) Skip(message string) {
+	o.printf("&warning; SKIPPED: %s\n", message)
+}
+
 // Close closes the underlying file. It should be called when all output is done.
 func (o *MarkdownOutput) Close() error {
 	return o.f.Close()

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

@@ -32,6 +32,9 @@ type OutputSink interface {
 	// Success indicates a successful check, allowing the output sink to format it accordingly.
 	Success(text string)
 
+	// Skip indicates that a test was skipped, allowing the output sink to format it accordingly.
+	Skip(message string)
+
 	// Close closes the output sink and releases any resources.
 	io.Closer
 }
@@ -64,3 +67,9 @@ func (h *ValidatorHarness) Success(format string, args ...interface{}) {
 	h.output.Success(s)
 	h.t.Logf("SUCCESS: "+format, args...)
 }
+
+// RecordConformance records that a specific conformance test was passed.
+func (h *ValidatorHarness) RecordConformance(testName string) {
+	// We should gather these in a structured way, but for now we'll just log them.
+	h.Logf("Conformance %q passed", testName)
+}

tests/e2e/scenarios/ai-conformance/validators/scheduling-orchestration/gangscheduling_test.go

@@ -0,0 +1,57 @@
+/*
+Copyright The Kubernetes Authors.
+
+Licensed 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 dra_support
+
+import (
+	"fmt"
+	"testing"
+
+	"k8s.io/kops/tests/e2e/scenarios/ai-conformance/validators"
+)
+
+// TestGangScheduling_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) {
+	// 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.
+
+	h := validators.NewValidatorHarness(t)
+
+	if !h.HasCRD("localqueues.kueue.x-k8s.io") {
+		h.Skip("Kueue CRDs not found, skipping gang scheduling test via Kueue")
+	}
+
+	h.Logf("# Applying manifest")
+
+	jobName := "gangscheduling-kueue"
+	{
+		ns := h.TestNamespace()
+		h.ShellExec(fmt.Sprintf("kubectl apply --namespace %s -f %s", ns, "testdata/gangscheduling/gangscheduling-kueue.yaml"))
+	}
+
+	// Wait for Job completion
+	{
+		ns := h.TestNamespace()
+		h.Logf("# Waiting for Job to complete")
+		h.ShellExec(fmt.Sprintf("kubectl wait --namespace %s --for=condition=complete job/%s", ns, jobName))
+
+	}
+
+	h.Success("Gang scheduling via Kueue test completed successfully.")
+
+	h.RecordConformance("schedulingOrchestration/gang_scheduling")
+}

tests/e2e/scenarios/ai-conformance/validators/scheduling-orchestration/testdata/gangscheduling/gangscheduling-kueue.yaml

@@ -0,0 +1,58 @@
+apiVersion: kueue.x-k8s.io/v1beta1
+kind: ResourceFlavor
+metadata:
+  name: default-flavor
+---
+apiVersion: kueue.x-k8s.io/v1beta1
+kind: ClusterQueue
+metadata:
+  name: cluster-queue
+spec:
+  namespaceSelector: {} # Available to all namespaces
+  queueingStrategy: BestEffortFIFO # Default queueing strategy
+  resourceGroups:
+  - coveredResources: ["cpu", "memory", "ephemeral-storage"]
+    flavors:
+    - name: "default-flavor"
+      resources:
+      - name: "cpu"
+        nominalQuota: 10
+      - name: "memory"
+        nominalQuota: 10Gi
+      - name: "ephemeral-storage"
+        nominalQuota: 10Gi
+---
+apiVersion: kueue.x-k8s.io/v1beta1
+kind: LocalQueue
+metadata:
+  name: team1
+spec:
+  clusterQueue: cluster-queue
+---
+apiVersion: batch/v1
+kind: Job
+metadata:
+  name: gangscheduling-kueue
+  annotations:
+    kueue.x-k8s.io/queue-name: team1 # Point to the LocalQueue
+spec:
+  ttlSecondsAfterFinished: 60 # Job will be deleted after 60 seconds
+  parallelism: 3 # This Job will have 3 replicas running at the same time
+  completions: 3 # This Job requires 3 completions
+  suspend: true # Set to true to allow Kueue to control the Job when it starts
+  template:
+    spec:
+      containers:
+      - name: dummy-job
+        image: debian:latest
+        command: ["sleep", "10s"] # Sleep for 10 seconds
+        resources:
+          requests:
+            cpu: "500m"
+            memory: "512Mi"
+            ephemeral-storage: "512Mi"
+          limits:
+            cpu: "500m"
+            memory: "512Mi"
+            ephemeral-storage: "512Mi"
+      restartPolicy: Never