[release-2.5] fix: Add label selector to ConfigMap cache to prevent OOM via informer flooding (#2881)

- Add label selector to ConfigMap cache filter in newCacheOptions()
- Add label to buildPrometheusConfigMap() for cache visibility
- Handle upgrade path for pre-existing ConfigMaps without the label
  using merge patch to avoid resourceVersion rejection and preserve existing labels
- Propagate label on ConfigMap updates, not just creation
- Add filteredCacheClient test wrapper to simulate label-filtered informer cache
- Add test coverage for label assertion and upgrade scenario exercising
  the full Get->NotFound->Create->AlreadyExists->Patch code path

Signed-off-by: roburishabh <roburishabh@outlook.com>
Co-authored-by: roburishabh <roburishabh@outlook.com>

Author: google-oss-robot

cmd/operator/controller/start.go

@@ -415,7 +415,11 @@ func newCacheOptions() cache.Options {
 					common.LabelLaunchedBySparkOperator: "true",
 				}),
 			},
-			&corev1.ConfigMap{}:                  {},
+			&corev1.ConfigMap{}: {
+				Label: labels.SelectorFromSet(labels.Set{
+					common.LabelCreatedBySparkOperator: "true",
+				}),
+			},
 			&corev1.PersistentVolumeClaim{}:      {},
 			&corev1.Service{}:                    {},
 			&v1beta2.SparkApplication{}:          {},

internal/controller/sparkapplication/monitoring_config.go

@@ -33,7 +33,7 @@ import (
 	"github.com/kubeflow/spark-operator/v2/pkg/util"
 )
 
