diff --git a/pkg/defaults/timeouts.go b/pkg/defaults/timeouts.go index 075be8cb7..c0b123f9e 100644 --- a/pkg/defaults/timeouts.go +++ b/pkg/defaults/timeouts.go @@ -694,6 +694,10 @@ const ( // against a separate lister that lags that strongly-consistent read — a freshness the // client cannot observe. This bounds how long we let the webhook cache catch up. TrainJobAdmissionRetryTimeout = 1 * time.Minute + + // CREWorkloadRunTimeout is the maximum time to wait for a Cluster Readiness + // Engine WorkloadRun (NCCL or training/goodput) to reach a terminal condition. + CREWorkloadRunTimeout = 30 * time.Minute ) // Inference performance validation timeouts. diff --git a/validators/performance/consts.go b/validators/performance/consts.go index 097cdac24..13910a86c 100644 --- a/validators/performance/consts.go +++ b/validators/performance/consts.go @@ -16,12 +16,13 @@ package main // Cross-file string constants for the performance validator. const ( - apiGroupAPIExtensions = "apiextensions.k8s.io" - resourceCRDs = "customresourcedefinitions" - versionV1alpha1 = "v1alpha1" - versionV1beta1 = "v1beta1" - keyName = "name" - checkNameNCCLAllReduceBW = "nccl-all-reduce-bw" + apiGroupAPIExtensions = "apiextensions.k8s.io" + resourceCRDs = "customresourcedefinitions" + versionV1alpha1 = "v1alpha1" + versionV1beta1 = "v1beta1" + keyName = "name" + checkNameNCCLAllReduceBW = "nccl-all-reduce-bw" + checkNameCRENCCLAllReduceBW = "nccl-cre-all-reduce-bw" // nodeJobName is the name of both the NCCL worker replicatedJob and its // primary container in testdata/{accelerator}/{service}/runtime.yaml. diff --git a/validators/performance/cre_fabric.go b/validators/performance/cre_fabric.go new file mode 100644 index 000000000..457e00955 --- /dev/null +++ b/validators/performance/cre_fabric.go @@ -0,0 +1,91 @@ +// Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// +// 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 main + +import ( + "sort" + "strconv" +) + +// EKS H100 values align the CRE WorkloadRun with AICR's proven EFA runtime. +// CRE's compiled WorkloadRun overrides do not swap the image and mpirun path, +// so AICR supplies the complete profile until CRE carries it itself. +const ( + creEFANCCLImage = "public.ecr.aws/hpc-cloud/nccl-tests:cuda12.8.1-efa1.43.2-ofiv1.16.3-ncclv2.27.7-1-testsv2.16.9" + creEFAMpirun = "/opt/amazon/openmpi/bin/mpirun" + creEFANCCLBin = "/opt/nccl-tests/build/all_reduce_perf" + + creEFAResource = "vpc.amazonaws.com/efa" + creEFACountH100 = "32" +) + +// creFabricProfile is the WorkloadRun image, MPI, environment, and +// extended-resource configuration for EKS H100. +type creFabricProfile struct { + image string + mpirunPath string + binary string + env map[string]string + mpiArgs []string + extraLimits map[string]string +} + +func creEKSH100EFAProfile() creFabricProfile { + env := map[string]string{ + "NCCL_DEBUG": "INFO", + "PATH": "$PATH:/opt/amazon/efa/bin:/usr/bin", + "FI_EFA_USE_DEVICE_RDMA": "1", + "FI_PROVIDER": "efa", + "NCCL_SOCKET_IFNAME": "eth0", + // Last -x wins over CRE's compiled AWS mpiArgs NCCL_NET_PLUGIN=none. + "NCCL_NET_PLUGIN": "ofi", + } + return creFabricProfile{ + image: creEFANCCLImage, + mpirunPath: creEFAMpirun, + binary: creEFANCCLBin, + env: env, + mpiArgs: mpiArgsFromEnv(env), + extraLimits: map[string]string{ + creEFAResource: creEFACountH100, + }, + } +} + +func mpiArgsFromEnv(env map[string]string) []string { + keys := make([]string, 0, len(env)) + for k := range env { + keys = append(keys, k) + } + sort.Strings(keys) + args := make([]string, 0, len(env)*2) + for _, k := range keys { + args = append(args, "-x", k+"="+env[k]) + } + return args +} + +func creResourceRequirements(gpuPerNode int, extra map[string]string) map[string]any { + limits := map[string]any{ + "nvidia.com/gpu": strconv.Itoa(gpuPerNode), + } + for k, v := range extra { + limits[k] = v + } + return map[string]any{ + "limits": limits, + "requests": limits, + } +} diff --git a/validators/performance/cre_nccl.go b/validators/performance/cre_nccl.go new file mode 100644 index 000000000..94cf416ab --- /dev/null +++ b/validators/performance/cre_nccl.go @@ -0,0 +1,142 @@ +// Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// +// 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 main + +import ( + "context" + "fmt" + "log/slog" + "strconv" + + "github.com/NVIDIA/aicr/pkg/defaults" + aicrErrors "github.com/NVIDIA/aicr/pkg/errors" + k8spod "github.com/NVIDIA/aicr/pkg/k8s/pod" + "github.com/NVIDIA/aicr/pkg/recipe" + "github.com/NVIDIA/aicr/validators" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +func checkCRENCCLAllReduceBW(ctx *validators.Context) error { + constraint, found := findPerformanceConstraint(ctx, checkNameCRENCCLAllReduceBW) + if !found { + return validators.Skip(fmt.Sprintf("no %s constraint in recipe", checkNameCRENCCLAllReduceBW)) + } + actual, passed, err := validateCRENcclAllReduceBw(ctx, constraint) + return classifyNCCLAllReduceBWResult(checkNameCRENCCLAllReduceBW, constraint, actual, passed, err) +} + +func validateCRENcclAllReduceBw(ctx *validators.Context, constraint recipe.Constraint) (string, bool, error) { + if ctx.ValidationInput == nil { + return skipMsgNCCLNoInput, true, nil + } + service := ctx.ValidationInput.Criteria.Service + accelerator := ctx.ValidationInput.Criteria.Accelerator + if service != recipe.CriteriaServiceEKS || accelerator != recipe.CriteriaAcceleratorH100 { + return fmt.Sprintf("skipped - CRE NCCL currently supports only eks × h100, got %s × %s", service, accelerator), true, nil + } + + threshold, err := parseThreshold(constraint.Value) + if err != nil { + return "", false, err + } + + gpuConfig, err := determineGPUConfig(ctx, service, accelerator, ctx.NodeSelector) + if err != nil { + return "", false, aicrErrors.Wrap(aicrErrors.ErrCodeInternal, "failed to determine GPU configuration", err) + } + if gpuConfig.WorkerCount < 2 { + return skipMsgNCCLFewNodes, true, nil + } + + dyn := ctx.DynamicClient + if dyn == nil { + return "", false, aicrErrors.New(aicrErrors.ErrCodeInternal, "dynamic client is required to create a WorkloadRun") + } + + obj := buildCRENCCLWorkloadRun(ctx.Namespace, gpuConfig, ctx.NodeSelector) + + if err := deleteCREWorkloadRun(ctx.Ctx, dyn, ctx.Namespace, creNCCLRunName); err != nil { + return "", false, err + } + + defer func() { + if delErr := deleteCREWorkloadRun(context.Background(), dyn, ctx.Namespace, creNCCLRunName); delErr != nil { + slog.Warn("failed to delete CRE NCCL WorkloadRun", "error", delErr) + } + }() + + if err := createUnstructured(ctx.Ctx, dyn, workloadRunGVR, ctx.Namespace, obj); err != nil { + return "", false, err + } + + run, err := waitForWorkloadRunTerminal(ctx.Ctx, dyn, ctx.Namespace, creNCCLRunName) + if err != nil { + return "", false, err + } + if unstructuredConditionTrue(run, "Failed") { + return "", false, aicrErrors.New(aicrErrors.ErrCodeInternal, "CRE WorkloadRun Failed") + } + + bw, err := listMaxBusBandwidth(ctx.Ctx, dyn, ctx.Namespace, creNCCLRunName, run.GetCreationTimestamp()) + if err != nil { + return "", false, err + } + + logs, logErr := creLauncherLogs(ctx, run.GetCreationTimestamp()) + if logErr != nil { + return "", false, aicrErrors.Wrap(aicrErrors.ErrCodeInternal, "CRE launcher logs required for transport assertion", logErr) + } + if err := verifyTransportFromLogs(logs, variantNET); err != nil { + return "", false, err + } + + actual := strconv.FormatFloat(bw, 'f', 2, 64) + return actual, bw >= threshold, nil +} + +func creLauncherLogs(ctx *validators.Context, createdAt metav1.Time) (string, error) { + listCtx, cancel := context.WithTimeout(ctx.Ctx, defaults.DiagnosticTimeout) + defer cancel() + pods, err := ctx.Clientset.CoreV1().Pods(ctx.Namespace).List(listCtx, metav1.ListOptions{ + LabelSelector: "jobset.sigs.k8s.io/replicatedjob-name=launcher", + }) + if err != nil { + return "", aicrErrors.Wrap(aicrErrors.ErrCodeInternal, "failed to list CRE worker pods", err) + } + if len(pods.Items) == 0 { + return "", aicrErrors.New(aicrErrors.ErrCodeNotFound, "no CRE worker pods found for transport assertion") + } + pod := youngestLivePodSince(pods.Items, createdAt) + if pod == nil { + return "", aicrErrors.New(aicrErrors.ErrCodeNotFound, "no live CRE worker pods found for transport assertion") + } + return k8spod.GetPodLogs(listCtx, ctx.Clientset, ctx.Namespace, pod.Name, nodeJobName) +} + +func youngestLivePodSince(pods []corev1.Pod, createdAt metav1.Time) *corev1.Pod { + var best *corev1.Pod + for i := range pods { + p := &pods[i] + if p.DeletionTimestamp != nil || p.Status.Phase == corev1.PodFailed || + p.CreationTimestamp.Time.Before(createdAt.Time) { + continue + } + if best == nil || p.CreationTimestamp.After(best.CreationTimestamp.Time) { + best = p + } + } + return best +} diff --git a/validators/performance/cre_nccl_test.go b/validators/performance/cre_nccl_test.go new file mode 100644 index 000000000..d5d21baca --- /dev/null +++ b/validators/performance/cre_nccl_test.go @@ -0,0 +1,173 @@ +// Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// +// 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 main + +import ( + "testing" + "time" + + "github.com/NVIDIA/aicr/pkg/recipe" + v1 "github.com/NVIDIA/aicr/pkg/validator/v1" + "github.com/NVIDIA/aicr/validators" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +func TestCheckCRENCCLSkipsWithoutConstraint(t *testing.T) { + ctx := &validators.Context{ + ValidationInput: &v1.ValidationInput{ + Config: v1.ValidationConfig{ + Performance: &v1.ValidationPhase{ + Constraints: []recipe.Constraint{{Name: "nccl-all-reduce-bw", Value: ">= 300"}}, + }, + }, + }, + } + err := checkCRENCCLAllReduceBW(ctx) + if !validators.IsSkip(err) { + t.Fatalf("expected Skip without CRE constraint, got %v", err) + } +} + +func TestMaxBusBandwidthGBps(t *testing.T) { + tests := []struct { + name string + results []any + want float64 + wantErr bool + }{ + { + name: "max of string busBW", + results: []any{ + map[string]any{"sizeBytes": int64(8), "busBW": "12.5"}, + map[string]any{"sizeBytes": int64(16), "busBW": "400.25"}, + map[string]any{"sizeBytes": int64(32), "busBW": "90"}, + }, + want: 400.25, + }, + { + name: "float64 from JSON decoder", + results: []any{ + map[string]any{"busBW": float64(300)}, + }, + want: 300, + }, + { + name: "empty", + results: nil, + wantErr: true, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := maxBusBandwidthGBps(tt.results) + if (err != nil) != tt.wantErr { + t.Fatalf("error = %v, wantErr %v", err, tt.wantErr) + } + if !tt.wantErr && got != tt.want { + t.Errorf("got %v, want %v", got, tt.want) + } + }) + } +} + +func TestBuildCRENCCLWorkloadRunEKSUsesEFAImage(t *testing.T) { + cfg := &gpuConfiguration{WorkerCount: 2, GPUCountPerNode: 8, Namespace: "ns"} + obj := buildCRENCCLWorkloadRun("ns", cfg, map[string]string{"foo": "bar"}) + spec, ok := obj.Object["spec"].(map[string]any) + if !ok { + t.Fatal("spec is not a map") + } + if spec["image"] != creEFANCCLImage { + t.Errorf("image = %v, want EFA nccl-tests image", spec["image"]) + } + fw := spec["framework"].(map[string]any) + mpi := fw["mpi"].(map[string]any) + if mpi["mpirunPath"] != creEFAMpirun { + t.Errorf("mpirunPath = %v, want %s", mpi["mpirunPath"], creEFAMpirun) + } + if mpi["binary"] != creEFANCCLBin { + t.Errorf("binary = %v, want %s", mpi["binary"], creEFANCCLBin) + } + res := spec["resources"].(map[string]any) + limits := res["limits"].(map[string]any) + if limits[creEFAResource] != creEFACountH100 { + t.Errorf("efa limit = %v, want %s", limits[creEFAResource], creEFACountH100) + } + if spec["enableMNNVL"] != false { + t.Errorf("enableMNNVL = %v, want false", spec["enableMNNVL"]) + } +} + +func TestUnstructuredConditionTrue(t *testing.T) { + obj := buildCRENCCLWorkloadRun("ns", &gpuConfiguration{WorkerCount: 2, GPUCountPerNode: 8}, nil) + obj.Object["status"] = map[string]any{ + "conditions": []any{ + map[string]any{"type": "Succeeded", "status": "True"}, + }, + } + if !unstructuredConditionTrue(obj, "Succeeded") { + t.Fatal("expected Succeeded=True") + } + if unstructuredConditionTrue(obj, "Failed") { + t.Fatal("did not expect Failed") + } +} + +func TestYoungestLivePodSince(t *testing.T) { + cutoff := metav1.NewTime(time.Unix(50, 0)) + older := corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: "old", + CreationTimestamp: metav1.NewTime(time.Unix(1, 0)), + }, + Status: corev1.PodStatus{Phase: corev1.PodSucceeded}, + } + newer := corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: "new", + CreationTimestamp: metav1.NewTime(time.Unix(100, 0)), + }, + Status: corev1.PodStatus{Phase: corev1.PodRunning}, + } + failed := corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: "failed", + CreationTimestamp: metav1.NewTime(time.Unix(200, 0)), + }, + Status: corev1.PodStatus{Phase: corev1.PodFailed}, + } + got := youngestLivePodSince([]corev1.Pod{older, failed, newer}, cutoff) + if got == nil || got.Name != "new" { + t.Fatalf("got %#v, want new", got) + } +} + +func TestMeasurementBelongsToRun(t *testing.T) { + createdAt := metav1.NewTime(time.Unix(100, 0)) + obj := buildCRENCCLWorkloadRun("ns", &gpuConfiguration{WorkerCount: 2, GPUCountPerNode: 8}, nil) + obj.SetCreationTimestamp(metav1.NewTime(time.Unix(101, 0))) + obj.SetOwnerReferences([]metav1.OwnerReference{{Kind: "Workflow", Name: creNCCLRunName}}) + if !measurementBelongsToRun(obj, creNCCLRunName, createdAt) { + t.Fatal("expected matching measurement") + } + if measurementBelongsToRun(obj, "aicr-cre-nemo", createdAt) { + t.Fatal("measurement from another WorkloadRun must not match") + } + obj.SetCreationTimestamp(metav1.NewTime(time.Unix(99, 0))) + if measurementBelongsToRun(obj, creNCCLRunName, createdAt) { + t.Fatal("stale measurement must not match") + } +} diff --git a/validators/performance/cre_workloadrun.go b/validators/performance/cre_workloadrun.go new file mode 100644 index 000000000..512ea775c --- /dev/null +++ b/validators/performance/cre_workloadrun.go @@ -0,0 +1,301 @@ +// Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// +// 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 main + +import ( + "context" + "fmt" + "log/slog" + "sort" + "strconv" + + "github.com/NVIDIA/aicr/pkg/defaults" + aicrErrors "github.com/NVIDIA/aicr/pkg/errors" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/client-go/dynamic" +) + +const ( + creAPIGroup = "excalibur.nvidia.com" + creNCCLRunName = "aicr-cre-nccl" + creLogProfileNCCL = "nccl-bandwidth" +) + +var ( + workloadRunGVR = schema.GroupVersionResource{ + Group: creAPIGroup, Version: versionV1alpha1, Resource: "workloadruns", + } + bandwidthMeasurementGVR = schema.GroupVersionResource{ + Group: creAPIGroup, Version: versionV1alpha1, Resource: "bandwidthmeasurements", + } +) + +func buildCRENCCLWorkloadRun(namespace string, gpuConfig *gpuConfiguration, nodeSelector map[string]string) *unstructured.Unstructured { + profile := creEKSH100EFAProfile() + + env := make([]any, 0, len(profile.env)) + envKeys := make([]string, 0, len(profile.env)) + for k := range profile.env { + envKeys = append(envKeys, k) + } + sort.Strings(envKeys) + for _, k := range envKeys { + env = append(env, map[string]any{"name": k, "value": profile.env[k]}) + } + + spec := map[string]any{ + "image": profile.image, + "numNodes": int64(gpuConfig.WorkerCount), + "gpusPerNode": int64(gpuConfig.GPUCountPerNode), + "enableMNNVL": false, + "framework": map[string]any{ + "mpi": map[string]any{ + "binary": profile.binary, + "mpirunPath": profile.mpirunPath, + "args": []any{ + "-b", "8", + "-e", "16G", + "-f", "2", + "-n", "100", + "-N", "10", + }, + }, + }, + "bandwidthMeasurement": map[string]any{ + "logProfileRef": creLogProfileNCCL, + "sampleInterval": "30s", + "testType": "all_reduce", + }, + } + + if len(profile.mpiArgs) > 0 { + mpi := spec["framework"].(map[string]any)["mpi"].(map[string]any) + args := make([]any, len(profile.mpiArgs)) + for i, a := range profile.mpiArgs { + args[i] = a + } + mpi["mpiArgs"] = args + } + if len(env) > 0 { + spec["env"] = env + } + if len(profile.extraLimits) > 0 { + spec["resources"] = creResourceRequirements(gpuConfig.GPUCountPerNode, profile.extraLimits) + } + + addCRETarget(spec, gpuConfig, nodeSelector) + + return newCREWorkloadRun(namespace, creNCCLRunName, spec) +} + +func addCRETarget(spec map[string]any, gpuConfig *gpuConfiguration, nodeSelector map[string]string) { + target := map[string]any{} + if len(nodeSelector) > 0 { + target["nodeSelector"] = stringMapToAny(nodeSelector) + } else if len(gpuConfig.Nodes) > 0 { + names := make([]any, 0, len(gpuConfig.Nodes)) + for _, n := range gpuConfig.Nodes { + names = append(names, n.Name) + } + target["nodeNames"] = names + } + if len(target) > 0 { + spec["target"] = target + } +} + +func newCREWorkloadRun(namespace, name string, spec map[string]any) *unstructured.Unstructured { + return &unstructured.Unstructured{ + Object: map[string]any{ + "apiVersion": creAPIGroup + "/" + versionV1alpha1, + "kind": "WorkloadRun", + "metadata": map[string]any{ + "name": name, + "namespace": namespace, + }, + "spec": spec, + }, + } +} + +func stringMapToAny(in map[string]string) map[string]any { + out := make(map[string]any, len(in)) + for k, v := range in { + out[k] = v + } + return out +} + +func maxBusBandwidthGBps(results []any) (float64, error) { + if len(results) == 0 { + return 0, aicrErrors.New(aicrErrors.ErrCodeNotFound, "BandwidthMeasurement status.results is empty") + } + var maxBW float64 + var found bool + for _, raw := range results { + row, ok := raw.(map[string]any) + if !ok { + continue + } + bw, err := parseBusBWField(row["busBW"]) + if err != nil { + return 0, err + } + if !found || bw > maxBW { + maxBW = bw + found = true + } + } + if !found { + return 0, aicrErrors.New(aicrErrors.ErrCodeNotFound, "no busBW values in BandwidthMeasurement status.results") + } + return maxBW, nil +} + +func parseBusBWField(v any) (float64, error) { + switch t := v.(type) { + case string: + bw, err := strconv.ParseFloat(t, 64) + if err != nil { + return 0, aicrErrors.Wrap(aicrErrors.ErrCodeInvalidRequest, "invalid busBW", err) + } + return bw, nil + case float64: + return t, nil + case int64: + return float64(t), nil + case int: + return float64(t), nil + default: + return 0, aicrErrors.New(aicrErrors.ErrCodeInvalidRequest, + fmt.Sprintf("unsupported busBW type %T", v)) + } +} + +func unstructuredConditionTrue(obj *unstructured.Unstructured, condType string) bool { + conds, found, err := unstructured.NestedSlice(obj.Object, "status", "conditions") + if err != nil || !found { + return false + } + for _, c := range conds { + m, ok := c.(map[string]any) + if !ok { + continue + } + if fmt.Sprint(m["type"]) == condType && fmt.Sprint(m["status"]) == "True" { + return true + } + } + return false +} + +func waitForWorkloadRunTerminal(ctx context.Context, client dynamic.Interface, namespace, name string) (*unstructured.Unstructured, error) { + waitCtx, cancel := context.WithTimeout(ctx, defaults.CREWorkloadRunTimeout) + defer cancel() + + res := client.Resource(workloadRunGVR).Namespace(namespace) + + if obj, err := res.Get(waitCtx, name, metav1.GetOptions{}); err == nil { + if unstructuredConditionTrue(obj, "Succeeded") || unstructuredConditionTrue(obj, "Failed") { + return obj, nil + } + } + + watcher, err := res.Watch(waitCtx, metav1.ListOptions{FieldSelector: "metadata.name=" + name}) + if err != nil { + return nil, aicrErrors.Wrap(aicrErrors.ErrCodeInternal, "failed to watch WorkloadRun", err) + } + defer watcher.Stop() + + for { + select { + case <-waitCtx.Done(): + return nil, aicrErrors.Wrap(aicrErrors.ErrCodeTimeout, "timed out waiting for WorkloadRun", waitCtx.Err()) + case event, ok := <-watcher.ResultChan(): + if !ok { + obj, getErr := res.Get(waitCtx, name, metav1.GetOptions{}) + if getErr == nil && (unstructuredConditionTrue(obj, "Succeeded") || unstructuredConditionTrue(obj, "Failed")) { + return obj, nil + } + if waitCtx.Err() != nil { + return nil, aicrErrors.Wrap(aicrErrors.ErrCodeTimeout, "timed out waiting for WorkloadRun", waitCtx.Err()) + } + return nil, aicrErrors.New(aicrErrors.ErrCodeUnavailable, "WorkloadRun watch channel closed before terminal condition") + } + obj, ok := event.Object.(*unstructured.Unstructured) + if !ok { + continue + } + if unstructuredConditionTrue(obj, "Succeeded") || unstructuredConditionTrue(obj, "Failed") { + return obj, nil + } + } + } +} + +func listMaxBusBandwidth(ctx context.Context, client dynamic.Interface, namespace, runName string, createdAt metav1.Time) (float64, error) { + listCtx, cancel := context.WithTimeout(ctx, defaults.DiagnosticTimeout) + defer cancel() + list, err := client.Resource(bandwidthMeasurementGVR).Namespace(namespace).List(listCtx, metav1.ListOptions{}) + if err != nil { + return 0, aicrErrors.Wrap(aicrErrors.ErrCodeInternal, "failed to list BandwidthMeasurements", err) + } + var maxBW float64 + var found bool + for i := range list.Items { + if !measurementBelongsToRun(&list.Items[i], runName, createdAt) { + continue + } + results, _, _ := unstructured.NestedSlice(list.Items[i].Object, "status", "results") + bw, err := maxBusBandwidthGBps(results) + if err != nil { + slog.Debug("skipping BandwidthMeasurement without results", "name", list.Items[i].GetName(), "error", err) + continue + } + if !found || bw > maxBW { + maxBW = bw + found = true + } + } + if !found { + return 0, aicrErrors.New(aicrErrors.ErrCodeNotFound, "no BandwidthMeasurement with busBW results") + } + return maxBW, nil +} + +func measurementBelongsToRun(obj *unstructured.Unstructured, runName string, createdAt metav1.Time) bool { + if obj.GetCreationTimestamp().Time.Before(createdAt.Time) { + return false + } + for _, owner := range obj.GetOwnerReferences() { + if owner.Kind == "Workflow" && owner.Name == runName { + return true + } + } + return false +} + +func deleteCREWorkloadRun(ctx context.Context, client dynamic.Interface, namespace, name string) error { + delCtx, cancel := context.WithTimeout(ctx, defaults.DiagnosticTimeout) + defer cancel() + err := client.Resource(workloadRunGVR).Namespace(namespace).Delete(delCtx, name, metav1.DeleteOptions{}) + if err != nil && !apierrors.IsNotFound(err) { + return aicrErrors.Wrap(aicrErrors.ErrCodeInternal, "failed to delete WorkloadRun", err) + } + return nil +} diff --git a/validators/performance/main.go b/validators/performance/main.go index 5ccdd5707..b803e92a8 100644 --- a/validators/performance/main.go +++ b/validators/performance/main.go @@ -20,6 +20,7 @@ // performance nccl-all-reduce-bw // performance nccl-all-reduce-bw-net // performance nccl-all-reduce-bw-nvls +// performance nccl-cre-all-reduce-bw package main import ( @@ -28,9 +29,10 @@ import ( func main() { validators.Run(map[string]validators.CheckFunc{ - checkNameNCCLAllReduceBW: checkNCCLAllReduceBW, - "nccl-all-reduce-bw-net": checkNCCLAllReduceBWNET, - "nccl-all-reduce-bw-nvls": checkNCCLAllReduceBWNVLS, - "inference-perf": checkInferencePerf, + checkNameNCCLAllReduceBW: checkNCCLAllReduceBW, + "nccl-all-reduce-bw-net": checkNCCLAllReduceBWNET, + "nccl-all-reduce-bw-nvls": checkNCCLAllReduceBWNVLS, + checkNameCRENCCLAllReduceBW: checkCRENCCLAllReduceBW, + "inference-perf": checkInferencePerf, }) }