feat(nvca): support per-function BYOO collector resource override

Adds a byooResources field to the nvcf-workload-config WorkloadConfig so a
Helm function can size its BYOO OTel collector sidecar above the cluster
default. The override is validated at decode time (positive quantities, >=1Gi
memory floor) and dropped with a warning if invalid so translate falls back to
the safe cluster default. At translate time it is overlaid per resource key
onto the cluster default, and any limit below its request is raised to keep the
pod spec valid.

Signed-off-by: shobham <shobham@nvidia.com>

Author: shobham-nv

src/compute-plane-services/nvca/internal/miniservice/translate_workload.go

@@ -51,7 +51,7 @@ func (r *Reconciler) translateWorkload(
 		InstanceTypeLabelSelectorKey: nodefeatures.UniformInstanceTypeLabelKey,
 		WorkloadResources:            corev1.ResourceRequirements{},
 		Tolerations:                  append([]corev1.Toleration(nil), r.cfg.Workload.Tolerations...),
-		OTelResources:                k8sutil.GetContainerResourcesBYOO(r.cfg),
+		OTelResources:                mergeBYOOResources(k8sutil.GetContainerResourcesBYOO(r.cfg), ms.Spec.WorkloadConfig.GetBYOOResources()),
 		FluentbitResources:           k8sutil.GetContainerResourcesFluentBit(r.cfg),
 		FluentbitEnabled:             r.FeatureFlagFetcher.IsFeatureFlagEnabled(featureflag.BYOOFluentBit),
 		ClusterRegion:                r.ClusterRegion,
@@ -127,6 +127,37 @@ func (r *Reconciler) translateTaskWorkload(
 	return metaToClientObjs(objs), nil
 }
 
+// mergeBYOOResources overlays a per-workload BYOO collector resource override on top of
+// the cluster-level default. Only the resource keys present in the override are replaced,
+// so a function can bump e.g. memory while keeping the default CPU. A nil override returns
+// the base unchanged. As a safety net, any limit that ends up below its request is raised
+// to the request so the resulting container spec stays valid.
+func mergeBYOOResources(base corev1.ResourceRequirements, override *corev1.ResourceRequirements) corev1.ResourceRequirements {
+	if override == nil {
+		return base
+	}
+	out := *base.DeepCopy()
+	apply := func(dst *corev1.ResourceList, src corev1.ResourceList) {
+		if len(src) == 0 {
+			return
+		}
+		if *dst == nil {
+			*dst = corev1.ResourceList{}
+		}
+		for name, q := range src {
+			(*dst)[name] = q.DeepCopy()
+		}
+	}
+	apply(&out.Requests, override.Requests)
+	apply(&out.Limits, override.Limits)
+	for name, req := range out.Requests {
+		if lim, ok := out.Limits[name]; ok && req.Cmp(lim) > 0 {
+			out.Limits[name] = req.DeepCopy()
+		}
+	}
+	return out
+}
+
 func metaToClientObjs(mobjs []metav1.Object) (cobjs []client.Object) {
 	cobjs = make([]client.Object, len(mobjs))
 	for i, obj := range mobjs {

src/compute-plane-services/nvca/internal/miniservice/translate_workload_test.go

@@ -27,6 +27,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	nvcav1alpha1 "github.com/NVIDIA/nvcf/src/compute-plane-services/nvca/pkg/apis/nvca/v1alpha1"
@@ -115,3 +116,44 @@ func TestTranslateWorkloadLLMUsesCanonicalPylonArgs(t *testing.T) {
 	assert.Equal(t, []string{"--backend-connectivity=reverse"}, backendConnectivityArgs)
 	assert.Equal(t, []string{"--initial-input-tps=100"}, initialInputTPSArgs)
 }
+
+func TestMergeBYOOResources(t *testing.T) {
+	base := corev1.ResourceRequirements{
+		Requests: corev1.ResourceList{
+			corev1.ResourceCPU:    resource.MustParse("1"),
+			corev1.ResourceMemory: resource.MustParse("2Gi"),
+		},
+		Limits: corev1.ResourceList{
+			corev1.ResourceCPU:    resource.MustParse("1"),
+			corev1.ResourceMemory: resource.MustParse("2Gi"),
+		},
+	}
+
+	t.Run("nil override returns base unchanged", func(t *testing.T) {
+		got := mergeBYOOResources(base, nil)
+		assert.Equal(t, base, got)
+	})
+
+	t.Run("memory-only override keeps default cpu", func(t *testing.T) {
+		override := &corev1.ResourceRequirements{
+			Requests: corev1.ResourceList{corev1.ResourceMemory: resource.MustParse("4Gi")},
+			Limits:   corev1.ResourceList{corev1.ResourceMemory: resource.MustParse("4Gi")},
+		}
+		got := mergeBYOOResources(base, override)
+		assert.Equal(t, "1", got.Requests.Cpu().String())
+		assert.Equal(t, "4Gi", got.Requests.Memory().String())
+		assert.Equal(t, "1", got.Limits.Cpu().String())
+		assert.Equal(t, "4Gi", got.Limits.Memory().String())
+		// base must not be mutated
+		assert.Equal(t, "2Gi", base.Limits.Memory().String())
+	})
+
+	t.Run("request above limit clamps limit up", func(t *testing.T) {
+		override := &corev1.ResourceRequirements{
+			Requests: corev1.ResourceList{corev1.ResourceMemory: resource.MustParse("4Gi")},
+		}
+		got := mergeBYOOResources(base, override)
+		assert.Equal(t, "4Gi", got.Requests.Memory().String())
+		assert.Equal(t, "4Gi", got.Limits.Memory().String())
+	})
+}

src/compute-plane-services/nvca/pkg/apis/nvca/v1alpha1/miniservice_types.go

@@ -19,6 +19,7 @@ package v1alpha1
 
 import (
 	"github.com/NVIDIA/nvcf/src/libraries/go/lib/pkg/icms-translate/translate/common"
+	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
@@ -58,6 +59,14 @@ type WorkloadConfig struct {
 	// FeatureFlags toggles per-workload behaviors, keyed by feature flag name.
 	// +optional
 	FeatureFlags map[string]bool `json:"featureFlags,omitempty"`
+
+	// BYOOResources overrides the BYOO OTel collector sidecar's CPU/memory for this
+	// workload. When set, it takes precedence over the cluster-level default at
+	// translate time (only the resource keys present are overridden). Intended for
+	// large-log / high-throughput functions whose collector needs more headroom than
+	// the cluster default.
+	// +optional
+	BYOOResources *corev1.ResourceRequirements `json:"byooResources,omitempty"`
 }
 
 // IsFeatureFlagEnabled reports whether the named feature flag is enabled.
@@ -68,6 +77,15 @@ func (c *WorkloadConfig) IsFeatureFlagEnabled(key string) bool {
 	return c.FeatureFlags[key]
 }
 
+// GetBYOOResources returns the per-workload BYOO collector resource override, or nil
+// if none is configured. It is nil-safe so callers can use it on a nil WorkloadConfig.
+func (c *WorkloadConfig) GetBYOOResources() *corev1.ResourceRequirements {
+	if c == nil {
+		return nil
+	}
+	return c.BYOOResources
+}
+
 type LocalObjectReference struct {
 	Name string `json:"name"`
 }

src/compute-plane-services/nvca/pkg/apis/nvca/v1alpha1/zz_generated.deepcopy.go

@@ -19,13 +19,19 @@ package featureflag
 
 import (
 	"context"
+	"fmt"
 
 	"github.com/NVIDIA/nvcf/src/compute-plane-services/nvca/pkg/apis/nvca/v1alpha1"
 	"github.com/go-logr/logr"
 	corev1 "k8s.io/api/core/v1"
 	"sigs.k8s.io/yaml"
 )
 
+// minBYOOMemoryBytes is the floor for a per-workload BYOO collector memory override.
+// Perf testing showed the collector OOMs between 512Mi and 1Gi under burst, so an
+// override below 1Gi is rejected (the safe cluster default is used instead).
+const minBYOOMemoryBytes = 1 << 30 // 1Gi
+
 const (
 	// WorkloadConfigConfigMapName is the fixed name of the ConfigMap a chart author may include
 	// to supply workload-specific configuration. The ConfigMap is never created on-cluster; it is
@@ -75,5 +81,31 @@ func DecodeWorkloadConfig(ctx context.Context, log logr.Logger, cm *corev1.Confi
 			delete(cfg.FeatureFlags, key)
 		}
 	}
+
+	// Drop an invalid BYOO resource override rather than failing the deploy: the
+	// translate path then falls back to the safe cluster-level default.
+	if cfg.BYOOResources != nil {
+		if err := validateBYOOResources(cfg.BYOOResources); err != nil {
+			log.Info("Ignoring invalid byooResources in ConfigMap %q: %v", WorkloadConfigConfigMapName, err)
+			cfg.BYOOResources = nil
+		}
+	}
 	return &cfg, nil
 }
+
+// validateBYOOResources rejects a per-workload BYOO collector resource override that
+// would be unsafe: every specified quantity must be positive, and any memory quantity
+// must be at least 1Gi (below that the collector OOMs under burst).
+func validateBYOOResources(rr *corev1.ResourceRequirements) error {
+	for kind, list := range map[string]corev1.ResourceList{"requests": rr.Requests, "limits": rr.Limits} {
+		for name, q := range list {
+			if q.Sign() <= 0 {
+				return fmt.Errorf("%s.%s must be positive (got %q)", kind, name, q.String())
+			}
+			if name == corev1.ResourceMemory && q.CmpInt64(minBYOOMemoryBytes) < 0 {
+				return fmt.Errorf("%s.memory must be at least 1Gi (got %q)", kind, q.String())
+			}
+		}
+	}
+	return nil
+}

src/compute-plane-services/nvca/pkg/featureflag/featureflag_workload.go

@@ -25,6 +25,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
 
 	"github.com/NVIDIA/nvcf/src/compute-plane-services/nvca/pkg/apis/nvca/v1alpha1"
 )
@@ -83,6 +84,36 @@ func TestDecodeWorkloadConfig(t *testing.T) {
 				FeatureFlags: map[string]bool{StatusByWorkerReadiness: true},
 			},
 		},
+		{
+			name: "valid byooResources override is kept",
+			cm: workloadConfigCM("byooResources:\n" +
+				"  requests:\n    cpu: \"1\"\n    memory: 4Gi\n" +
+				"  limits:\n    cpu: \"1\"\n    memory: 4Gi\n"),
+			want: &v1alpha1.WorkloadConfig{
+				BYOOResources: &corev1.ResourceRequirements{
+					Requests: corev1.ResourceList{
+						corev1.ResourceCPU:    resource.MustParse("1"),
+						corev1.ResourceMemory: resource.MustParse("4Gi"),
+					},
+					Limits: corev1.ResourceList{
+						corev1.ResourceCPU:    resource.MustParse("1"),
+						corev1.ResourceMemory: resource.MustParse("4Gi"),
+					},
+				},
+			},
+		},
+		{
+			name: "byooResources below 1Gi memory floor is dropped",
+			cm: workloadConfigCM("byooResources:\n" +
+				"  limits:\n    memory: 512Mi\n"),
+			want: &v1alpha1.WorkloadConfig{},
+		},
+		{
+			name: "byooResources with non-positive cpu is dropped",
+			cm: workloadConfigCM("byooResources:\n" +
+				"  requests:\n    cpu: \"0\"\n    memory: 2Gi\n"),
+			want: &v1alpha1.WorkloadConfig{},
+		},
 		{
 			name:    "invalid yaml returns error",
 			cm:      workloadConfigCM("featureFlags: [not-a-map"),
@@ -126,3 +157,62 @@ func TestWorkloadConfigIsFeatureFlagEnabled(t *testing.T) {
 	assert.True(t, cfg.IsFeatureFlagEnabled(StatusByWorkerReadiness))
 	assert.False(t, cfg.IsFeatureFlagEnabled("SomeOtherFlag"))
 }
+
+func TestWorkloadConfigGetBYOOResources(t *testing.T) {
+	var nilCfg *v1alpha1.WorkloadConfig
+	assert.Nil(t, nilCfg.GetBYOOResources())
+
+	empty := &v1alpha1.WorkloadConfig{}
+	assert.Nil(t, empty.GetBYOOResources())
+
+	rr := &corev1.ResourceRequirements{
+		Limits: corev1.ResourceList{corev1.ResourceMemory: resource.MustParse("4Gi")},
+	}
+	cfg := &v1alpha1.WorkloadConfig{BYOOResources: rr}
+	assert.Equal(t, rr, cfg.GetBYOOResources())
+}
+
+func TestValidateBYOOResources(t *testing.T) {
+	tests := []struct {
+		name    string
+		rr      *corev1.ResourceRequirements
+		wantErr bool
+	}{
+		{
+			name: "valid requests and limits",
+			rr: &corev1.ResourceRequirements{
+				Requests: corev1.ResourceList{corev1.ResourceCPU: resource.MustParse("1"), corev1.ResourceMemory: resource.MustParse("4Gi")},
+				Limits:   corev1.ResourceList{corev1.ResourceCPU: resource.MustParse("1"), corev1.ResourceMemory: resource.MustParse("4Gi")},
+			},
+		},
+		{
+			name: "memory exactly at 1Gi floor is valid",
+			rr:   &corev1.ResourceRequirements{Limits: corev1.ResourceList{corev1.ResourceMemory: resource.MustParse("1Gi")}},
+		},
+		{
+			name:    "memory below 1Gi floor is rejected",
+			rr:      &corev1.ResourceRequirements{Limits: corev1.ResourceList{corev1.ResourceMemory: resource.MustParse("512Mi")}},
+			wantErr: true,
+		},
+		{
+			name:    "zero cpu is rejected",
+			rr:      &corev1.ResourceRequirements{Requests: corev1.ResourceList{corev1.ResourceCPU: resource.MustParse("0")}},
+			wantErr: true,
+		},
+		{
+			name:    "negative memory is rejected",
+			rr:      &corev1.ResourceRequirements{Limits: corev1.ResourceList{corev1.ResourceMemory: resource.MustParse("-1Gi")}},
+			wantErr: true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := validateBYOOResources(tt.rr)
+			if tt.wantErr {
+				require.Error(t, err)
+				return
+			}
+			require.NoError(t, err)
+		})
+	}
+}