-func configPrometheusMonitoring(ctx context.Context, app *v1beta2.SparkApplication, client client.Client) error {
+func configPrometheusMonitoring(ctx context.Context, app *v1beta2.SparkApplication, c client.Client) error {
 	logger := log.FromContext(ctx)
 	port := common.DefaultPrometheusJavaAgentPort
 	if app.Spec.Monitoring.Prometheus != nil && app.Spec.Monitoring.Prometheus.Port != nil {
@@ -47,14 +47,41 @@ func configPrometheusMonitoring(ctx context.Context, app *v1beta2.SparkApplicati
 		key := types.NamespacedName{Namespace: configMap.Namespace, Name: configMap.Name}
 		if retryErr := retry.RetryOnConflict(retry.DefaultRetry, func() error {
 			cm := &corev1.ConfigMap{}
-			if err := client.Get(ctx, key, cm); err != nil {
+			if err := c.Get(ctx, key, cm); err != nil {
 				if errors.IsNotFound(err) {
-					return client.Create(ctx, configMap)
+					if createErr := c.Create(ctx, configMap); createErr != nil {
+						// Handle upgrade scenario: ConfigMap exists in the cluster but is not
+						// visible in the filtered cache because it was created before the
+						// LabelCreatedBySparkOperator label selector was added to the informer.
+						// Use Patch (merge patch) instead of Update because we cannot Get the
+						// object from the filtered cache to obtain its resourceVersion.
+						if errors.IsAlreadyExists(createErr) {
+							base := &corev1.ConfigMap{
+								ObjectMeta: metav1.ObjectMeta{
+									Name:      key.Name,
+									Namespace: key.Namespace,
+								},
+							}
+							desired := base.DeepCopy()
+							desired.Labels = map[string]string{
+								common.LabelCreatedBySparkOperator: "true",
+							}
+							desired.Data = configMap.Data
+							desired.OwnerReferences = configMap.OwnerReferences
+							return c.Patch(ctx, desired, client.MergeFrom(base))
+						}
+						return createErr
+					}
+					return nil
 				}
 				return err
 			}
 			cm.Data = configMap.Data
-			return client.Update(ctx, cm)
+			if cm.Labels == nil {
+				cm.Labels = map[string]string{}
+			}
+			cm.Labels[common.LabelCreatedBySparkOperator] = "true"
+			return c.Update(ctx, cm)
 		}); retryErr != nil {
 			return retryErr
 		}
@@ -146,6 +173,9 @@ func buildPrometheusConfigMap(app *v1beta2.SparkApplication, prometheusConfigMap
 			Name:            prometheusConfigMapName,
 			Namespace:       app.Namespace,
 			OwnerReferences: []metav1.OwnerReference{util.GetOwnerReference(app)},
+			Labels: map[string]string{
+				common.LabelCreatedBySparkOperator: "true",
+			},
 		},
 		Data: configMapData,
 	}

internal/controller/sparkapplication/monitoring_config_test.go

@@ -23,7 +23,9 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/utils/ptr"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
@@ -33,6 +35,28 @@ import (
 	"github.com/kubeflow/spark-operator/v2/pkg/util"
 )
 
+// filteredCacheClient wraps a client.Client and overrides Get to simulate a
+// label-filtered informer cache. ConfigMaps that lack the required label are
+// invisible to Get (returns NotFound), just like a real ByObject label selector
+// would behave at runtime.
+type filteredCacheClient struct {
+	client.Client
+	requiredLabel string
+	requiredValue string
+}
+
+func (f *filteredCacheClient) Get(ctx context.Context, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error {
+	if err := f.Client.Get(ctx, key, obj, opts...); err != nil {
+		return err
+	}
+	if cm, ok := obj.(*corev1.ConfigMap); ok {
+		if cm.Labels[f.requiredLabel] != f.requiredValue {
+			return errors.NewNotFound(schema.GroupResource{Group: "", Resource: "configmaps"}, key.Name)
+		}
+	}
+	return nil
+}
+
 func TestConfigPrometheusMonitoring(t *testing.T) {
 	type testcase struct {
 		app                   *v1beta2.SparkApplication
@@ -57,6 +81,9 @@ func TestConfigPrometheusMonitoring(t *testing.T) {
 		err = fakeClient.Get(context.TODO(), client.ObjectKeyFromObject(configMap), configMap)
 		assert.NoError(t, err, "failed to get ConfigMap %s", configMapName)
 
+		assert.Equal(t, "true", configMap.Labels[common.LabelCreatedBySparkOperator],
+			"ConfigMap %s should have LabelCreatedBySparkOperator label", configMapName)
+
 		if test.app.Spec.Monitoring.Prometheus != nil && test.app.Spec.Monitoring.Prometheus.ConfigFile == nil &&
 			test.app.Spec.Monitoring.MetricsPropertiesFile == nil {
 			assert.Len(t, configMap.Data, 2, "expected 2 data items")
@@ -342,3 +369,67 @@ func TestConfigPrometheusMonitoring(t *testing.T) {
 		testFn(test, t)
 	}
 }
+
+func TestConfigPrometheusMonitoring_UpgradeExistingConfigMapWithoutLabel(t *testing.T) {
+	app := &v1beta2.SparkApplication{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "upgrade-app",
+			Namespace: "default",
+			UID:       "test-uid",
+		},
+		Spec: v1beta2.SparkApplicationSpec{
+			Monitoring: &v1beta2.MonitoringSpec{
+				ExposeDriverMetrics:   true,
+				ExposeExecutorMetrics: true,
+				Prometheus: &v1beta2.PrometheusSpec{
+					JmxExporterJar: "/prometheus/exporter.jar",
+				},
+			},
+		},
+	}
+
+	configMapName := util.GetPrometheusConfigMapName(app)
+
+	// Pre-create a ConfigMap WITHOUT the LabelCreatedBySparkOperator label,
+	// simulating a ConfigMap created by a previous version of the operator.
+	existingCM := &corev1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      configMapName,
+			Namespace: app.Namespace,
+			Labels: map[string]string{
+				"some-other-label": "should-be-preserved",
+			},
+		},
+		Data: map[string]string{
+			"old-key": "old-value",
+		},
+	}
+
+	underlying := fake.NewFakeClient(existingCM)
+
+	// Wrap the fake client to simulate a label-filtered informer cache:
+	// Get returns NotFound for ConfigMaps that lack LabelCreatedBySparkOperator,
+	// forcing the code through the Create -> AlreadyExists -> Patch path.
+	filtered := &filteredCacheClient{
+		Client:        underlying,
+		requiredLabel: common.LabelCreatedBySparkOperator,
+		requiredValue: "true",
+	}
+
+	err := configPrometheusMonitoring(context.TODO(), app, filtered)
+	assert.NoError(t, err, "configPrometheusMonitoring should handle pre-existing ConfigMap without label")
+
+	// Read back from the underlying (unfiltered) client to verify the patch was applied.
+	updatedCM := &corev1.ConfigMap{}
+	err = underlying.Get(context.TODO(), client.ObjectKeyFromObject(existingCM), updatedCM)
+	assert.NoError(t, err, "failed to get updated ConfigMap")
+
+	assert.Equal(t, "true", updatedCM.Labels[common.LabelCreatedBySparkOperator],
+		"Pre-existing ConfigMap should have LabelCreatedBySparkOperator added after upgrade")
+
+	assert.Equal(t, "should-be-preserved", updatedCM.Labels["some-other-label"],
+		"Merge patch should preserve pre-existing labels")
+
+	assert.Contains(t, updatedCM.Data, common.MetricsPropertiesKey,
+		"ConfigMap data should be updated with metrics properties")
+}