Guard multicluster Create against cross-cluster name collisions

cmd/manager/main.go

@@ -374,10 +374,12 @@ func main() {
 	committedResourceGVK := schema.GroupVersionKind{Group: "cortex.cloud", Version: "v1alpha1", Kind: "CommittedResource"}
 	flavorGroupCapacityGVK := schema.GroupVersionKind{Group: "cortex.cloud", Version: "v1alpha1", Kind: "FlavorGroupCapacity"}
 	projectQuotaGVK := schema.GroupVersionKind{Group: "cortex.cloud", Version: "v1alpha1", Kind: "ProjectQuota"}
+	multiclusterMonitor := multicluster.NewMonitor("cortex_")
 	multiclusterClient := &multicluster.Client{
 		HomeCluster:    homeCluster,
 		HomeRestConfig: restConfig,
 		HomeScheme:     scheme,
+		Monitor:        multiclusterMonitor,
 		ResourceRouters: map[schema.GroupVersionKind]multicluster.ResourceRouter{
 			hvGVK:                  multicluster.HypervisorResourceRouter{},
 			reservationGVK:         multicluster.ReservationsResourceRouter{},
@@ -398,6 +400,7 @@ func main() {
 	metricsConfig := conf.GetConfigOrDie[monitoring.Config]()
 	metrics.Registry = monitoring.WrapRegistry(metrics.Registry, metricsConfig)
 	metrics.Registry.MustRegister(&logMetricsMonitor)
+	metrics.Registry.MustRegister(multiclusterMonitor)
 
 	// TODO: Remove me after scheduling pipeline steps don't require DB connections anymore.
 	metrics.Registry.MustRegister(&db.Monitor)

cmd/shim/main.go

@@ -285,6 +285,7 @@ func main() {
 	// This is useful to distinguish metrics from different deployments.
 	metricsConfig := conf.GetConfigOrDie[monitoring.Config]()
 	metrics.Registry = monitoring.WrapRegistry(metrics.Registry, metricsConfig)
+	metrics.Registry.MustRegister(multiclusterClient.Monitor)
 
 	// API endpoint.
 	mux := http.NewServeMux()
@@ -360,6 +361,7 @@ func setupMulticlusterClient(ctx context.Context, mgr manager.Manager, restConfi
 		HomeRestConfig:  restConfig,
 		HomeScheme:      scheme,
 		ResourceRouters: multicluster.DefaultResourceRouters,
+		Monitor:         multicluster.NewMonitor("cortex_"),
 	}
 	mclConfig := conf.GetConfigOrDie[multicluster.ClientConfig]()
 	if err := mcl.InitFromConf(ctx, mgr, mclConfig); err != nil {

helm/bundles/cortex-nova/templates/alerts.yaml

@@ -669,4 +669,26 @@ spec:
           The committed resource quota API (Limes LIQUID integration) is returning
           HTTP 5xx errors. This indicates internal problems computing or applying
           quota. Limes may not be able to enforce committed resource quotas.
+
+    - alert: CortexNovaMulticlusterNameConflicts
+      expr: |
+        sum by (method, gvk) (increase(cortex_multicluster_cross_cluster_name_conflicts_total{service="cortex-nova-metrics"}[10m])) > 0
+      for: 5m
+      labels:
+        context: multicluster
+        dashboard: cortex-status-dashboard/cortex-status-dashboard
+        service: cortex
+        severity: warning
+        support_group: workload-management
+      annotations:
+        summary: "Cross-cluster name conflicts detected for `{{ "{{" }} $labels.gvk {{ "}}" }}`"
+        description: >
+          The multicluster client detected the same resource name for
+          `{{ "{{" }} $labels.gvk {{ "}}" }}` on more than one cluster during
+          `{{ "{{" }} $labels.method {{ "}}" }}` operations. This means reads fan
+          out to duplicates (surfaced as duplicate errors) and creates may be
+          rejected to avoid introducing new collisions. This usually indicates a
+          resource router is mapping the same object to multiple clusters, or an
+          object was created out-of-band on the wrong cluster. Investigate the
+          affected resources and the routing configuration.
 {{- end }}

helm/bundles/cortex-placement-shim/templates/alerts.yaml

@@ -188,4 +188,27 @@ spec:
           against a limit of 500m. Under normal operation the shim should use
           much less since it primarily proxies requests. This may indicate a
           hot loop, excessive logging, or an unusual traffic spike.
+
+    # Multicluster
+    - alert: CortexPlacementShimMulticlusterNameConflicts
+      expr: |
+        sum by (method, gvk) (increase(cortex_multicluster_cross_cluster_name_conflicts_total{service="cortex-placement-shim-metrics-service"}[10m])) > 0
+      for: 5m
+      labels:
+        context: multicluster
+        dashboard: cortex-placement-shim-status-dashboard/cortex-placement-shim-status-dashboard
+        service: cortex
+        severity: warning
+        support_group: workload-management
+      annotations:
+        summary: "Cross-cluster name conflicts detected for `{{ "{{" }} $labels.gvk {{ "}}" }}`"
+        description: >
+          The multicluster client detected the same resource name for
+          `{{ "{{" }} $labels.gvk {{ "}}" }}` on more than one cluster during
+          `{{ "{{" }} $labels.method {{ "}}" }}` operations. This means reads fan
+          out to duplicates (surfaced as duplicate errors) and creates may be
+          rejected to avoid introducing new collisions. This usually indicates a
+          resource router is mapping the same object to multiple clusters, or an
+          object was created out-of-band on the wrong cluster. Investigate the
+          affected resources and the routing configuration.
 {{- end }}

pkg/multicluster/client.go

@@ -40,6 +40,10 @@ type Client struct {
 	// This scheme should include all types used in the remote clusters.
 	HomeScheme *runtime.Scheme
 
+	// Optional monitor for Prometheus metrics. A nil Monitor causes recording
+	// to be skipped, so the client can be used without wiring metrics.
+	Monitor Monitor
+
 	// Remote clusters to use by resource type. Multiple clusters can serve
 	// the same GVK (e.g. one per availability zone).
 	remoteClusters map[schema.GroupVersionKind][]remoteCluster
@@ -352,6 +356,9 @@ func (c *Client) Get(ctx context.Context, key client.ObjectKey, obj client.Objec
 			err := cl.GetClient().Get(ctx, key, candidate, opts...)
 			if err == nil {
 				// In this case Get() was already called and the object set.
+				if c.Monitor != nil {
+					c.Monitor.recordCrossClusterNameConflict("get", gvk)
+				}
 				return &duplicateError{msg: fmt.Sprintf("duplicate %s %s/%s in multiple clusters",
 					gvk, key.Namespace, key.Name)}
 			}
@@ -436,6 +443,9 @@ func (c *Client) List(ctx context.Context, list client.ObjectList, opts ...clien
 		return err
 	}
 	if len(duplicates) > 0 {
+		if c.Monitor != nil {
+			c.Monitor.recordCrossClusterNameConflict("list", gvk)
+		}
 		return &duplicateError{msg: fmt.Sprintf("duplicate %s [%s] in multiple clusters",
 			gvk, strings.Join(duplicates, ", "))}
 	}
@@ -450,7 +460,15 @@ func (c *Client) Apply(ctx context.Context, obj runtime.ApplyConfiguration, opts
 
 // Create routes the object to the matching cluster using the ResourceRouter
 // and performs a Create operation.
+//
+// Before writing, it performs a best-effort Get against the other clusters
+// serving the same GVK to detect a cross-cluster name collision. If the object
+// name already exists on another cluster, a duplicateError is returned (checkable
+// with IsDuplicateError) and no create is performed. Non-NotFound errors from the
+// probe clusters are logged and ignored so that a single unavailable cluster does
+// not block writes.
 func (c *Client) Create(ctx context.Context, obj client.Object, opts ...client.CreateOption) error {
+	log := ctrl.LoggerFrom(ctx)
 	gvk, err := c.GVKFromHomeScheme(obj)
 	if err != nil {
 		return err
@@ -459,6 +477,35 @@ func (c *Client) Create(ctx context.Context, obj client.Object, opts ...client.C
 	if err != nil {
 		return err
 	}
+
+	// Best-effort cross-cluster name collision check: the same namespace/name
+	// must not already exist on another cluster serving this GVK, otherwise
+	// reads would fan out to a duplicate (see IsDuplicateError).
+	clusters, err := c.ClustersForGVK(gvk)
+	if err != nil {
+		return err
+	}
+	key := client.ObjectKeyFromObject(obj)
+	for _, other := range clusters {
+		if other == cl {
+			continue
+		}
+		candidate := obj.DeepCopyObject().(client.Object)
+		getErr := other.GetClient().Get(ctx, key, candidate)
+		if getErr == nil {
+			if c.Monitor != nil {
+				c.Monitor.recordCrossClusterNameConflict("create", gvk)
+			}
+			return &duplicateError{msg: fmt.Sprintf("cannot create %s %s/%s: already exists on another cluster",
+				gvk, key.Namespace, key.Name)}
+		}
+		if !apierrors.IsNotFound(getErr) {
+			log.Error(getErr, "error checking for cross-cluster name conflict before create",
+				"gvk", gvk, "namespace", key.Namespace, "name", key.Name,
+				"host", other.GetConfig().Host)
+		}
+	}
+
 	return cl.GetClient().Create(ctx, obj, opts...)
 }
 
@@ -641,6 +688,9 @@ func (c *subResourceClient) Get(ctx context.Context, obj, subResource client.Obj
 				Get(ctx, candidateObj, candidateSub, opts...)
 			if err == nil {
 				// In this case Get() was already called and the object set.
+				if c.multiclusterClient.Monitor != nil {
+					c.multiclusterClient.Monitor.recordCrossClusterNameConflict("subresource_get", gvk)
+				}
 				return &duplicateError{msg: fmt.Sprintf("duplicate %s %s/%s subresource %s in multiple clusters",
 					gvk, candidateObj.GetNamespace(), candidateObj.GetName(), c.subResource)}
 			}

pkg/multicluster/client_test.go

@@ -10,6 +10,7 @@ import (
 	"sync"
 	"testing"
 
+	"github.com/prometheus/client_golang/prometheus/testutil"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -1060,6 +1061,136 @@ func TestClient_Create_NoMatchReturnsError(t *testing.T) {
 	}
 }
 
+func TestClient_Create_CrossClusterNameConflict(t *testing.T) {
+	scheme := newTestScheme(t)
+	// The same name already exists on remote1.
+	existing := &corev1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{Name: "dup-cm", Namespace: "default"},
+	}
+	homeCluster := newFakeCluster(scheme)
+	remote1 := newFakeCluster(scheme, existing)
+	remote2 := newFakeCluster(scheme)
+
+	mon := NewMonitor("cortex_")
+	c := &Client{
+		HomeCluster: homeCluster,
+		HomeScheme:  scheme,
+		Monitor:     mon,
+		ResourceRouters: map[schema.GroupVersionKind]ResourceRouter{
+			configMapGVK: testRouter{},
+		},
+		remoteClusters: map[schema.GroupVersionKind][]remoteCluster{
+			configMapGVK: {
+				{cluster: remote1, labels: map[string]string{"az": "az-1"}},
+				{cluster: remote2, labels: map[string]string{"az": "az-2"}},
+			},
+		},
+	}
+
+	// Routes to remote2 (az-2), but the name already exists on remote1.
+	cm := &corev1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "dup-cm",
+			Namespace: "default",
+			Labels:    map[string]string{"az": "az-2"},
+		},
+	}
+	err := c.Create(context.Background(), cm)
+	if err == nil {
+		t.Fatal("expected error due to cross-cluster name conflict")
+	}
+	if !IsDuplicateError(err) {
+		t.Errorf("expected duplicate error, got %v", err)
+	}
+
+	// Must NOT have been created on the target cluster remote2.
+	result := &corev1.ConfigMap{}
+	if err := remote2.GetClient().Get(context.Background(), client.ObjectKey{Name: "dup-cm", Namespace: "default"}, result); err == nil {
+		t.Error("object should not have been created on remote2 after a conflict")
+	}
+
+	// The conflict counter should have been incremented for method "create".
+	cm2 := mon.(*monitor)
+	if got := testutil.ToFloat64(cm2.crossClusterNameConflicts.WithLabelValues("create", configMapGVK.String())); got != 1 {
+		t.Errorf("expected conflict counter = 1, got %v", got)
+	}
+}
+
+func TestClient_Create_NoConflictWhenNameFreeElsewhere(t *testing.T) {
+	scheme := newTestScheme(t)
+	homeCluster := newFakeCluster(scheme)
+	remote1 := newFakeCluster(scheme)
+	remote2 := newFakeCluster(scheme)
+
+	c := &Client{
+		HomeCluster: homeCluster,
+		HomeScheme:  scheme,
+		ResourceRouters: map[schema.GroupVersionKind]ResourceRouter{
+			configMapGVK: testRouter{},
+		},
+		remoteClusters: map[schema.GroupVersionKind][]remoteCluster{
+			configMapGVK: {
+				{cluster: remote1, labels: map[string]string{"az": "az-1"}},
+				{cluster: remote2, labels: map[string]string{"az": "az-2"}},
+			},
+		},
+	}
+
+	cm := &corev1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "unique-cm",
+			Namespace: "default",
+			Labels:    map[string]string{"az": "az-2"},
+		},
+	}
+	if err := c.Create(context.Background(), cm); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	// Should have been created on the routed cluster remote2.
+	result := &corev1.ConfigMap{}
+	if err := remote2.GetClient().Get(context.Background(), client.ObjectKey{Name: "unique-cm", Namespace: "default"}, result); err != nil {
+		t.Errorf("expected object on remote2: %v", err)
+	}
+}
+
+func TestClient_Create_NilMonitorSafe(t *testing.T) {
+	scheme := newTestScheme(t)
+	existing := &corev1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{Name: "dup-cm", Namespace: "default"},
+	}
+	homeCluster := newFakeCluster(scheme)
+	remote1 := newFakeCluster(scheme, existing)
+	remote2 := newFakeCluster(scheme)
+
+	// No Monitor set — recording the conflict must not panic.
+	c := &Client{
+		HomeCluster: homeCluster,
+		HomeScheme:  scheme,
+		ResourceRouters: map[schema.GroupVersionKind]ResourceRouter{
+			configMapGVK: testRouter{},
+		},
+		remoteClusters: map[schema.GroupVersionKind][]remoteCluster{
+			configMapGVK: {
+				{cluster: remote1, labels: map[string]string{"az": "az-1"}},
+				{cluster: remote2, labels: map[string]string{"az": "az-2"}},
+			},
+		},
+	}
+
+	cm := &corev1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "dup-cm",
+			Namespace: "default",
+			Labels:    map[string]string{"az": "az-2"},
+		},
+	}
+	err := c.Create(context.Background(), cm)
+	if !IsDuplicateError(err) {
+		t.Errorf("expected duplicate error, got %v", err)
+	}
+}
+
 func TestClient_Delete_SingleRemoteCluster(t *testing.T) {
 	scheme := newTestScheme(t)
 	existingCM := &corev1.ConfigMap{

pkg/multicluster/monitor.go

@@ -0,0 +1,62 @@
+// Copyright SAP SE
+// SPDX-License-Identifier: Apache-2.0
+
+package multicluster
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+)
+
+// duplicateConflictLabels labels the cross-cluster name conflict counter by the
+// method of access (e.g. "create", "get", "list") and the resource GVK.
+var duplicateConflictLabels = []string{"method", "gvk"}
+
+// Monitor is the metrics sink for the multicluster client. It is optional on
+// the Client: a nil Monitor causes recording to be skipped entirely. It embeds
+// prometheus.Collector so a concrete implementation can be registered with a
+// Prometheus registry.
+type Monitor interface {
+	prometheus.Collector
+
+	// recordCrossClusterNameConflict is called when the same namespace/name was
+	// detected on more than one cluster serving the GVK, labeled by the method
+	// of access and the resource GVK.
+	recordCrossClusterNameConflict(method string, gvk schema.GroupVersionKind)
+}
+
+// monitor is the default Prometheus-backed Monitor implementation.
+type monitor struct {
+	// crossClusterNameConflicts counts how often the same namespace/name was
+	// detected on more than one cluster serving the GVK, labeled by the method
+	// of access and the resource GVK.
+	crossClusterNameConflicts *prometheus.CounterVec
+}
+
+// NewMonitor creates a new Prometheus-backed multicluster client monitor. The
+// prefix is prepended to every metric name (e.g. pass "cortex_" to produce
+// "cortex_multicluster_cross_cluster_name_conflicts_total").
+func NewMonitor(prefix string) Monitor {
+	return &monitor{
+		crossClusterNameConflicts: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Name: prefix + "multicluster_cross_cluster_name_conflicts_total",
+			Help: "Total number of times the same resource name was detected on more than one cluster serving the same GVK",
+		}, duplicateConflictLabels),
+	}
+}
+
+// recordCrossClusterNameConflict increments the conflict counter for the given
+// access method and GVK.
+func (m *monitor) recordCrossClusterNameConflict(method string, gvk schema.GroupVersionKind) {
+	m.crossClusterNameConflicts.WithLabelValues(method, gvk.String()).Inc()
+}
+
+// Describe implements prometheus.Collector.
+func (m *monitor) Describe(ch chan<- *prometheus.Desc) {
+	m.crossClusterNameConflicts.Describe(ch)
+}
+
+// Collect implements prometheus.Collector.
+func (m *monitor) Collect(ch chan<- prometheus.Metric) {
+	m.crossClusterNameConflicts.Collect(ch)
+}