diff --git a/src/compute-plane-services/nvca/pkg/storage/BUILD.bazel b/src/compute-plane-services/nvca/pkg/storage/BUILD.bazel index eb31ee50c..a3637d9bf 100644 --- a/src/compute-plane-services/nvca/pkg/storage/BUILD.bazel +++ b/src/compute-plane-services/nvca/pkg/storage/BUILD.bazel @@ -39,7 +39,6 @@ go_library( "//src/compute-plane-services/nvca/pkg/nodefeatures", "//src/compute-plane-services/nvca/pkg/nvca/enforce/kaischeduler", "//src/compute-plane-services/nvca/pkg/nvca/health", - "//src/compute-plane-services/nvca/pkg/storage/cacheprobe", "//src/compute-plane-services/nvca/pkg/types", "//src/compute-plane-services/nvca/vendor/github.com/NVIDIA/nvcf/src/libraries/go/lib/pkg/core", "//src/compute-plane-services/nvca/vendor/github.com/NVIDIA/nvcf/src/libraries/go/lib/pkg/icms-translate/translate/common", @@ -140,7 +139,6 @@ go_test( "//src/compute-plane-services/nvca/pkg/featureflag", "//src/compute-plane-services/nvca/pkg/featureflag/mock", "//src/compute-plane-services/nvca/pkg/nodefeatures", - "//src/compute-plane-services/nvca/pkg/storage/cacheprobe", "//src/compute-plane-services/nvca/pkg/types", "//src/compute-plane-services/nvca/vendor/github.com/NVIDIA/nvcf/src/libraries/go/lib/pkg/core", "//src/compute-plane-services/nvca/vendor/github.com/NVIDIA/nvcf/src/libraries/go/lib/pkg/icms-translate/translate/common", diff --git a/src/compute-plane-services/nvca/pkg/storage/cacheprobe/BUILD.bazel b/src/compute-plane-services/nvca/pkg/storage/cacheprobe/BUILD.bazel deleted file mode 100644 index b21c46828..000000000 --- a/src/compute-plane-services/nvca/pkg/storage/cacheprobe/BUILD.bazel +++ /dev/null @@ -1,43 +0,0 @@ -load("@rules_go//go:def.bzl", "go_library", "go_test") - -go_library( - name = "cacheprobe", - srcs = [ - "configmap.go", - "probe.go", - ], - importpath = "github.com/NVIDIA/nvcf/src/compute-plane-services/nvca/pkg/storage/cacheprobe", - visibility = ["//visibility:public"], - deps = [ - "//src/compute-plane-services/nvca/vendor/k8s.io/api/core/v1:core", - "//src/compute-plane-services/nvca/vendor/k8s.io/apimachinery/pkg/api/errors", - "//src/compute-plane-services/nvca/vendor/k8s.io/apimachinery/pkg/api/resource", - "//src/compute-plane-services/nvca/vendor/k8s.io/apimachinery/pkg/apis/meta/v1:meta", - "//src/compute-plane-services/nvca/vendor/k8s.io/apimachinery/pkg/util/wait", - "//src/compute-plane-services/nvca/vendor/sigs.k8s.io/controller-runtime/pkg/client", - "//src/compute-plane-services/nvca/vendor/sigs.k8s.io/controller-runtime/pkg/log", - ], -) - -alias( - name = "go_default_library", - actual = ":cacheprobe", - visibility = ["//visibility:public"], -) - -go_test( - name = "cacheprobe_test", - srcs = ["cacheprobe_test.go"], - embed = [":cacheprobe"], - deps = [ - "//src/compute-plane-services/nvca/vendor/github.com/stretchr/testify/assert", - "//src/compute-plane-services/nvca/vendor/github.com/stretchr/testify/require", - "//src/compute-plane-services/nvca/vendor/k8s.io/api/core/v1:core", - "//src/compute-plane-services/nvca/vendor/k8s.io/apimachinery/pkg/api/errors", - "//src/compute-plane-services/nvca/vendor/k8s.io/apimachinery/pkg/apis/meta/v1:meta", - "//src/compute-plane-services/nvca/vendor/k8s.io/apimachinery/pkg/runtime", - "//src/compute-plane-services/nvca/vendor/sigs.k8s.io/controller-runtime/pkg/client", - "//src/compute-plane-services/nvca/vendor/sigs.k8s.io/controller-runtime/pkg/client/fake", - "//src/compute-plane-services/nvca/vendor/sigs.k8s.io/controller-runtime/pkg/client/interceptor", - ], -) diff --git a/src/compute-plane-services/nvca/pkg/storage/cacheprobe/cacheprobe_test.go b/src/compute-plane-services/nvca/pkg/storage/cacheprobe/cacheprobe_test.go deleted file mode 100644 index ed5542d51..000000000 --- a/src/compute-plane-services/nvca/pkg/storage/cacheprobe/cacheprobe_test.go +++ /dev/null @@ -1,213 +0,0 @@ -/* -SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. -SPDX-License-Identifier: Apache-2.0 - -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 cacheprobe - -import ( - "context" - "errors" - "testing" - "time" - - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" - corev1 "k8s.io/api/core/v1" - apierrors "k8s.io/apimachinery/pkg/api/errors" - metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "k8s.io/apimachinery/pkg/runtime" - "sigs.k8s.io/controller-runtime/pkg/client" - "sigs.k8s.io/controller-runtime/pkg/client/fake" - "sigs.k8s.io/controller-runtime/pkg/client/interceptor" -) - -func newFakeClient(t *testing.T, objs ...runtime.Object) *fake.ClientBuilder { - t.Helper() - sch := runtime.NewScheme() - require.NoError(t, corev1.AddToScheme(sch)) - b := fake.NewClientBuilder().WithScheme(sch) - for _, o := range objs { - b = b.WithRuntimeObjects(o) - } - return b -} - -const probeNS = "model-cache-system" - -func TestStateStore_SaveLoadGetStrategy(t *testing.T) { - c := newFakeClient(t).Build() - store := NewStateStore(c, probeNS) - sc := "nvcf-miniservice-sc" - - // No ConfigMap yet -> fallback. - got, err := store.GetStrategy(t.Context(), sc) - require.NoError(t, err) - assert.Equal(t, StrategyFallback, got) - - // Persist ROX supported -> GetStrategy returns ROX, round-trips via Load. - results := map[string]Result{ - ResultKey(sc, StrategyROX): {State: StateSupported, CheckedAt: time.Now(), TTL: 3600}, - ResultKey(sc, StrategyRWX): {State: StateSupported, CheckedAt: time.Now(), TTL: 3600}, - } - require.NoError(t, store.Save(t.Context(), results)) - - loaded, err := store.Load(t.Context()) - require.NoError(t, err) - assert.Equal(t, StateSupported, loaded[ResultKey(sc, StrategyROX)].State) - - got, err = store.GetStrategy(t.Context(), sc) - require.NoError(t, err) - assert.Equal(t, StrategyROX, got, "ROX preferred when both supported") - - // Only RWX supported -> RWX. - require.NoError(t, store.Save(t.Context(), map[string]Result{ - ResultKey(sc, StrategyROX): {State: StateUnsupported, CheckedAt: time.Now(), TTL: 3600}, - ResultKey(sc, StrategyRWX): {State: StateSupported, CheckedAt: time.Now(), TTL: 3600}, - })) - got, err = store.GetStrategy(t.Context(), sc) - require.NoError(t, err) - assert.Equal(t, StrategyRWX, got) - - // Expired ROX -> fallback. - require.NoError(t, store.Save(t.Context(), map[string]Result{ - ResultKey(sc, StrategyROX): {State: StateSupported, CheckedAt: time.Now().Add(-2 * time.Hour), TTL: 60}, - })) - got, err = store.GetStrategy(t.Context(), sc) - require.NoError(t, err) - assert.Equal(t, StrategyFallback, got) -} - -func TestIsExpired(t *testing.T) { - assert.False(t, isExpired(Result{TTL: 0, CheckedAt: time.Now().Add(-24 * time.Hour)}), "TTL<=0 never expires") - assert.False(t, isExpired(Result{TTL: 3600, CheckedAt: time.Now()})) - assert.True(t, isExpired(Result{TTL: 60, CheckedAt: time.Now().Add(-2 * time.Minute)})) -} - -func TestProber_createProbePVC(t *testing.T) { - c := newFakeClient(t).Build() - p := NewProber(c, probeNS, "nvcf-miniservice-sc", 3600) - - require.NoError(t, p.createProbePVC(t.Context(), "probe-pvc", corev1.ReadOnlyMany)) - - pvc := &corev1.PersistentVolumeClaim{} - require.NoError(t, c.Get(t.Context(), client.ObjectKey{Name: "probe-pvc", Namespace: probeNS}, pvc)) - assert.Equal(t, []corev1.PersistentVolumeAccessMode{corev1.ReadOnlyMany}, pvc.Spec.AccessModes) - require.NotNil(t, pvc.Spec.StorageClassName) - assert.Equal(t, "nvcf-miniservice-sc", *pvc.Spec.StorageClassName) - - // Re-create errors with AlreadyExists: leftovers are deleted before each - // probe run, so an existing PVC must not be silently reused. - assert.True(t, apierrors.IsAlreadyExists(p.createProbePVC(t.Context(), "probe-pvc", corev1.ReadOnlyMany))) -} - -func TestProber_cleanupProbe(t *testing.T) { - pod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "probe-pod", Namespace: probeNS}} - pvc := &corev1.PersistentVolumeClaim{ - ObjectMeta: metav1.ObjectMeta{Name: "probe-pvc", Namespace: probeNS}, - Spec: corev1.PersistentVolumeClaimSpec{VolumeName: "probe-pv"}, - } - // Simulates a Retain-reclaim class: the PV outlives the PVC deletion and - // must be deleted explicitly, or one Released PV accumulates per probe. - pv := &corev1.PersistentVolume{ObjectMeta: metav1.ObjectMeta{Name: "probe-pv"}} - c := newFakeClient(t, pod, pvc, pv).Build() - p := NewProber(c, probeNS, "nvcf-miniservice-sc", 3600) - - p.cleanupProbe(t.Context(), "probe-pod", "probe-pvc") - - for name, obj := range map[string]client.Object{ - "pod": &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "probe-pod", Namespace: probeNS}}, - "pvc": &corev1.PersistentVolumeClaim{ObjectMeta: metav1.ObjectMeta{Name: "probe-pvc", Namespace: probeNS}}, - "pv": &corev1.PersistentVolume{ObjectMeta: metav1.ObjectMeta{Name: "probe-pv"}}, - } { - err := c.Get(t.Context(), client.ObjectKeyFromObject(obj), obj) - assert.Truef(t, apierrors.IsNotFound(err), "%s must be deleted by cleanupProbe", name) - } - - // Cleanup with nothing left is a no-op (all NotFound tolerated). - p.cleanupProbe(t.Context(), "probe-pod", "probe-pvc") -} - -func TestProber_waitForPodRunning(t *testing.T) { - running := &corev1.Pod{ - ObjectMeta: metav1.ObjectMeta{Name: "running", Namespace: probeNS}, - Status: corev1.PodStatus{Phase: corev1.PodRunning}, - } - failed := &corev1.Pod{ - ObjectMeta: metav1.ObjectMeta{Name: "failed", Namespace: probeNS}, - Status: corev1.PodStatus{Phase: corev1.PodFailed, Message: "boom"}, - } - c := newFakeClient(t, running, failed).Build() - p := NewProber(c, probeNS, "nvcf-miniservice-sc", 3600) - - assert.NoError(t, p.waitForPodRunning(t.Context(), "running")) - assert.Error(t, p.waitForPodRunning(t.Context(), "failed")) -} - -func TestStateStore_HasFreshResult(t *testing.T) { - c := newFakeClient(t).Build() - store := NewStateStore(c, probeNS) - sc := "nvcf-miniservice-sc" - - // No results yet -> not fresh (caller should probe). - fresh, err := store.HasFreshResult(t.Context(), sc) - require.NoError(t, err) - assert.False(t, fresh) - - // A fresh negative (Unsupported) result is still "fresh": GetStrategy - // returns Fallback, but the class should not be re-probed until TTL elapses. - require.NoError(t, store.Save(t.Context(), map[string]Result{ - ResultKey(sc, StrategyROX): {State: StateUnsupported, CheckedAt: time.Now(), TTL: 3600}, - ResultKey(sc, StrategyRWX): {State: StateUnsupported, CheckedAt: time.Now(), TTL: 3600}, - })) - got, err := store.GetStrategy(t.Context(), sc) - require.NoError(t, err) - assert.Equal(t, StrategyFallback, got) - fresh, err = store.HasFreshResult(t.Context(), sc) - require.NoError(t, err) - assert.True(t, fresh, "fresh negative result must be honoured (no re-probe)") - - // Once the negative result expires, it is no longer fresh -> re-probe. - require.NoError(t, store.Save(t.Context(), map[string]Result{ - ResultKey(sc, StrategyROX): {State: StateUnsupported, CheckedAt: time.Now().Add(-2 * time.Hour), TTL: 60}, - ResultKey(sc, StrategyRWX): {State: StateUnsupported, CheckedAt: time.Now().Add(-2 * time.Hour), TTL: 60}, - })) - fresh, err = store.HasFreshResult(t.Context(), sc) - require.NoError(t, err) - assert.False(t, fresh, "expired result must not be considered fresh") -} - -// TestProbeAccessMode_UnsupportedTTLCapped guards the negative-result TTL cap: -// a failed probe may be a transient environment problem, so it must not carry -// the full positive TTL and suppress re-probing (which would cascade terminal -// failures onto every storage request for that window). -func TestProbeAccessMode_UnsupportedTTLCapped(t *testing.T) { - c := newFakeClient(t).WithInterceptorFuncs(interceptor.Funcs{ - Create: func(context.Context, client.WithWatch, client.Object, ...client.CreateOption) error { - return errors.New("apiserver timeout") - }, - }).Build() - p := NewProber(c, probeNS, "some-sc", 3600) - - res := p.ProbeAccessMode(context.Background(), corev1.ReadOnlyMany) - assert.Equal(t, StateUnsupported, res.State) - assert.Equal(t, UnsupportedResultTTLSeconds, res.TTL, - "negative results must carry the capped TTL, not the positive TTL") - - // A prober configured with a TTL below the cap keeps its own (shorter) TTL. - pShort := NewProber(c, probeNS, "some-sc", 60) - res = pShort.ProbeAccessMode(context.Background(), corev1.ReadOnlyMany) - assert.Equal(t, 60, res.TTL) -} diff --git a/src/compute-plane-services/nvca/pkg/storage/cacheprobe/configmap.go b/src/compute-plane-services/nvca/pkg/storage/cacheprobe/configmap.go deleted file mode 100644 index bb1b47897..000000000 --- a/src/compute-plane-services/nvca/pkg/storage/cacheprobe/configmap.go +++ /dev/null @@ -1,154 +0,0 @@ -/* -SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. -SPDX-License-Identifier: Apache-2.0 - -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 cacheprobe - -import ( - "context" - "encoding/json" - "fmt" - "time" - - corev1 "k8s.io/api/core/v1" - apierrors "k8s.io/apimachinery/pkg/api/errors" - metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "sigs.k8s.io/controller-runtime/pkg/client" -) - -// ProbeConfigMapName is the ConfigMap that persists probe results in the -// control namespace. -const ProbeConfigMapName = "model-cache-probe-results" - -// StateStore persists probe results in a ConfigMap so the strategy survives -// restarts and is reused within its TTL instead of re-probing every reconcile. -type StateStore struct { - client client.Client - namespace string -} - -// NewStateStore creates a StateStore backed by a ConfigMap in the given namespace. -func NewStateStore(c client.Client, namespace string) *StateStore { - return &StateStore{client: c, namespace: namespace} -} - -// Save writes probe results to the ConfigMap, creating it if necessary. -func (s *StateStore) Save(ctx context.Context, results map[string]Result) error { - data := make(map[string]string, len(results)) - for key, result := range results { - b, err := json.Marshal(result) - if err != nil { - return fmt.Errorf("marshal probe result for %s: %w", key, err) - } - data[key] = string(b) - } - - cm := &corev1.ConfigMap{} - err := s.client.Get(ctx, client.ObjectKey{Name: ProbeConfigMapName, Namespace: s.namespace}, cm) - if apierrors.IsNotFound(err) { - cm = &corev1.ConfigMap{ - ObjectMeta: metav1.ObjectMeta{ - Name: ProbeConfigMapName, - Namespace: s.namespace, - Labels: map[string]string{managedByLabel: managedByValue, probeLabel: "true"}, - }, - Data: data, - } - if createErr := s.client.Create(ctx, cm); createErr != nil { - return fmt.Errorf("create probe ConfigMap: %w", createErr) - } - return nil - } - if err != nil { - return fmt.Errorf("get probe ConfigMap: %w", err) - } - - cm.Data = data - if err := s.client.Update(ctx, cm); err != nil { - return fmt.Errorf("update probe ConfigMap: %w", err) - } - return nil -} - -// Load reads persisted probe results from the ConfigMap. Missing ConfigMap -// yields a nil map (no results yet). -func (s *StateStore) Load(ctx context.Context) (map[string]Result, error) { - cm := &corev1.ConfigMap{} - if err := s.client.Get(ctx, client.ObjectKey{Name: ProbeConfigMapName, Namespace: s.namespace}, cm); err != nil { - if apierrors.IsNotFound(err) { - return nil, nil - } - return nil, fmt.Errorf("get probe ConfigMap: %w", err) - } - - results := make(map[string]Result, len(cm.Data)) - for key, val := range cm.Data { - var r Result - if err := json.Unmarshal([]byte(val), &r); err != nil { - return nil, fmt.Errorf("unmarshal probe result for %s: %w", key, err) - } - results[key] = r - } - return results, nil -} - -// GetStrategy returns the best persisted, unexpired strategy for the storage -// class, or StrategyFallback when results are missing/expired (signalling the -// caller to re-probe). -func (s *StateStore) GetStrategy(ctx context.Context, storageClassName string) (AccessModeStrategy, error) { - results, err := s.Load(ctx) - if err != nil { - return StrategyFallback, err - } - if results == nil { - return StrategyFallback, nil - } - - if r, ok := results[ResultKey(storageClassName, StrategyROX)]; ok && r.State == StateSupported && !isExpired(r) { - return StrategyROX, nil - } - if r, ok := results[ResultKey(storageClassName, StrategyRWX)]; ok && r.State == StateSupported && !isExpired(r) { - return StrategyRWX, nil - } - return StrategyFallback, nil -} - -// HasFreshResult reports whether an unexpired probe result (of any state, -// supported or unsupported) exists for the storage class. Callers use this to -// honour the TTL of a negative (Unsupported) result: GetStrategy returns -// StrategyFallback both when results are missing/expired and when a fresh probe -// determined the class is unsupported, so a Fallback result alone does not mean -// "re-probe". A fresh negative result means the class was recently probed and -// found unusable, and should not be re-probed until its TTL elapses. -func (s *StateStore) HasFreshResult(ctx context.Context, storageClassName string) (bool, error) { - results, err := s.Load(ctx) - if err != nil { - return false, err - } - for _, strategy := range []AccessModeStrategy{StrategyROX, StrategyRWX} { - if r, ok := results[ResultKey(storageClassName, strategy)]; ok && !isExpired(r) { - return true, nil - } - } - return false, nil -} - -func isExpired(r Result) bool { - if r.TTL <= 0 { - return false - } - return time.Since(r.CheckedAt) > time.Duration(r.TTL)*time.Second -} diff --git a/src/compute-plane-services/nvca/pkg/storage/cacheprobe/probe.go b/src/compute-plane-services/nvca/pkg/storage/cacheprobe/probe.go deleted file mode 100644 index 1be594e1b..000000000 --- a/src/compute-plane-services/nvca/pkg/storage/cacheprobe/probe.go +++ /dev/null @@ -1,328 +0,0 @@ -/* -SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. -SPDX-License-Identifier: Apache-2.0 - -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 cacheprobe runs runtime CSI capability probes against a Helm model -// cache storage class to decide the reader access-mode strategy (ROX preferred, -// RWX fallback) for non-NVMesh shared-filesystem backends (nvcf-miniservice-sc). -package cacheprobe - -import ( - "context" - "fmt" - "time" - - corev1 "k8s.io/api/core/v1" - apierrors "k8s.io/apimachinery/pkg/api/errors" - "k8s.io/apimachinery/pkg/api/resource" - metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "k8s.io/apimachinery/pkg/util/wait" - "sigs.k8s.io/controller-runtime/pkg/client" - "sigs.k8s.io/controller-runtime/pkg/log" -) - -const ( - // ProbeTimeout bounds how long a single probe waits for its pod to run. - ProbeTimeout = 60 * time.Second - // ProbePollInterval is the probe pod status poll cadence. - ProbePollInterval = 2 * time.Second - // ProbePVCSize is the (tiny) size requested for probe PVCs. - ProbePVCSize = "1Gi" - // UnsupportedResultTTLSeconds caps the TTL of Unsupported probe results. A - // negative result cannot be distinguished from a transient failure (CSI - // driver briefly unavailable, apiserver timeout, node pressure), so honoring - // the full positive TTL would let one bad probe poison every storage request - // for that whole window. Re-probe soon instead; a genuinely unsupported - // class just re-fails a cheap probe every few minutes. - UnsupportedResultTTLSeconds = 300 - // DefaultProbeImage is the image used by the probe pod. Override via - // Prober.Image for air-gapped clusters. - DefaultProbeImage = "busybox:1.36" - - managedByLabel = "app.kubernetes.io/managed-by" - managedByValue = "nvca" - probeLabel = "cache.nvcf.nvidia.com/probe" -) - -// ProbeState represents the result of a capability probe. -type ProbeState string - -const ( - StateUnknown ProbeState = "Unknown" - StateProbing ProbeState = "Probing" - StateSupported ProbeState = "Supported" - StateUnsupported ProbeState = "Unsupported" -) - -// AccessModeStrategy is the chosen read strategy based on probe results. -type AccessModeStrategy string - -const ( - // StrategyROX: readers mount ReadOnlyMany (preferred). - StrategyROX AccessModeStrategy = "ROX" - // StrategyRWX: readers mount ReadWriteMany in read-only mode. - StrategyRWX AccessModeStrategy = "RWX" - // StrategyFallback: no shared access mode works; caller falls back to - // non-shared (per-pod) behavior. - StrategyFallback AccessModeStrategy = "Fallback" -) - -// Result holds the outcome of probing a single access mode. -type Result struct { - State ProbeState `json:"state"` - CheckedAt time.Time `json:"checkedAt"` - Reason string `json:"reason,omitempty"` - TTL int `json:"ttlSeconds"` -} - -// Prober performs runtime CSI capability probes against a storage class by -// creating a tiny PVC + pod for each candidate access mode. -type Prober struct { - client client.Client - namespace string - storageClassName string - ttlSeconds int - // Image is the probe pod image; defaults to DefaultProbeImage. - Image string -} - -// NewProber creates a Prober for the given storage class. -func NewProber(c client.Client, namespace, storageClassName string, ttlSeconds int) *Prober { - return &Prober{ - client: c, - namespace: namespace, - storageClassName: storageClassName, - ttlSeconds: ttlSeconds, - Image: DefaultProbeImage, - } -} - -// ProbeAccessMode tests whether the storage class supports the given access -// mode by creating a tiny PVC and a pod that mounts it; Supported if the pod -// reaches Running/Succeeded within the timeout. Probe resources are always -// cleaned up. -func (p *Prober) ProbeAccessMode(ctx context.Context, mode corev1.PersistentVolumeAccessMode) Result { - logger := log.FromContext(ctx).WithValues("storageClass", p.storageClassName, "accessMode", string(mode)) - logger.Info("starting CSI capability probe") - - suffix := accessModeSuffix(mode) - pvcName := fmt.Sprintf("probe-%s-%s", p.storageClassName, suffix) - podName := fmt.Sprintf("probe-%s-%s", p.storageClassName, suffix) - - result := Result{State: StateProbing, CheckedAt: time.Now(), TTL: p.ttlSeconds} - - defer func() { - cleanupCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second) - defer cancel() - p.cleanupProbe(cleanupCtx, podName, pvcName) - }() - - // A crashed prior run can leave a completed probe pod/PVC behind, and a - // stale Succeeded pod must not be accepted as the current probe's result. - // Delete leftovers first; if deletion has not finished by the time the - // creates below run, they fail with AlreadyExists and the probe returns a - // short-TTL Unsupported result and retries, never a stale success. - p.cleanupProbe(ctx, podName, pvcName) - - // Unsupported results carry a capped TTL (see UnsupportedResultTTLSeconds): - // a failed probe may be a transient environment problem, not a property of - // the storage class, and must not suppress re-probing for the full window. - if err := p.createProbePVC(ctx, pvcName, mode); err != nil { - result.State, result.Reason = StateUnsupported, fmt.Sprintf("PVC creation failed: %v", err) - result.TTL = min(p.ttlSeconds, UnsupportedResultTTLSeconds) - return result - } - if err := p.createProbePod(ctx, podName, pvcName); err != nil { - result.State, result.Reason = StateUnsupported, fmt.Sprintf("Pod creation failed: %v", err) - result.TTL = min(p.ttlSeconds, UnsupportedResultTTLSeconds) - return result - } - if err := p.waitForPodRunning(ctx, podName); err != nil { - logger.Info("probe pod did not reach Running", "error", err) - result.State, result.Reason = StateUnsupported, fmt.Sprintf("Pod did not reach Running: %v", err) - result.TTL = min(p.ttlSeconds, UnsupportedResultTTLSeconds) - return result - } - - logger.Info("probe succeeded") - result.State = StateSupported - return result -} - -// DetermineStrategy probes ROX then RWX and returns the best supported strategy -// (or StrategyFallback), along with the per-mode results keyed by -// "_". -func (p *Prober) DetermineStrategy(ctx context.Context) (AccessModeStrategy, map[string]Result) { - results := make(map[string]Result) - - rox := p.ProbeAccessMode(ctx, corev1.ReadOnlyMany) - results[ResultKey(p.storageClassName, StrategyROX)] = rox - if rox.State == StateSupported { - return StrategyROX, results - } - - rwx := p.ProbeAccessMode(ctx, corev1.ReadWriteMany) - results[ResultKey(p.storageClassName, StrategyRWX)] = rwx - if rwx.State == StateSupported { - return StrategyRWX, results - } - - return StrategyFallback, results -} - -// ResultKey is the map/ConfigMap key for a storage class + strategy result. -func ResultKey(storageClassName string, strategy AccessModeStrategy) string { - return fmt.Sprintf("%s_%s", storageClassName, strategy) -} - -func (p *Prober) probeLabels() map[string]string { - return map[string]string{managedByLabel: managedByValue, probeLabel: "true"} -} - -func (p *Prober) createProbePVC(ctx context.Context, name string, mode corev1.PersistentVolumeAccessMode) error { - pvc := &corev1.PersistentVolumeClaim{ - ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: p.namespace, Labels: p.probeLabels()}, - Spec: corev1.PersistentVolumeClaimSpec{ - AccessModes: []corev1.PersistentVolumeAccessMode{mode}, - StorageClassName: &p.storageClassName, - Resources: corev1.VolumeResourceRequirements{ - Requests: corev1.ResourceList{corev1.ResourceStorage: resource.MustParse(ProbePVCSize)}, - }, - }, - } - // No AlreadyExists tolerance: leftovers are deleted before each probe run, - // so an existing object means that deletion is still in flight, and reusing - // it could accept a stale prior result. - return p.client.Create(ctx, pvc) -} - -func (p *Prober) createProbePod(ctx context.Context, name, pvcName string) error { - image := p.Image - if image == "" { - image = DefaultProbeImage - } - automount := false - pod := &corev1.Pod{ - ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: p.namespace, Labels: p.probeLabels()}, - Spec: corev1.PodSpec{ - AutomountServiceAccountToken: &automount, - RestartPolicy: corev1.RestartPolicyNever, - SecurityContext: &corev1.PodSecurityContext{ - RunAsNonRoot: boolPtr(true), - RunAsUser: int64Ptr(1000), - RunAsGroup: int64Ptr(1000), - FSGroup: int64Ptr(1000), - }, - Containers: []corev1.Container{{ - Name: "probe", - Image: image, - // Exit immediately: a successful mount + start is the signal. - // waitForPodRunning accepts PodSucceeded, so there is no need to - // keep the pod alive (which would only delay probe cleanup). - Command: []string{"sh", "-c", "echo probe-ok"}, - Resources: corev1.ResourceRequirements{ - Requests: corev1.ResourceList{ - corev1.ResourceCPU: resource.MustParse("10m"), - corev1.ResourceMemory: resource.MustParse("16Mi"), - }, - Limits: corev1.ResourceList{ - corev1.ResourceCPU: resource.MustParse("50m"), - corev1.ResourceMemory: resource.MustParse("32Mi"), - }, - }, - SecurityContext: &corev1.SecurityContext{ - ReadOnlyRootFilesystem: boolPtr(true), - Capabilities: &corev1.Capabilities{Drop: []corev1.Capability{"ALL"}}, - }, - VolumeMounts: []corev1.VolumeMount{{Name: "probe-vol", MountPath: "/mnt/probe"}}, - }}, - Volumes: []corev1.Volume{{ - Name: "probe-vol", - VolumeSource: corev1.VolumeSource{ - PersistentVolumeClaim: &corev1.PersistentVolumeClaimVolumeSource{ClaimName: pvcName}, - }, - }}, - }, - } - // See createProbePVC for why AlreadyExists is not tolerated here. - return p.client.Create(ctx, pod) -} - -func (p *Prober) waitForPodRunning(ctx context.Context, name string) error { - ctx, cancel := context.WithTimeout(ctx, ProbeTimeout) - defer cancel() - - return wait.PollUntilContextCancel(ctx, ProbePollInterval, true, func(ctx context.Context) (bool, error) { - pod := &corev1.Pod{} - if err := p.client.Get(ctx, client.ObjectKey{Namespace: p.namespace, Name: name}, pod); err != nil { - return false, nil - } - switch pod.Status.Phase { - case corev1.PodRunning, corev1.PodSucceeded: - return true, nil - case corev1.PodFailed: - return false, fmt.Errorf("probe pod failed: %s", pod.Status.Message) - default: - return false, nil - } - }) -} - -func (p *Prober) cleanupProbe(ctx context.Context, podName, pvcName string) { - logger := log.FromContext(ctx) - pod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: podName, Namespace: p.namespace}} - if err := p.client.Delete(ctx, pod); err != nil && !apierrors.IsNotFound(err) { - logger.Error(err, "failed to delete probe pod", "pod", podName) - } - - // Record the bound PV before deleting the PVC so it can be deleted - // explicitly below. - boundPVName := "" - pvc := &corev1.PersistentVolumeClaim{} - if err := p.client.Get(ctx, client.ObjectKey{Namespace: p.namespace, Name: pvcName}, pvc); err == nil { - boundPVName = pvc.Spec.VolumeName - } else if !apierrors.IsNotFound(err) { - logger.Error(err, "failed to get probe PVC before cleanup", "pvc", pvcName) - } - pvc = &corev1.PersistentVolumeClaim{ObjectMeta: metav1.ObjectMeta{Name: pvcName, Namespace: p.namespace}} - if err := p.client.Delete(ctx, pvc); err != nil && !apierrors.IsNotFound(err) { - logger.Error(err, "failed to delete probe PVC", "pvc", pvcName) - } - - // A storage class with reclaimPolicy Delete removes the PV with the PVC, - // but a Retain class leaves one Released PV object per probe. Delete the - // PV explicitly; NotFound is the normal Delete-reclaim case. - if boundPVName != "" { - pv := &corev1.PersistentVolume{ObjectMeta: metav1.ObjectMeta{Name: boundPVName}} - if err := p.client.Delete(ctx, pv); err != nil && !apierrors.IsNotFound(err) { - logger.Error(err, "failed to delete probe PV", "pv", boundPVName) - } - } -} - -func accessModeSuffix(mode corev1.PersistentVolumeAccessMode) string { - switch mode { - case corev1.ReadOnlyMany: - return "rox" - case corev1.ReadWriteMany: - return "rwx" - default: - return "unknown" - } -} - -func boolPtr(b bool) *bool { return &b } -func int64Ptr(i int64) *int64 { return &i } diff --git a/src/compute-plane-services/nvca/pkg/storage/modelcache.go b/src/compute-plane-services/nvca/pkg/storage/modelcache.go index 1e44c88cc..02dfd990a 100644 --- a/src/compute-plane-services/nvca/pkg/storage/modelcache.go +++ b/src/compute-plane-services/nvca/pkg/storage/modelcache.go @@ -48,7 +48,7 @@ import ( "github.com/NVIDIA/nvcf/src/compute-plane-services/nvca/internal/util/k8sutil" nvcav1new "github.com/NVIDIA/nvcf/src/compute-plane-services/nvca/pkg/apis/nvca/v1" nvcav2beta1 "github.com/NVIDIA/nvcf/src/compute-plane-services/nvca/pkg/apis/nvca/v2beta1" - "github.com/NVIDIA/nvcf/src/compute-plane-services/nvca/pkg/storage/cacheprobe" + "github.com/NVIDIA/nvcf/src/compute-plane-services/nvca/pkg/featureflag" "github.com/NVIDIA/nvcf/src/compute-plane-services/nvca/pkg/types" ) @@ -227,42 +227,6 @@ func (r *Reconciler) doModelCacheRouted(ctx context.Context, return res, err } -// sharedFSProbeTTLSeconds is how long a cached CSI ROX/RWX probe result for -// nvcf-miniservice-sc is reused before re-probing. -const sharedFSProbeTTLSeconds = 3600 - -// resolveSharedFSStrategy returns the cached probe strategy for nvcf-miniservice-sc, -// re-probing (and persisting the result) when no valid cached strategy exists. -func (r *Reconciler) resolveSharedFSStrategy(ctx context.Context) (cacheprobe.AccessModeStrategy, error) { - log := logf.FromContext(ctx) - store := cacheprobe.NewStateStore(r.Client, ModelCacheInitNamespace) - strategy, err := store.GetStrategy(ctx, HelmCacheSharedStorageClassName) - if err != nil { - return cacheprobe.StrategyFallback, err - } - if strategy != cacheprobe.StrategyFallback { - return strategy, nil - } - // GetStrategy returns Fallback both when results are missing/expired and - // when a fresh probe found the class unsupported. Honour the TTL of a fresh - // negative result: if the class was recently probed and found unusable, do - // not re-probe (which would create a PVC+pod) on every reconcile. - fresh, err := store.HasFreshResult(ctx, HelmCacheSharedStorageClassName) - if err != nil { - return cacheprobe.StrategyFallback, err - } - if fresh { - return cacheprobe.StrategyFallback, nil - } - // No valid cached result: probe ROX then RWX and persist. - prober := cacheprobe.NewProber(r.Client, ModelCacheInitNamespace, HelmCacheSharedStorageClassName, sharedFSProbeTTLSeconds) - strategy, results := prober.DetermineStrategy(ctx) - if err := store.Save(ctx, results); err != nil { - log.Error(err, "Failed to persist shared-FS probe results") - } - return strategy, nil -} - // doModelCacheSharedFS populates and exposes the model cache on a shared // filesystem storage class (nvcf-miniservice-sc), used when NVMesh is not present. The // cache is populated once (single-writer via the init lease/job, writing to the @@ -336,52 +300,41 @@ func (r *Reconciler) doModelCacheSharedFS(ctx context.Context, } } - // Reader RO PVC in the workload namespace on the shared class. With a - // shared-capable class all consumers see the same published cache data. + // Reader in the workload namespace, derived from the volume the writer + // populated. A claim that names only the shared StorageClass is not enough: + // a dynamic provisioner gives each claim its own volume, so the reader + // would mount an empty directory and the workload would find no model. roPVCName := "ro-pvc-" + cacheHandle roPVC := &corev1.PersistentVolumeClaim{} if err := r.Client.Get(ctx, client.ObjectKey{Name: roPVCName, Namespace: stCopy.Namespace}, roPVC); err != nil { if !apierrors.IsNotFound(err) { return reconcile.Result{}, err } - // The reader PVC does not exist yet: probe the shared class once to - // choose its access mode. The probe is intentionally scoped to reader - // creation; once the reader exists we never re-probe, so a transient - // probe failure after the TTL expires can never mark an already-healthy - // cache as failed. - strategy, err := r.resolveSharedFSStrategy(ctx) + writerPV, err := r.sharedFSWriterPV(ctx, writerPVC) if err != nil { return reconcile.Result{}, err } - if strategy == cacheprobe.StrategyFallback { - return reconcile.Result{}, r.terminalErrorWithMetric("shared_fs_unusable", - fmt.Sprintf("%s supports neither ROX nor RWX; shared caching disabled", HelmCacheSharedStorageClassName)) + if writerPV == nil { + log.V(1).Info("Shared-FS writer volume not bound yet, waiting", "pvc", rwPVC.Name) + return reconcile.Result{}, nil } - accessMode := corev1.ReadWriteMany - if strategy == cacheprobe.StrategyROX { - accessMode = corev1.ReadOnlyMany + roPV, err := r.newSharedFSReaderPV(ctx, stCopy, icmsReq, writerPV, roPVCName) + if err != nil { + return reconcile.Result{}, err } - roPVC = &corev1.PersistentVolumeClaim{ - ObjectMeta: metav1.ObjectMeta{ - Name: roPVCName, - Namespace: stCopy.Namespace, - Labels: types.GetLabelsForRequest(icmsReq, r.fff), - Annotations: types.GetAnnotationsForRequest(icmsReq), - }, - Spec: corev1.PersistentVolumeClaimSpec{ - AccessModes: []corev1.PersistentVolumeAccessMode{accessMode}, - StorageClassName: &sharedSC, - Resources: rwPVC.Spec.Resources, - }, + if err := r.Client.Create(ctx, roPV); err != nil && !apierrors.IsAlreadyExists(err) { + return reconcile.Result{}, err } + roPVC = newDerivedModelCacheReaderPVC(roPVCName, stCopy.Namespace, roPV, icmsReq, r.fff) maps.Copy(roPVC.Labels, getClusterWideResourceLabels(stCopy)) if err := r.setControlledObjectMeta(ctx, stCopy, roPVC); err != nil { return reconcile.Result{}, err } - if err := r.Client.Create(ctx, roPVC); err != nil { + if err := r.Client.Create(ctx, roPVC); err != nil && !apierrors.IsAlreadyExists(err) { return reconcile.Result{}, err } - log.Info("Shared-FS reader RO PVC created", "pvc", roPVCName, "accessMode", accessMode) + log.Info("Shared-FS reader created from the writer volume", + "pvc", roPVCName, "pv", roPV.Name, "volumeHandle", roPV.Spec.CSI.VolumeHandle) } switch r.getPVCState(roPVC) { @@ -971,10 +924,11 @@ func (r *Reconciler) doModelCacheNVMesh(ctx context.Context, //nolint:gocyclo Name: roPVCName, Namespace: stCopy.Namespace, } - secondaryPV.Spec.CSI.VolumeHandle, err = updateSecondaryPVVolumeHandle(secondaryPV.Spec.CSI.VolumeHandle, st.Namespace) + secondaryPV.Spec.CSI.VolumeHandle, err = deriveReaderVolumeHandle( + secondaryPV.Spec.CSI.Driver, secondaryPV.Spec.CSI.VolumeHandle, st.Namespace) if err != nil { return reconcile.Result{}, r.terminalErrorWithMetricErr("pvc_setup_failed", - fmt.Errorf("update secondary PV volume handle: %w", err)) + fmt.Errorf("derive reader PV volume handle: %w", err)) } secondaryPV.Status = corev1.PersistentVolumeStatus{} @@ -1796,6 +1750,139 @@ func (r *Reconciler) isVolumeDetached(ctx context.Context, volumeName string) (b // Cross-namespace NVMesh volumes (in NVMesh 3.2) are a required feature for this controller's // model caching implementation. This feature requires that the volumeHandle's 4-element encoding // contains the target PVC's namespace as the last element. +// deriveReaderVolumeHandle returns the CSI volume handle a read-only reader PV +// in namespace should use to address the volume the writer populated. +// +// NVMesh encodes the consuming namespace in the handle, so the reader needs +// its own namespace substituted in. Every other driver addresses one volume by +// one handle regardless of who mounts it, so the writer's handle is reused +// unchanged. Both were measured on real clusters: Weka handles look like +// weka/v2/csivol- and OCI FSS handles like +// ::, neither carrying a +// namespace. +func deriveReaderVolumeHandle(provisioner, volumeHandle, namespace string) (string, error) { + if provisioner != NVMeshStorageClassProvisioner { + return volumeHandle, nil + } + return updateSecondaryPVVolumeHandle(volumeHandle, namespace) +} + +// sharedFSWriterPV returns the PV the writer claim is bound to, or nil when the +// claim is not bound yet. That volume is the cache: readers are derived from it +// rather than provisioned separately. +func (r *Reconciler) sharedFSWriterPV( + ctx context.Context, writerPVC *corev1.PersistentVolumeClaim, +) (*corev1.PersistentVolume, error) { + if writerPVC == nil || writerPVC.Spec.VolumeName == "" { + return nil, nil + } + pv := &corev1.PersistentVolume{} + if err := r.Client.Get(ctx, client.ObjectKey{Name: writerPVC.Spec.VolumeName}, pv); err != nil { + if apierrors.IsNotFound(err) { + return nil, nil + } + return nil, err + } + if pv.Spec.CSI == nil || pv.Spec.CSI.VolumeHandle == "" { + return nil, r.terminalErrorWithMetric(modelcachetypes.ReasonCacheSpecInvalid, + fmt.Sprintf("shared-FS writer PV %s has no CSI volume handle", pv.Name)) + } + return pv, nil +} + +// newSharedFSReaderPV builds the read-only PV that exposes the writer's volume +// in the request namespace. It is the shared-filesystem counterpart of the +// NVMesh secondary PV and differs only in the volume handle, which +// deriveReaderVolumeHandle leaves unchanged for drivers that do not encode the +// namespace in it. +func (r *Reconciler) newSharedFSReaderPV( + ctx context.Context, + stCopy *nvcav1new.StorageRequest, + icmsReq *nvcav2beta1.ICMSRequest, + writerPV *corev1.PersistentVolume, + roPVCName string, +) (*corev1.PersistentVolume, error) { + roPV := writerPV.DeepCopy() + roPV.ObjectMeta = metav1.ObjectMeta{ + Name: "secondary-pv-" + stCopy.Spec.ICMSRequestName, + Labels: types.GetLabelsForRequest(icmsReq, r.fff), + Annotations: types.GetAnnotationsForRequest(icmsReq), + } + maps.Copy(roPV.Labels, getClusterWideResourceLabels(stCopy)) + roPV.Spec.AccessModes = []corev1.PersistentVolumeAccessMode{corev1.ReadOnlyMany} + // The reader PV is created statically and pre-bound by claimRef, so no + // provisioner is involved and it must carry no class. The claim asks for + // no class too; Kubernetes refuses to bind a pre-bound pair whose classes + // disagree, so inheriting the writer's class here would leave every reader + // claim Pending. + roPV.Spec.StorageClassName = "" + // Retain so removing one namespace's reader never destroys the cache the + // writer populated and other namespaces are still reading. + roPV.Spec.PersistentVolumeReclaimPolicy = corev1.PersistentVolumeReclaimRetain + roPV.Spec.ClaimRef = &corev1.ObjectReference{ + APIVersion: "v1", + Kind: "PersistentVolumeClaim", + Name: roPVCName, + Namespace: stCopy.Namespace, + } + handle, err := deriveReaderVolumeHandle( + roPV.Spec.CSI.Driver, roPV.Spec.CSI.VolumeHandle, stCopy.Namespace) + if err != nil { + return nil, r.terminalErrorWithMetricErr("pvc_setup_failed", + fmt.Errorf("derive reader PV volume handle: %w", err)) + } + roPV.Spec.CSI.VolumeHandle = handle + // Access modes are only used for binding; the kubelet does not enforce + // them. Without this the reader inherits the writer's read-write CSI + // source, and the only thing left standing between a consumer and the + // shared cache is mount options, which are empty for a provisioner that + // declares none. A reader could then mount the cache read-write and + // corrupt it for every other namespace reading the same volume. + roPV.Spec.CSI.ReadOnly = true + // A derived reader is a read-only model cache PV like any other, so it + // takes the provisioner's required reader options rather than inheriting + // whatever the writer was provisioned with. This matters for NVMesh, whose + // reader attaches the same XFS filesystem as the writer and needs nouuid + // and norecovery or the mount fails outright. deriveReaderVolumeHandle + // above already handles the NVMesh driver reaching this path. + roPV.Spec.MountOptions = r.resolveCacheMountOptions(ctx, roPV) + roPV.Status = corev1.PersistentVolumeStatus{} + if err := r.setControlledObjectMeta(ctx, stCopy, roPV); err != nil { + return nil, err + } + return roPV, nil +} + +// newDerivedModelCacheReaderPVC builds the claim bound by name to a derived +// reader PV. StorageClassName is empty so no provisioner is involved: the +// claim can only bind to the PV that names it. +func newDerivedModelCacheReaderPVC( + pvcName, namespace string, + roPV *corev1.PersistentVolume, + icmsReq *nvcav2beta1.ICMSRequest, + fff featureflag.Fetcher, +) *corev1.PersistentVolumeClaim { + emptySC := "" + return &corev1.PersistentVolumeClaim{ + ObjectMeta: metav1.ObjectMeta{ + Name: pvcName, + Namespace: namespace, + Labels: types.GetLabelsForRequest(icmsReq, fff), + Annotations: types.GetAnnotationsForRequest(icmsReq), + }, + Spec: corev1.PersistentVolumeClaimSpec{ + AccessModes: []corev1.PersistentVolumeAccessMode{corev1.ReadOnlyMany}, + StorageClassName: &emptySC, + VolumeName: roPV.Name, + Resources: corev1.VolumeResourceRequirements{ + Requests: corev1.ResourceList{ + corev1.ResourceStorage: roPV.Spec.Capacity[corev1.ResourceStorage], + }, + }, + }, + } +} + func updateSecondaryPVVolumeHandle(volumeHandle, namespace string) (string, error) { lastColonIdx := strings.LastIndex(volumeHandle, ":") if lastColonIdx == -1 { diff --git a/src/compute-plane-services/nvca/pkg/storage/modelcache_test.go b/src/compute-plane-services/nvca/pkg/storage/modelcache_test.go index f508a00bd..8e2fb28e6 100644 --- a/src/compute-plane-services/nvca/pkg/storage/modelcache_test.go +++ b/src/compute-plane-services/nvca/pkg/storage/modelcache_test.go @@ -42,6 +42,7 @@ import ( "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/util/sets" + "k8s.io/client-go/util/retry" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/client/fake" @@ -55,7 +56,6 @@ import ( nvcav1new "github.com/NVIDIA/nvcf/src/compute-plane-services/nvca/pkg/apis/nvca/v1" nvcav2beta1 "github.com/NVIDIA/nvcf/src/compute-plane-services/nvca/pkg/apis/nvca/v2beta1" featureflagmock "github.com/NVIDIA/nvcf/src/compute-plane-services/nvca/pkg/featureflag/mock" - "github.com/NVIDIA/nvcf/src/compute-plane-services/nvca/pkg/storage/cacheprobe" "github.com/NVIDIA/nvcf/src/compute-plane-services/nvca/pkg/types" ) @@ -217,7 +217,10 @@ func TestReconcile_ModelCache(t *testing.T) { } volumeHandlePrefix := "single-zone-cluster:csi-5326ce57-8cae-456c:ef7bc990-47e7-11f0-91b6-c952fffeea08:" primaryPV.Spec.CSI = &corev1.CSIPersistentVolumeSource{ - Driver: "nvmesh", + // The real driver name: NVCA keys the reader handle rewrite on it, and + // the rest of the model cache code compares it against the selected + // StorageClass provisioner. + Driver: NVMeshStorageClassProvisioner, VolumeHandle: volumeHandlePrefix + ModelCacheInitNamespace, } primaryPV.Spec.AccessModes = []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOnce} @@ -1332,19 +1335,11 @@ func TestReconcile_ModelCacheSharedFS(t *testing.T) { require.NoError(t, c.Create(ctx, srNamespace)) require.NoError(t, c.Create(ctx, NewModelCacheInitNamespace())) - // The shared class exists (operator- or Samba-provided) and its reader - // access mode is pre-resolved to ROX so resolveSharedFSStrategy does not - // run a live probe. + // The shared class exists (operator- or Samba-provided). require.NoError(t, c.Create(ctx, &storagev1.StorageClass{ ObjectMeta: metav1.ObjectMeta{Name: HelmCacheSharedStorageClassName}, Provisioner: SMBCSIDriverName, })) - store := cacheprobe.NewStateStore(c, ModelCacheInitNamespace) - require.NoError(t, store.Save(ctx, map[string]cacheprobe.Result{ - cacheprobe.ResultKey(HelmCacheSharedStorageClassName, cacheprobe.StrategyROX): { - State: cacheprobe.StateSupported, - }, - })) cacheHandle := "sharedfshandle" workloadNS := &corev1.Namespace{} @@ -1403,32 +1398,77 @@ func TestReconcile_ModelCacheSharedFS(t *testing.T) { } }, 5*time.Second, 50*time.Millisecond) - // Bind the writer RW PVC and complete the job: shared-FS keeps the writer - // claim (no primary PV finalize) and moves to Creating. - require.NoError(t, c.Get(ctx, client.ObjectKeyFromObject(rwPVC), rwPVC)) - rwPVC.Status.Phase = corev1.ClaimBound - require.NoError(t, c.Status().Update(ctx, rwPVC)) + // Bind the writer RW PVC to a volume and complete the job. The writer + // volume is the cache, so the reader must be derived from it. + writerPV := &corev1.PersistentVolume{ + ObjectMeta: metav1.ObjectMeta{Name: "writer-pv-" + cacheHandle}, + Spec: corev1.PersistentVolumeSpec{ + Capacity: corev1.ResourceList{corev1.ResourceStorage: resource.MustParse("1Gi")}, + AccessModes: []corev1.PersistentVolumeAccessMode{corev1.ReadWriteMany}, + PersistentVolumeReclaimPolicy: corev1.PersistentVolumeReclaimRetain, + StorageClassName: HelmCacheSharedStorageClassName, + PersistentVolumeSource: corev1.PersistentVolumeSource{ + CSI: &corev1.CSIPersistentVolumeSource{ + Driver: SMBCSIDriverName, + VolumeHandle: "shared-writer-volume", + }, + }, + }, + } + require.NoError(t, c.Create(ctx, writerPV)) + // The controller also writes this claim, so re-read before each update + // rather than racing it with a stale resourceVersion. + require.NoError(t, retry.RetryOnConflict(retry.DefaultRetry, func() error { + if err := c.Get(ctx, client.ObjectKeyFromObject(rwPVC), rwPVC); err != nil { + return err + } + rwPVC.Spec.VolumeName = writerPV.Name + return c.Update(ctx, rwPVC) + })) + require.NoError(t, retry.RetryOnConflict(retry.DefaultRetry, func() error { + if err := c.Get(ctx, client.ObjectKeyFromObject(rwPVC), rwPVC); err != nil { + return err + } + rwPVC.Status.Phase = corev1.ClaimBound + return c.Status().Update(ctx, rwPVC) + })) require.NoError(t, c.Get(ctx, client.ObjectKeyFromObject(initJob), initJob)) completeJob(ctx, t, c, initJob) - // A read-only reader PVC is created in the workload namespace on the shared - // class with the probed ROX access mode. + // A read-only reader PV is derived from the writer volume, and the reader + // claim binds to it by name. A claim naming only the shared class would let + // a dynamic provisioner hand back a new empty volume, which is what this + // path used to do. + roPV := &corev1.PersistentVolume{} + assert.EventuallyWithT(t, func(ct *assert.CollectT) { + err := c.Get(ctx, client.ObjectKey{Name: "secondary-pv-" + sr.Name}, roPV) + assert.NoError(ct, err) + }, 5*time.Second, 50*time.Millisecond) + require.NotNil(t, roPV.Spec.CSI) + assert.Equal(t, "shared-writer-volume", roPV.Spec.CSI.VolumeHandle, + "the reader must address the volume the writer populated") + assert.Equal(t, []corev1.PersistentVolumeAccessMode{corev1.ReadOnlyMany}, roPV.Spec.AccessModes) + assert.Equal(t, corev1.PersistentVolumeReclaimRetain, roPV.Spec.PersistentVolumeReclaimPolicy) + roPVC := &corev1.PersistentVolumeClaim{} assert.EventuallyWithT(t, func(ct *assert.CollectT) { err := c.Get(ctx, client.ObjectKey{Name: "ro-pvc-" + cacheHandle, Namespace: workloadNS.Name}, roPVC) assert.NoError(ct, err) }, 5*time.Second, 50*time.Millisecond) + assert.Equal(t, roPV.Name, roPVC.Spec.VolumeName) if assert.NotNil(t, roPVC.Spec.StorageClassName) { - assert.Equal(t, HelmCacheSharedStorageClassName, *roPVC.Spec.StorageClassName) + assert.Empty(t, *roPVC.Spec.StorageClassName, + "a StorageClass here would provision a new empty volume") + // Kubernetes refuses to bind a pre-bound pair whose classes disagree, + // and nothing in this suite runs the binding controller, so the pair + // has to be checked against each other here rather than one side at a + // time. The reader PV deep-copies the writer, which carries a class. + assert.Equal(t, *roPVC.Spec.StorageClassName, roPV.Spec.StorageClassName, + "the pre-bound reader PV and claim must agree on storage class or the claim never binds") } assert.Equal(t, []corev1.PersistentVolumeAccessMode{corev1.ReadOnlyMany}, roPVC.Spec.AccessModes) - // Shared-FS does not create cross-namespace primary/secondary PVs. - secondaryPV := &corev1.PersistentVolume{} - err = c.Get(ctx, client.ObjectKey{Name: "secondary-pv-" + st.Name}, secondaryPV) - assert.True(t, apierrors.IsNotFound(err), "shared-FS must not create a secondary PV") - // Bind the reader PVC: the request becomes Ready and exposes the RO PVC. roPVC.Status.Phase = corev1.ClaimBound require.NoError(t, c.Status().Update(ctx, roPVC)) @@ -1927,3 +1967,156 @@ func TestReclaimIdleSharedFSModelCaches(t *testing.T) { require.NoError(t, c.Get(ctx, client.ObjectKeyFromObject(sambaPVC), &corev1.PersistentVolumeClaim{}), "samba backing PVCs are reclaimed by the samba pass, not here") } + +// TestDeriveReaderVolumeHandle pins the one vendor specific step in reader +// derivation. NVMesh encodes the consuming namespace in the CSI volume handle +// so the reader needs its own substituted in. Every other driver addresses one +// volume by one handle, so reusing the writer's handle is what gives the +// reader the writer's data. Both were measured on real clusters. +func TestDeriveReaderVolumeHandle(t *testing.T) { + tests := []struct { + name string + provisioner string + handle string + want string + wantErr bool + }{ + { + name: "NVMesh substitutes the reader namespace", + provisioner: NVMeshStorageClassProvisioner, + handle: "nvmesh/csivol-abc:nvcf-modelcache-init", + want: "nvmesh/csivol-abc:tenant-ns", + }, + { + name: "Weka reuses the handle unchanged", + provisioner: "csi.weka.io", + handle: "weka/v2/csivol-pvc-8e38c07d-I6LIT56NBYME", + want: "weka/v2/csivol-pvc-8e38c07d-I6LIT56NBYME", + }, + { + // The FSS handle contains colons, so a rewrite would corrupt the + // export path rather than address another namespace. + name: "OCI FSS reuses the handle unchanged", + provisioner: "fss.csi.oraclecloud.com", + handle: "ocid1.filesystem.oc1.ap_kulai_2.aaaa:100.64.0.56:/csi-fss-eaf964b0", + want: "ocid1.filesystem.oc1.ap_kulai_2.aaaa:100.64.0.56:/csi-fss-eaf964b0", + }, + { + name: "an unknown driver reuses the handle unchanged", + provisioner: "csi.example.test", + handle: "opaque-handle", + want: "opaque-handle", + }, + { + name: "a malformed NVMesh handle is an error", + provisioner: NVMeshStorageClassProvisioner, + handle: "no-colons-here", + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := deriveReaderVolumeHandle(tt.provisioner, tt.handle, "tenant-ns") + if tt.wantErr { + require.Error(t, err) + return + } + require.NoError(t, err) + assert.Equal(t, tt.want, got) + }) + } +} + +// TestNewSharedFSReaderPVIsReadOnlyWithoutMountOptions covers the shared +// filesystem case that motivates the read-only CSI source. A provisioner with +// no declared reader options gets an empty mount option list, so mount options +// cannot be what keeps the reader read-only, and access modes are only used +// for binding. Without a read-only CSI source nothing stops a consumer +// mounting the shared cache read-write. +func TestNewSharedFSReaderPVIsReadOnlyWithoutMountOptions(t *testing.T) { + const wekaProvisioner = "csi.weka.io" + c := fake.NewClientBuilder(). + WithScheme(mgrScheme). + WithRESTMapper(newTestRESTMapper(mgrScheme)). + WithObjects(newMountOptionDefaultsObjects(wekaProvisioner, nvmeshMountOptionDefaults)...). + Build() + r := newMountOptionsReconciler(t, c, nil) + + writerPV := &corev1.PersistentVolume{ + ObjectMeta: metav1.ObjectMeta{Name: "writer-pv"}, + Spec: corev1.PersistentVolumeSpec{ + Capacity: corev1.ResourceList{corev1.ResourceStorage: resource.MustParse("1Gi")}, + PersistentVolumeSource: corev1.PersistentVolumeSource{ + CSI: &corev1.CSIPersistentVolumeSource{ + Driver: wekaProvisioner, + VolumeHandle: "weka/v2/csivol-pvc-8e38c07d", + ReadOnly: false, + }, + }, + }, + } + st := &nvcav1new.StorageRequest{ + ObjectMeta: metav1.ObjectMeta{Name: "sr", Namespace: "reader-ns"}, + Spec: nvcav1new.StorageRequestSpec{ICMSRequestName: "req"}, + } + icmsReq := &nvcav2beta1.ICMSRequest{ObjectMeta: metav1.ObjectMeta{Name: "req", Namespace: "reader-ns"}} + + roPV, err := r.newSharedFSReaderPV(context.Background(), st, icmsReq, writerPV, "ro-pvc") + require.NoError(t, err) + assert.Empty(t, roPV.Spec.MountOptions, + "this provisioner declares no reader options, so mount options cannot enforce read-only") + assert.True(t, roPV.Spec.CSI.ReadOnly, + "with no mount options the CSI source is the only thing keeping the reader read-only") + assert.Equal(t, "weka/v2/csivol-pvc-8e38c07d", roPV.Spec.CSI.VolumeHandle, + "a non-NVMesh handle addresses one volume and is reused unchanged") +} + +// TestNewSharedFSReaderPVResolvesMountOptions pins that a derived reader takes +// the provisioner's required reader options instead of inheriting the writer's. +// +// This matters once NVMesh reaches this path. NVMesh is detected today by the +// nvcf-sc-30 marker class; when that goes away it is identified by provisioner +// like every other backend and resolves to the shared filesystem flow. Its +// reader attaches the same XFS filesystem as the writer, so without nouuid and +// norecovery the mount fails outright, and inheriting the writer's read-write +// options is exactly the wrong answer. +func TestNewSharedFSReaderPVResolvesMountOptions(t *testing.T) { + c := fake.NewClientBuilder(). + WithScheme(mgrScheme). + WithRESTMapper(newTestRESTMapper(mgrScheme)). + WithObjects(newMountOptionDefaultsObjects(NVMeshStorageClassProvisioner, nvmeshMountOptionDefaults)...). + Build() + r := newMountOptionsReconciler(t, c, nil) + + writerPV := &corev1.PersistentVolume{ + ObjectMeta: metav1.ObjectMeta{Name: "writer-pv"}, + Spec: corev1.PersistentVolumeSpec{ + // What the writer was provisioned with, which the reader must not keep. + MountOptions: []string{"rw"}, + Capacity: corev1.ResourceList{corev1.ResourceStorage: resource.MustParse("1Gi")}, + PersistentVolumeSource: corev1.PersistentVolumeSource{ + CSI: &corev1.CSIPersistentVolumeSource{ + Driver: NVMeshStorageClassProvisioner, + VolumeHandle: "single-zone-cluster:csi-5326ce57-8cae-456c:ef7bc990-47e7-11f0-91b6-c952fffeea08:writer-ns", + }, + }, + }, + } + st := &nvcav1new.StorageRequest{ + ObjectMeta: metav1.ObjectMeta{Name: "sr", Namespace: "reader-ns"}, + Spec: nvcav1new.StorageRequestSpec{ICMSRequestName: "req"}, + } + icmsReq := &nvcav2beta1.ICMSRequest{ObjectMeta: metav1.ObjectMeta{Name: "req", Namespace: "reader-ns"}} + + roPV, err := r.newSharedFSReaderPV(context.Background(), st, icmsReq, writerPV, "ro-pvc") + require.NoError(t, err) + assert.Equal(t, []string{"ro", "norecovery", "nouuid"}, roPV.Spec.MountOptions, + "a derived reader must take the provisioner's reader options, not the writer's") + assert.NotContains(t, roPV.Spec.MountOptions, "rw", + "the writer's read-write option must not survive onto a read-only reader") + assert.True(t, strings.HasSuffix(roPV.Spec.CSI.VolumeHandle, ":reader-ns"), + "the NVMesh handle must be rewritten for the reader namespace") + assert.True(t, roPV.Spec.CSI.ReadOnly, + "the CSI source must be read-only: access modes are not enforced by the kubelet") +}