diff --git a/cmd/manager/main.go b/cmd/manager/main.go index a7ae683d3..fa18b7a9f 100644 --- a/cmd/manager/main.go +++ b/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) diff --git a/cmd/shim/main.go b/cmd/shim/main.go index 29865e0c5..e17eb2200 100644 --- a/cmd/shim/main.go +++ b/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 { diff --git a/helm/bundles/cortex-nova/templates/alerts.yaml b/helm/bundles/cortex-nova/templates/alerts.yaml index 6655c5528..e6e08e8e3 100644 --- a/helm/bundles/cortex-nova/templates/alerts.yaml +++ b/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 }} diff --git a/helm/bundles/cortex-placement-shim/templates/alerts.yaml b/helm/bundles/cortex-placement-shim/templates/alerts.yaml index c570ccd91..f6d3dbaa4 100644 --- a/helm/bundles/cortex-placement-shim/templates/alerts.yaml +++ b/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 }} diff --git a/pkg/multicluster/client.go b/pkg/multicluster/client.go index 979258a44..583d74574 100644 --- a/pkg/multicluster/client.go +++ b/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)} } diff --git a/pkg/multicluster/client_test.go b/pkg/multicluster/client_test.go index 6e965d440..1ee07bb7c 100644 --- a/pkg/multicluster/client_test.go +++ b/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{ diff --git a/pkg/multicluster/monitor.go b/pkg/multicluster/monitor.go new file mode 100644 index 000000000..a33cfffb8 --- /dev/null +++ b/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) +} diff --git a/pkg/multicluster/monitor_test.go b/pkg/multicluster/monitor_test.go new file mode 100644 index 000000000..104751280 --- /dev/null +++ b/pkg/multicluster/monitor_test.go @@ -0,0 +1,93 @@ +// Copyright SAP SE +// SPDX-License-Identifier: Apache-2.0 + +package multicluster + +import ( + "testing" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/testutil" + "k8s.io/apimachinery/pkg/runtime/schema" +) + +func TestMonitor_Registration(t *testing.T) { + monitor := NewMonitor("cortex_") + + registry := prometheus.NewRegistry() + if err := registry.Register(monitor); err != nil { + t.Fatalf("failed to register monitor: %v", err) + } + + // The counter has no values until something is recorded, so it does not + // appear in the gathered families yet. Recording one makes it show up. + gvk := schema.GroupVersionKind{Group: "cortex.cloud", Version: "v1alpha1", Kind: "Reservation"} + monitor.recordCrossClusterNameConflict("create", gvk) + + families, err := registry.Gather() + if err != nil { + t.Fatalf("failed to gather metrics: %v", err) + } + var found bool + for _, f := range families { + if f.GetName() == "cortex_multicluster_cross_cluster_name_conflicts_total" { + found = true + } + } + if !found { + t.Error("expected cortex_multicluster_cross_cluster_name_conflicts_total to be registered") + } +} + +func TestMonitor_Prefix(t *testing.T) { + monitor := NewMonitor("myprefix_") + + registry := prometheus.NewRegistry() + if err := registry.Register(monitor); err != nil { + t.Fatalf("failed to register monitor: %v", err) + } + + gvk := schema.GroupVersionKind{Group: "cortex.cloud", Version: "v1alpha1", Kind: "Reservation"} + monitor.recordCrossClusterNameConflict("create", gvk) + + families, err := registry.Gather() + if err != nil { + t.Fatalf("failed to gather metrics: %v", err) + } + var found bool + for _, f := range families { + if f.GetName() == "myprefix_multicluster_cross_cluster_name_conflicts_total" { + found = true + } + } + if !found { + t.Error("expected metric name to use the supplied prefix") + } +} + +func TestMonitor_RecordCrossClusterNameConflict(t *testing.T) { + gvk := schema.GroupVersionKind{Group: "cortex.cloud", Version: "v1alpha1", Kind: "Reservation"} + otherGVK := schema.GroupVersionKind{Group: "kvm.cloud.sap", Version: "v1", Kind: "Hypervisor"} + + m := NewMonitor("cortex_").(*monitor) + + // Recording accumulates per (method, gvk) label pair. + m.recordCrossClusterNameConflict("create", gvk) + m.recordCrossClusterNameConflict("create", gvk) + m.recordCrossClusterNameConflict("get", gvk) + m.recordCrossClusterNameConflict("list", otherGVK) + + if got := testutil.ToFloat64(m.crossClusterNameConflicts.WithLabelValues("create", gvk.String())); got != 2 { + t.Errorf("create/%s: got %v, want 2", gvk, got) + } + if got := testutil.ToFloat64(m.crossClusterNameConflicts.WithLabelValues("get", gvk.String())); got != 1 { + t.Errorf("get/%s: got %v, want 1", gvk, got) + } + if got := testutil.ToFloat64(m.crossClusterNameConflicts.WithLabelValues("list", otherGVK.String())); got != 1 { + t.Errorf("list/%s: got %v, want 1", otherGVK, got) + } + // A label pair that was never recorded stays at zero. + if got := testutil.ToFloat64(m.crossClusterNameConflicts.WithLabelValues("list", gvk.String())); got != 0 { + t.Errorf("list/%s: got %v, want 0", gvk, got) + } +}