diff --git a/src/compute-plane-services/nvca/pkg/nvca/BUILD.bazel b/src/compute-plane-services/nvca/pkg/nvca/BUILD.bazel index f7364dd10..6654cd21e 100644 --- a/src/compute-plane-services/nvca/pkg/nvca/BUILD.bazel +++ b/src/compute-plane-services/nvca/pkg/nvca/BUILD.bazel @@ -309,6 +309,7 @@ go_test( "//src/compute-plane-services/nvca/vendor/k8s.io/client-go/dynamic", "//src/compute-plane-services/nvca/vendor/k8s.io/client-go/dynamic/fake", "//src/compute-plane-services/nvca/vendor/k8s.io/client-go/informers", + "//src/compute-plane-services/nvca/vendor/k8s.io/client-go/kubernetes", "//src/compute-plane-services/nvca/vendor/k8s.io/client-go/kubernetes/fake", "//src/compute-plane-services/nvca/vendor/k8s.io/client-go/listers/core/v1:core", "//src/compute-plane-services/nvca/vendor/k8s.io/client-go/rest", @@ -318,6 +319,7 @@ go_test( "//src/compute-plane-services/nvca/vendor/k8s.io/utils/ptr", "//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/envtest", "//src/compute-plane-services/nvca/vendor/sigs.k8s.io/yaml", ], ) diff --git a/src/compute-plane-services/nvca/pkg/nvca/backendk8scache.go b/src/compute-plane-services/nvca/pkg/nvca/backendk8scache.go index 09a6e78a9..c3107211c 100644 --- a/src/compute-plane-services/nvca/pkg/nvca/backendk8scache.go +++ b/src/compute-plane-services/nvca/pkg/nvca/backendk8scache.go @@ -734,6 +734,13 @@ func (b *BackendK8sCacheBuilder) Start(ctx context.Context) (*BackendK8sCache, < if err != nil && !k8serrors.IsAlreadyExists(err) { return nil, nil, fmt.Errorf("failed to create model cache init namespace: %w", err) } + // Patch WorkloadInstanceTypeLabel onto the namespace so the Kyverno + // add-unbound-dns policy injects nvcf-unbound nameservers into writer + // job pods. Done here (not only in Create) so pre-existing namespaces + // on upgraded clusters receive the label immediately at startup. + if err := ensureModelCacheNamespaceLabel(ctx, c.clients.K8s.CoreV1().Namespaces(), mcInitNamespace.Name); err != nil { + return nil, nil, fmt.Errorf("failed to patch model cache init namespace labels: %w", err) + } // Network policies must exist in all workload namespaces; // the Helm handler methods will do this for each new namespace. diff --git a/src/compute-plane-services/nvca/pkg/nvca/backendk8scache_gxcache.go b/src/compute-plane-services/nvca/pkg/nvca/backendk8scache_gxcache.go index 5570dcc5a..3d83f9685 100644 --- a/src/compute-plane-services/nvca/pkg/nvca/backendk8scache_gxcache.go +++ b/src/compute-plane-services/nvca/pkg/nvca/backendk8scache_gxcache.go @@ -46,3 +46,21 @@ func ensureGXCacheNamespaceLabels(ctx context.Context, nsPatcher k8sNamespacePat _, err := nsPatcher.Patch(ctx, namespace, k8sapitypes.JSONPatchType, patchData, metav1.PatchOptions{}) return err } + +// ensureModelCacheNamespaceLabel patches WorkloadInstanceTypeLabel onto the +// model-cache init namespace so the Kyverno add-unbound-dns policy injects +// the nvcf-unbound nameserver into writer job pods. Called at NVCA startup so +// the label is applied immediately on upgrade, before any model cache reconcile +// runs. +// +// A strategic merge patch is used instead of JSON patch "add" because JSON patch +// requires the parent path (/metadata/labels) to already exist; a namespace +// with a nil labels map would cause the patch to be rejected. The merge patch +// creates the labels map when absent and merges into it when present, so it is +// safe regardless of whether the namespace was freshly created or pre-existing. +func ensureModelCacheNamespaceLabel(ctx context.Context, nsPatcher k8sNamespacePatcher, namespace string) error { + patchData := []byte(fmt.Sprintf(`{"metadata":{"labels":{%q:%q}}}`, + nvcatypes.WorkloadInstanceTypeLabel, nvcatypes.WorkloadInstanceTypeValueMiniService)) + _, err := nsPatcher.Patch(ctx, namespace, k8sapitypes.StrategicMergePatchType, patchData, metav1.PatchOptions{}) + return err +} diff --git a/src/compute-plane-services/nvca/pkg/nvca/backendk8scache_test.go b/src/compute-plane-services/nvca/pkg/nvca/backendk8scache_test.go index efa33f3f1..c7681c4de 100644 --- a/src/compute-plane-services/nvca/pkg/nvca/backendk8scache_test.go +++ b/src/compute-plane-services/nvca/pkg/nvca/backendk8scache_test.go @@ -40,6 +40,7 @@ import ( "github.com/prometheus/client_golang/prometheus" "github.com/sirupsen/logrus" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" "go.opentelemetry.io/otel" "go.opentelemetry.io/otel/propagation" @@ -90,6 +91,8 @@ import ( queuesqs "github.com/NVIDIA/nvcf/src/compute-plane-services/nvca/pkg/queue/sqs" "github.com/NVIDIA/nvcf/src/compute-plane-services/nvca/pkg/types" nvcatypes "github.com/NVIDIA/nvcf/src/compute-plane-services/nvca/pkg/types" + "k8s.io/client-go/kubernetes" + "sigs.k8s.io/controller-runtime/pkg/envtest" ) // Helper function to safely update mock transport @@ -4426,9 +4429,9 @@ func TestGetGPUUsageStats_FallbackToNonSuffixSingleType(t *testing.T) { Status: corev1.NodeStatus{ Conditions: []corev1.NodeCondition{{Type: corev1.NodeReady, Status: corev1.ConditionTrue}}, Allocatable: corev1.ResourceList{ - corev1.ResourceCPU: resource.MustParse("5"), - corev1.ResourceMemory: resource.MustParse("32Gi"), - corev1.ResourceEphemeralStorage: resource.MustParse("256Gi"), + corev1.ResourceCPU: resource.MustParse("5"), + corev1.ResourceMemory: resource.MustParse("32Gi"), + corev1.ResourceEphemeralStorage: resource.MustParse("256Gi"), corev1.ResourceName(nodefeatures.GPUResourceKey): resource.MustParse("4"), }, }, @@ -5460,3 +5463,127 @@ func TestUpdateSchedulerWorkloadMetrics(t *testing.T) { assert.Equal(t, float64(1), vals[gaugeKey{"kai-scheduler", "function"}]) }) } + +func TestEnsureModelCacheNamespaceLabel_PatchesWithCorrectPayload(t *testing.T) { + namespace := "nvca-modelcache-init" + expectedPatch := []byte(fmt.Sprintf(`{"metadata":{"labels":{%q:%q}}}`, + nvcatypes.WorkloadInstanceTypeLabel, nvcatypes.WorkloadInstanceTypeValueMiniService)) + + nsPatcher := &mockNamespacePatcher{} + nsPatcher.On("Patch", mock.Anything, namespace, apitypes.StrategicMergePatchType, expectedPatch, metav1.PatchOptions{}). + Return(&corev1.Namespace{}, nil) + + err := ensureModelCacheNamespaceLabel(context.Background(), nsPatcher, namespace) + assert.NoError(t, err) + nsPatcher.AssertExpectations(t) +} + +func TestEnsureModelCacheNamespaceLabel_PatchError(t *testing.T) { + nsPatcher := &mockNamespacePatcher{} + nsPatcher.On("Patch", mock.Anything, mock.Anything, apitypes.StrategicMergePatchType, mock.Anything, metav1.PatchOptions{}). + Return(nil, fmt.Errorf("patch error")) + + err := ensureModelCacheNamespaceLabel(context.Background(), nsPatcher, "nvca-modelcache-init") + assert.Error(t, err) +} + +// TestEnsureModelCacheNamespaceLabel_IdempotentWhenLabelPresent confirms that +// ensureModelCacheNamespaceLabel always issues the patch even when the label is +// already set. A strategic merge patch is safe and idempotent regardless of +// whether the namespace was freshly created or already labelled. +func TestEnsureModelCacheNamespaceLabel_IdempotentWhenLabelPresent(t *testing.T) { + namespace := "nvca-modelcache-init" + expectedPatch := []byte(fmt.Sprintf(`{"metadata":{"labels":{%q:%q}}}`, + nvcatypes.WorkloadInstanceTypeLabel, nvcatypes.WorkloadInstanceTypeValueMiniService)) + + alreadyLabelled := &corev1.Namespace{} + alreadyLabelled.Labels = map[string]string{ + nvcatypes.WorkloadInstanceTypeLabel: nvcatypes.WorkloadInstanceTypeValueMiniService, + } + + nsPatcher := &mockNamespacePatcher{} + nsPatcher.On("Patch", mock.Anything, namespace, apitypes.StrategicMergePatchType, expectedPatch, metav1.PatchOptions{}). + Return(alreadyLabelled, nil) + + err := ensureModelCacheNamespaceLabel(context.Background(), nsPatcher, namespace) + assert.NoError(t, err) + nsPatcher.AssertNumberOfCalls(t, "Patch", 1) +} + +// TestEnsureModelCacheNamespaceLabel_Envtest exercises ensureModelCacheNamespaceLabel +// against a real Kubernetes API server to confirm the strategic merge patch succeeds +// in both the nil-labels case (JSON patch "add" would have failed here because +// /metadata/labels has no parent) and the pre-existing-labels case. +// +// Uses a plain envtest.Environment without NVCA CRDs — only core Kubernetes +// resources (Namespace) are needed, so loading the NVCA CRD directory is +// unnecessary and avoids the CRD path resolution issues in Bazel sandboxes. +func TestEnsureModelCacheNamespaceLabel_Envtest(t *testing.T) { + binAssetsDir := os.Getenv("KUBEBUILDER_ASSETS") + if binAssetsDir == "" { + t.Skip("KUBEBUILDER_ASSETS not set") + } + env := &envtest.Environment{ + BinaryAssetsDirectory: binAssetsDir, + } + cfg, err := env.Start() + require.NoError(t, err) + t.Cleanup(func() { _ = env.Stop() }) + + k8sClient, err := kubernetes.NewForConfig(cfg) + require.NoError(t, err) + + tests := []struct { + name string + initialLabels map[string]string + wantLabelValue string + }{ + { + name: "nil labels — strategic merge patch must not fail on missing /metadata/labels", + initialLabels: nil, + wantLabelValue: nvcatypes.WorkloadInstanceTypeValueMiniService, + }, + { + name: "pre-existing labels — target label added while other labels are preserved", + initialLabels: map[string]string{"existing-key": "existing-value"}, + wantLabelValue: nvcatypes.WorkloadInstanceTypeValueMiniService, + }, + { + name: "label already correct — idempotent, no error", + initialLabels: map[string]string{ + nvcatypes.WorkloadInstanceTypeLabel: nvcatypes.WorkloadInstanceTypeValueMiniService, + }, + wantLabelValue: nvcatypes.WorkloadInstanceTypeValueMiniService, + }, + } + + for i, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctx := context.Background() + nsName := fmt.Sprintf("test-modelcache-label-%d", i) + ns := &corev1.Namespace{} + ns.Name = nsName + ns.Labels = tt.initialLabels + _, err := k8sClient.CoreV1().Namespaces().Create(ctx, ns, metav1.CreateOptions{}) + require.NoError(t, err) + t.Cleanup(func() { + _ = k8sClient.CoreV1().Namespaces().Delete(ctx, nsName, metav1.DeleteOptions{}) + }) + + err = ensureModelCacheNamespaceLabel(ctx, k8sClient.CoreV1().Namespaces(), nsName) + require.NoError(t, err) + + got, err := k8sClient.CoreV1().Namespaces().Get(ctx, nsName, metav1.GetOptions{}) + require.NoError(t, err) + assert.Equal(t, tt.wantLabelValue, got.Labels[nvcatypes.WorkloadInstanceTypeLabel]) + + if tt.initialLabels != nil { + for k, v := range tt.initialLabels { + if k != nvcatypes.WorkloadInstanceTypeLabel { + assert.Equal(t, v, got.Labels[k], "pre-existing label %s must be preserved", k) + } + } + } + }) + } +}