diff --git a/.github/workflows/stale.yaml b/.github/workflows/stale.yaml index 6c53930a0..95da30679 100644 --- a/.github/workflows/stale.yaml +++ b/.github/workflows/stale.yaml @@ -14,7 +14,7 @@ jobs: issues: write pull-requests: write steps: - - uses: actions/stale@v10 + - uses: actions/stale@v11 with: repo-token: ${{ secrets.GITHUB_TOKEN }} days-before-issue-stale: 90 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/go.mod b/go.mod index e5933302a..20ecab4bd 100644 --- a/go.mod +++ b/go.mod @@ -6,10 +6,10 @@ require ( github.com/cobaltcore-dev/openstack-hypervisor-operator v1.2.3 github.com/go-gorp/gorp v2.2.0+incompatible github.com/gophercloud/gophercloud/v2 v2.13.0 - github.com/ironcore-dev/ironcore v0.4.3 + github.com/ironcore-dev/ironcore v0.5.0 github.com/prometheus/client_golang v1.24.1 github.com/prometheus/client_model v0.6.2 - github.com/sapcc/go-bits v0.0.0-20260723170232-89c8670b5841 + github.com/sapcc/go-bits v0.0.0-20260730170321-f6f727520601 go.xyrillian.de/gg v1.11.1 k8s.io/api v0.36.3 k8s.io/apimachinery v0.36.3 @@ -29,7 +29,7 @@ require ( github.com/go-openapi/swag/stringutils v0.25.1 // indirect github.com/go-openapi/swag/typeutils v0.25.1 // indirect github.com/go-openapi/swag/yamlutils v0.25.1 // indirect - github.com/gofrs/uuid/v5 v5.4.0 // indirect + github.com/gofrs/uuid/v5 v5.5.0 // indirect github.com/gorilla/mux v1.8.1 // indirect github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect k8s.io/streaming v0.36.3 // indirect @@ -63,7 +63,7 @@ require ( github.com/go-openapi/swag v0.25.1 // indirect github.com/go-openapi/swag/jsonname v0.25.4 // indirect github.com/golang-migrate/migrate/v4 v4.19.1 // indirect - github.com/google/cel-go v0.26.0 // indirect + github.com/google/cel-go v0.29.0 // indirect github.com/google/gnostic-models v0.7.1 // indirect github.com/google/go-cmp v0.7.0 // indirect github.com/google/uuid v1.6.0 @@ -73,7 +73,7 @@ require ( github.com/json-iterator/go v1.1.12 // indirect github.com/kylelemons/godebug v1.1.0 // indirect github.com/lib/pq v1.12.3 - github.com/mattn/go-sqlite3 v1.14.48 + github.com/mattn/go-sqlite3 v1.14.49 github.com/moby/sys/user v0.4.0 // indirect github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee // indirect @@ -91,7 +91,6 @@ require ( github.com/sirupsen/logrus v1.9.3 // indirect github.com/spf13/cobra v1.10.2 // indirect github.com/spf13/pflag v1.0.10 // indirect - github.com/stoewer/go-strcase v1.3.1 // indirect github.com/x448/float16 v0.8.4 // indirect github.com/ziutek/mymysql v1.5.4 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect diff --git a/go.sum b/go.sum index cfd3b3f69..480a3f594 100644 --- a/go.sum +++ b/go.sum @@ -102,16 +102,16 @@ github.com/go-sql-driver/mysql v1.5.0 h1:ozyZYNQW3x3HtqT1jira07DN2PArx2v7/mN66gG github.com/go-sql-driver/mysql v1.5.0/go.mod h1:DCzpHaOWr8IXmIStZouvnhqoel9Qv2LBy8hT2VhHyBg= github.com/go-task/slim-sprig/v3 v3.0.0 h1:sUs3vkvUymDpBKi3qH1YSqBQk9+9D/8M2mN1vB6EwHI= github.com/go-task/slim-sprig/v3 v3.0.0/go.mod h1:W848ghGpv3Qj3dhTPRyJypKRiqCdHZiAzKg9hl15HA8= -github.com/gofrs/uuid/v5 v5.4.0 h1:EfbpCTjqMuGyq5ZJwxqzn3Cbr2d0rUZU7v5ycAk/e/0= -github.com/gofrs/uuid/v5 v5.4.0/go.mod h1:CDOjlDMVAtN56jqyRUZh58JT31Tiw7/oQyEXZV+9bD8= +github.com/gofrs/uuid/v5 v5.5.0 h1:FkPv6jYQRbZtH3bD8yC7106u+CedTCLF8+t7CLHSZNo= +github.com/gofrs/uuid/v5 v5.5.0/go.mod h1:bbAA98EoIlxyRHIVg6ektCSsZ5n8mSbwgEhvhMYlZgg= github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q= github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q= github.com/golang-migrate/migrate/v4 v4.19.1 h1:OCyb44lFuQfYXYLx1SCxPZQGU7mcaZ7gH9yH4jSFbBA= github.com/golang-migrate/migrate/v4 v4.19.1/go.mod h1:CTcgfjxhaUtsLipnLoQRWCrjYXycRz/g5+RWDuYgPrE= github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= -github.com/google/cel-go v0.26.0 h1:DPGjXackMpJWH680oGY4lZhYjIameYmR+/6RBdDGmaI= -github.com/google/cel-go v0.26.0/go.mod h1:A9O8OU9rdvrK5MQyrqfIxo1a0u4g3sF8KB6PUIaryMM= +github.com/google/cel-go v0.29.0 h1:fEG+Ja3YRwNOqnQxTyJwoByAUAvTuxUGiro/jhrm4F4= +github.com/google/cel-go v0.29.0/go.mod h1:X0bD6iVNR8pkROSOoHVdgTkzmRcosof7WQqCD6wcMc8= github.com/google/gnostic-models v0.7.1 h1:SisTfuFKJSKM5CPZkffwi6coztzzeYUhc3v4yxLWH8c= github.com/google/gnostic-models v0.7.1/go.mod h1:whL5G0m6dmc5cPxKc5bdKdEN3UjI7OUGxBlw57miDrQ= github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= @@ -135,8 +135,8 @@ github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM= github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8= github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw= -github.com/ironcore-dev/ironcore v0.4.3 h1:G3TmR4r3LtbmakMkxDa2pPlLCmxAfGVBEdy+QVyxl8Q= -github.com/ironcore-dev/ironcore v0.4.3/go.mod h1:vFpdcyC4QD7o3j01lpux12NET6vNPDAq3hzrFOY9QOc= +github.com/ironcore-dev/ironcore v0.5.0 h1:QAbXuqD8ZfS8wNP++EamrV7Xav3xfeZsC+28hddpkNE= +github.com/ironcore-dev/ironcore v0.5.0/go.mod h1:HPW1+WJGharyrX30vOm43cGpeBXQDP43r66S4ADnhZ0= github.com/itchyny/gojq v0.12.19 h1:ttXA0XCLEMoaLOz5lSeFOZ6u6Q3QxmG46vfgI4O0DEs= github.com/itchyny/gojq v0.12.19/go.mod h1:5galtVPDywX8SPSOrqjGxkBeDhSxEW1gSxoy7tn1iZY= github.com/itchyny/timefmt-go v0.1.8 h1:1YEo1JvfXeAHKdjelbYr/uCuhkybaHCeTkH8Bo791OI= @@ -153,8 +153,8 @@ github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0 github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw= github.com/lib/pq v1.12.3 h1:tTWxr2YLKwIvK90ZXEw8GP7UFHtcbTtty8zsI+YjrfQ= github.com/lib/pq v1.12.3/go.mod h1:/p+8NSbOcwzAEI7wiMXFlgydTwcgTr3OSKMsD2BitpA= -github.com/mattn/go-sqlite3 v1.14.48 h1:7XHIgl0a8HwOaiK4E47ozLkST78rR9+OtNGx27D/TFs= -github.com/mattn/go-sqlite3 v1.14.48/go.mod h1:6JTjA44L93a0QCyJef5YvlPoKXntQPjzWv5gtm9sB6w= +github.com/mattn/go-sqlite3 v1.14.49 h1:B8jBHC3xhxZgxztrgruTuLucebnULQnx4W7cF7SAE9w= +github.com/mattn/go-sqlite3 v1.14.49/go.mod h1:6JTjA44L93a0QCyJef5YvlPoKXntQPjzWv5gtm9sB6w= github.com/moby/docker-image-spec v1.3.1 h1:jMKff3w6PgbfSa69GfNg+zN/XLhfXJGnEx3Nl2EsFP0= github.com/moby/docker-image-spec v1.3.1/go.mod h1:eKmb5VW8vQEh/BAr2yvVNvuiJuY6UIocYsFu/DxxRpo= github.com/moby/sys/user v0.4.0 h1:jhcMKit7SA80hivmFJcbB1vqmw//wU61Zdui2eQXuMs= @@ -204,8 +204,8 @@ github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7 github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM= github.com/sapcc/go-api-declarations v1.24.0 h1:sGBvOMVSM1olJlyvNoQSk7NX5uatXHKkztGDBPnTWMs= github.com/sapcc/go-api-declarations v1.24.0/go.mod h1:ZWRTijvgF8o8aHg5stgg7u4DF6jFrd0X97le/uGlZsA= -github.com/sapcc/go-bits v0.0.0-20260723170232-89c8670b5841 h1:xU7PCY7Vcc/0UZhvybOsE+bhh8/RmZlH9UaKIiKzNAY= -github.com/sapcc/go-bits v0.0.0-20260723170232-89c8670b5841/go.mod h1:wUu2ss9a/5/bR4QOcaPIklrZdconXafkbsLgProV0Ro= +github.com/sapcc/go-bits v0.0.0-20260730170321-f6f727520601 h1:pF4eF41dp0AxM0IfICn3ttpI5IwQoUmC0IkGuXLTmBA= +github.com/sapcc/go-bits v0.0.0-20260730170321-f6f727520601/go.mod h1:4z4Vd1C7d6hDdbJ5q1CgzE9B8ESWuiMTbkZSJA5gSjA= github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ= github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ= github.com/spf13/cobra v1.10.2 h1:DMTTonx5m65Ic0GOoRY2c16WCbHxOOw6xxezuLaBpcU= @@ -213,18 +213,11 @@ github.com/spf13/cobra v1.10.2/go.mod h1:7C1pvHqHw5A4vrJfjNwvOdzYu0Gml16OCs2GRiT github.com/spf13/pflag v1.0.9/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= github.com/spf13/pflag v1.0.10 h1:4EBh2KAYBwaONj6b2Ye1GiHfwjqyROoF4RwYO+vPwFk= github.com/spf13/pflag v1.0.10/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= -github.com/stoewer/go-strcase v1.3.1 h1:iS0MdW+kVTxgMoE1LAZyMiYJFKlOzLooE4MxjirtkAs= -github.com/stoewer/go-strcase v1.3.1/go.mod h1:fAH5hQ5pehh+j3nZfvwdk2RgEgQjAoM8wodgtPmh1xo= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= -github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw= -github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo= github.com/stretchr/objx v0.5.2 h1:xuMeJ0Sdp5ZMRXx/aWO6RZxdr3beISkG5/G/aIRr3pY= github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA= github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= -github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= -github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU= -github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM= 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/helm/dev/cortex-prometheus-operator/Chart.yaml b/helm/dev/cortex-prometheus-operator/Chart.yaml index 406d41f58..35e4f1a77 100644 --- a/helm/dev/cortex-prometheus-operator/Chart.yaml +++ b/helm/dev/cortex-prometheus-operator/Chart.yaml @@ -10,4 +10,4 @@ dependencies: # CRDs of the prometheus operator, such as PrometheusRule, ServiceMonitor, etc. - name: kube-prometheus-stack repository: oci://ghcr.io/prometheus-community/charts - version: 87.19.2 + version: 87.21.0 diff --git a/internal/scheduling/cinder/external_scheduler_api.go b/internal/scheduling/cinder/external_scheduler_api.go index 3be3b6d2c..b234e5600 100644 --- a/internal/scheduling/cinder/external_scheduler_api.go +++ b/internal/scheduling/cinder/external_scheduler_api.go @@ -157,7 +157,7 @@ func (httpAPI *httpAPI) CinderExternalScheduler(w http.ResponseWriter, r *http.R } ctx := r.Context() if err := httpAPI.delegate.ProcessNewDecisionFromAPI(ctx, decision); err != nil { - c.Respond(logger, http.StatusInternalServerError, err, "failed to process scheduling decision") + c.Respond(logger, http.StatusInternalServerError, err, fmt.Sprintf("failed to process scheduling decision: %v", err)) return } // Check if the decision contains status conditions indicating an error. diff --git a/internal/scheduling/cinder/filter_weigher_pipeline_controller.go b/internal/scheduling/cinder/filter_weigher_pipeline_controller.go index e746ca0b2..bb1ea8e31 100644 --- a/internal/scheduling/cinder/filter_weigher_pipeline_controller.go +++ b/internal/scheduling/cinder/filter_weigher_pipeline_controller.go @@ -7,6 +7,7 @@ import ( "context" "encoding/json" "errors" + "fmt" "sync" "time" @@ -99,7 +100,7 @@ func (c *FilterWeigherPipelineController) process(ctx context.Context, decision pipeline, ok := c.GetPipeline(decision.Spec.PipelineRef.Name) if !ok { log.Error(nil, "pipeline not found or not ready", "pipelineName", decision.Spec.PipelineRef.Name) - return errors.New("pipeline not found or not ready") + return fmt.Errorf("pipeline not found or not ready: %q", decision.Spec.PipelineRef.Name) } if decision.Spec.CinderRaw == nil { log.Error(nil, "skipping decision, no cinderRaw spec defined") diff --git a/internal/scheduling/machines/filter_weigher_pipeline_controller.go b/internal/scheduling/machines/filter_weigher_pipeline_controller.go index cbbf684c5..ebfddcc5c 100644 --- a/internal/scheduling/machines/filter_weigher_pipeline_controller.go +++ b/internal/scheduling/machines/filter_weigher_pipeline_controller.go @@ -6,6 +6,7 @@ package machines import ( "context" "errors" + "fmt" "sync" "time" @@ -120,7 +121,7 @@ func (c *FilterWeigherPipelineController) process(ctx context.Context, decision pipeline, ok := c.GetPipeline(decision.Spec.PipelineRef.Name) if !ok { log.Error(nil, "pipeline not found or not ready", "pipelineName", decision.Spec.PipelineRef.Name) - return errors.New("pipeline not found or not ready") + return fmt.Errorf("pipeline not found or not ready: %q", decision.Spec.PipelineRef.Name) } // Find all available machine pools. diff --git a/internal/scheduling/manila/external_scheduler_api.go b/internal/scheduling/manila/external_scheduler_api.go index a406c1778..ae7b2eb95 100644 --- a/internal/scheduling/manila/external_scheduler_api.go +++ b/internal/scheduling/manila/external_scheduler_api.go @@ -157,7 +157,7 @@ func (httpAPI *httpAPI) ManilaExternalScheduler(w http.ResponseWriter, r *http.R } ctx := r.Context() if err := httpAPI.delegate.ProcessNewDecisionFromAPI(ctx, decision); err != nil { - c.Respond(logger, http.StatusInternalServerError, err, "failed to process scheduling decision") + c.Respond(logger, http.StatusInternalServerError, err, fmt.Sprintf("failed to process scheduling decision: %v", err)) return } // Check if the decision contains status conditions indicating an error. diff --git a/internal/scheduling/manila/filter_weigher_pipeline_controller.go b/internal/scheduling/manila/filter_weigher_pipeline_controller.go index 686ce4c2a..347e14681 100644 --- a/internal/scheduling/manila/filter_weigher_pipeline_controller.go +++ b/internal/scheduling/manila/filter_weigher_pipeline_controller.go @@ -7,6 +7,7 @@ import ( "context" "encoding/json" "errors" + "fmt" "sync" "time" @@ -98,8 +99,8 @@ func (c *FilterWeigherPipelineController) process(ctx context.Context, decision pipeline, ok := c.GetPipeline(decision.Spec.PipelineRef.Name) if !ok { - log.Error(nil, "skipping decision, pipeline not found or not ready") - return errors.New("pipeline not found or not ready") + log.Error(nil, "skipping decision, pipeline not found or not ready", "pipelineName", decision.Spec.PipelineRef.Name) + return fmt.Errorf("pipeline not found or not ready: %q", decision.Spec.PipelineRef.Name) } if decision.Spec.ManilaRaw == nil { log.Error(nil, "skipping decision, no manilaRaw spec defined") diff --git a/internal/scheduling/nova/external_scheduler_api.go b/internal/scheduling/nova/external_scheduler_api.go index 7bd88ba8f..5773da9c7 100644 --- a/internal/scheduling/nova/external_scheduler_api.go +++ b/internal/scheduling/nova/external_scheduler_api.go @@ -245,7 +245,7 @@ func (httpAPI *httpAPI) NovaExternalScheduler(w http.ResponseWriter, r *http.Req } ctx := r.Context() if err := httpAPI.delegate.ProcessNewDecisionFromAPI(ctx, decision); err != nil { - c.Respond(logger, http.StatusInternalServerError, err, "failed to process scheduling decision") + c.Respond(logger, http.StatusInternalServerError, err, fmt.Sprintf("failed to process scheduling decision: %v", err)) return } // Check if the decision contains status conditions indicating an error. diff --git a/internal/scheduling/nova/filter_weigher_pipeline_controller.go b/internal/scheduling/nova/filter_weigher_pipeline_controller.go index 2682c96b4..5b4b9cae3 100644 --- a/internal/scheduling/nova/filter_weigher_pipeline_controller.go +++ b/internal/scheduling/nova/filter_weigher_pipeline_controller.go @@ -7,6 +7,7 @@ import ( "context" "encoding/json" "errors" + "fmt" "sync" "time" @@ -163,7 +164,7 @@ func (c *FilterWeigherPipelineController) process(ctx context.Context, decision pipeline, ok := c.GetPipeline(decision.Spec.PipelineRef.Name) if !ok { log.Error(nil, "pipeline not found or not ready", "pipelineName", decision.Spec.PipelineRef.Name) - return nil, errors.New("pipeline not found or not ready") + return nil, fmt.Errorf("pipeline not found or not ready: %q", decision.Spec.PipelineRef.Name) } if decision.Spec.NovaRaw == nil { log.Error(nil, "skipping decision, no novaRaw spec defined") @@ -187,7 +188,7 @@ func (c *FilterWeigherPipelineController) process(ctx context.Context, decision pipelineConf, ok := c.GetPipelineConfig(decision.Spec.PipelineRef.Name) if !ok { log.Error(nil, "pipeline config not found", "pipelineName", decision.Spec.PipelineRef.Name) - return &request, errors.New("pipeline config not found") + return &request, fmt.Errorf("pipeline config not found: %q", decision.Spec.PipelineRef.Name) } if pipelineConf.Spec.IgnorePreselection { log.Info("gathering all placement candidates before filtering") diff --git a/internal/scheduling/nova/plugins/filters/filter_aggregate_metadata.go b/internal/scheduling/nova/plugins/filters/filter_aggregate_metadata.go index f55e8ab7b..08a7e55c3 100644 --- a/internal/scheduling/nova/plugins/filters/filter_aggregate_metadata.go +++ b/internal/scheduling/nova/plugins/filters/filter_aggregate_metadata.go @@ -7,6 +7,7 @@ import ( "context" "log/slog" "slices" + "strings" api "github.com/cobaltcore-dev/cortex/api/external/nova" "github.com/cobaltcore-dev/cortex/api/v1alpha1" @@ -40,15 +41,31 @@ func (s *FilterAggregateMetadata) Run(traceLog *slog.Logger, request api.Externa restrictedProjectsByHost := make(map[string][]string) for _, hv := range hvs.Items { for _, aggregate := range hv.Status.Aggregates { - tenantID, ok := aggregate.Metadata["filter_tenant_id"] - if !ok { + // Any metadata key prefixed with "filter_tenant_id" restricts the + // aggregate to the referenced projects. Multiple numbered keys (e.g. + // filter_tenant_id, filter_tenant_id1, ...) circumvent per-field + // database string length limits; each of their values can itself be + // a comma-separated list of project ids. + foundFilter := false + for key, value := range aggregate.Metadata { + if !strings.HasPrefix(key, "filter_tenant_id") { + continue + } + foundFilter = true + for projectID := range strings.SplitSeq(value, ",") { + projectID = strings.TrimSpace(projectID) + if projectID == "" { + continue + } + restrictedProjectsByHost[hv.Name] = append(restrictedProjectsByHost[hv.Name], projectID) + } + traceLog.Info("host is in aggregate with filter_tenant_id, adding restriction", + "host", hv.Name, "aggregate", aggregate.Name, "key", key, "tenant_id", value) + } + if !foundFilter { traceLog.Info("aggregate does not have filter_tenant_id metadata, skipping", "aggregate", aggregate.Name) - continue } - restrictedProjectsByHost[hv.Name] = append(restrictedProjectsByHost[hv.Name], tenantID) - traceLog.Info("host is in aggregate with filter_tenant_id, adding restriction", - "host", hv.Name, "aggregate", aggregate.Name, "tenant_id", tenantID) } } diff --git a/internal/scheduling/nova/plugins/filters/filter_aggregate_metadata_test.go b/internal/scheduling/nova/plugins/filters/filter_aggregate_metadata_test.go index 42c096a2b..5449bc362 100644 --- a/internal/scheduling/nova/plugins/filters/filter_aggregate_metadata_test.go +++ b/internal/scheduling/nova/plugins/filters/filter_aggregate_metadata_test.go @@ -316,6 +316,155 @@ func TestFilterAggregateMetadata_Run(t *testing.T) { expectedHosts: []string{"host1"}, filteredHosts: []string{}, }, + { + name: "Comma-separated filter_tenant_id - project matches one of many", + request: api.ExternalSchedulerRequest{ + Spec: api.NovaObject[api.NovaSpec]{ + Data: api.NovaSpec{ + ProjectID: "project-b", + }, + }, + Hosts: []api.ExternalSchedulerHost{ + {ComputeHost: "host1"}, + }, + }, + hypervisors: []hv1.Hypervisor{ + { + ObjectMeta: metav1.ObjectMeta{Name: "host1"}, + Status: hv1.HypervisorStatus{ + Aggregates: []hv1.Aggregate{{Name: "restricted", Metadata: map[string]string{"filter_tenant_id": "project-a,project-b,project-c"}}}, + }, + }, + }, + expectedHosts: []string{"host1"}, + filteredHosts: []string{}, + }, + { + name: "Comma-separated filter_tenant_id - project matches none", + request: api.ExternalSchedulerRequest{ + Spec: api.NovaObject[api.NovaSpec]{ + Data: api.NovaSpec{ + ProjectID: "project-d", + }, + }, + Hosts: []api.ExternalSchedulerHost{ + {ComputeHost: "host1"}, + }, + }, + hypervisors: []hv1.Hypervisor{ + { + ObjectMeta: metav1.ObjectMeta{Name: "host1"}, + Status: hv1.HypervisorStatus{ + Aggregates: []hv1.Aggregate{{Name: "restricted", Metadata: map[string]string{"filter_tenant_id": "project-a,project-b,project-c"}}}, + }, + }, + }, + expectedHosts: []string{}, + filteredHosts: []string{"host1"}, + }, + { + name: "Comma-separated filter_tenant_id with surrounding whitespace - project matches", + request: api.ExternalSchedulerRequest{ + Spec: api.NovaObject[api.NovaSpec]{ + Data: api.NovaSpec{ + ProjectID: "project-b", + }, + }, + Hosts: []api.ExternalSchedulerHost{ + {ComputeHost: "host1"}, + }, + }, + hypervisors: []hv1.Hypervisor{ + { + ObjectMeta: metav1.ObjectMeta{Name: "host1"}, + Status: hv1.HypervisorStatus{ + Aggregates: []hv1.Aggregate{{Name: "restricted", Metadata: map[string]string{"filter_tenant_id": "project-a, project-b , project-c"}}}, + }, + }, + }, + expectedHosts: []string{"host1"}, + filteredHosts: []string{}, + }, + { + name: "Numbered filter_tenant_id keys - project matches one of the numbered keys", + request: api.ExternalSchedulerRequest{ + Spec: api.NovaObject[api.NovaSpec]{ + Data: api.NovaSpec{ + ProjectID: "project-c", + }, + }, + Hosts: []api.ExternalSchedulerHost{ + {ComputeHost: "host1"}, + }, + }, + hypervisors: []hv1.Hypervisor{ + { + ObjectMeta: metav1.ObjectMeta{Name: "host1"}, + Status: hv1.HypervisorStatus{ + Aggregates: []hv1.Aggregate{{Name: "restricted", Metadata: map[string]string{ + "filter_tenant_id": "project-a", + "filter_tenant_id1": "project-b", + "filter_tenant_id2": "project-c", + }}}, + }, + }, + }, + expectedHosts: []string{"host1"}, + filteredHosts: []string{}, + }, + { + name: "Numbered filter_tenant_id keys with comma-separated values - project matches", + request: api.ExternalSchedulerRequest{ + Spec: api.NovaObject[api.NovaSpec]{ + Data: api.NovaSpec{ + ProjectID: "project-e", + }, + }, + Hosts: []api.ExternalSchedulerHost{ + {ComputeHost: "host1"}, + }, + }, + hypervisors: []hv1.Hypervisor{ + { + ObjectMeta: metav1.ObjectMeta{Name: "host1"}, + Status: hv1.HypervisorStatus{ + Aggregates: []hv1.Aggregate{{Name: "restricted", Metadata: map[string]string{ + "filter_tenant_id": "project-a,project-b", + "filter_tenant_id1": "project-c,project-d", + "filter_tenant_id2": "project-e,project-f", + }}}, + }, + }, + }, + expectedHosts: []string{"host1"}, + filteredHosts: []string{}, + }, + { + name: "Numbered filter_tenant_id keys - project matches none", + request: api.ExternalSchedulerRequest{ + Spec: api.NovaObject[api.NovaSpec]{ + Data: api.NovaSpec{ + ProjectID: "project-z", + }, + }, + Hosts: []api.ExternalSchedulerHost{ + {ComputeHost: "host1"}, + }, + }, + hypervisors: []hv1.Hypervisor{ + { + ObjectMeta: metav1.ObjectMeta{Name: "host1"}, + Status: hv1.HypervisorStatus{ + Aggregates: []hv1.Aggregate{{Name: "restricted", Metadata: map[string]string{ + "filter_tenant_id": "project-a,project-b", + "filter_tenant_id1": "project-c,project-d", + }}}, + }, + }, + }, + expectedHosts: []string{}, + filteredHosts: []string{"host1"}, + }, } for _, tt := range tests { diff --git a/internal/scheduling/pods/filter_weigher_pipeline_controller.go b/internal/scheduling/pods/filter_weigher_pipeline_controller.go index c251c8b45..cf27558ed 100644 --- a/internal/scheduling/pods/filter_weigher_pipeline_controller.go +++ b/internal/scheduling/pods/filter_weigher_pipeline_controller.go @@ -6,6 +6,7 @@ package pods import ( "context" "errors" + "fmt" "sync" "time" @@ -120,7 +121,7 @@ func (c *FilterWeigherPipelineController) process(ctx context.Context, decision pipeline, ok := c.GetPipeline(decision.Spec.PipelineRef.Name) if !ok { log.Error(nil, "pipeline not found or not ready", "pipelineName", decision.Spec.PipelineRef.Name) - return errors.New("pipeline not found or not ready") + return fmt.Errorf("pipeline not found or not ready: %q", decision.Spec.PipelineRef.Name) } // Check if the pod is already assigned to a node. diff --git a/internal/scheduling/reservations/failover/reservation_scheduling.go b/internal/scheduling/reservations/failover/reservation_scheduling.go index efd3876f3..7d7ea0c99 100644 --- a/internal/scheduling/reservations/failover/reservation_scheduling.go +++ b/internal/scheduling/reservations/failover/reservation_scheduling.go @@ -40,6 +40,38 @@ func inferFailoverPipeline(extraSpecs map[string]string) string { return "kvm-general-purpose-load-balancing" } +// imagePropertiesFromFlavor synthesizes Glance-style image properties from a flavor's +// capabilities:hypervisor_type extra spec. The failover controller does not have access +// to the real Glance image metadata of the VM, but for reservation scheduling the VM +// already exists and has passed FilterImagePropertiesStep at original create time, so +// this synthesis is safe: it exists only to keep the filter from emitting spurious +// "image_properties_hv_type_undetermined" events when the failover controller reuses +// the EvacuateIntent to validate reservation hosts. +// +// The extra-spec values below duplicate the canonical mapping in +// nova.Flavor.GetHypervisorType() (see internal/knowledge/datasources/plugins/openstack/nova/nova_types.go); +// keep the two in sync. Real observed fixtures also carry lowercase "qemu"/"ch", so +// those are accepted here too. +// +// Returns nil for unknown/unspecified hypervisor types, which preserves the filter's +// existing tolerant fallback (log a warning event, keep all hosts). +// +// TODO(vm-crd): remove this and populate image properties from real Glance metadata +// once the VM CRD carries them. See also the TODO on validateVMViaSchedulerEvacuation. +func imagePropertiesFromFlavor(flavorExtraSpecs map[string]string) map[string]any { + switch flavorExtraSpecs["capabilities:hypervisor_type"] { + case "QEMU", "CH", "qemu", "ch": + // KVM family; the kvm-flavored key is "img_hv_type". + return map[string]any{"img_hv_type": string(api.NovaImageMetaHVTypeKVM)} + case "VMware vCenter Server": + return map[string]any{"hypervisor_type": string(api.NovaImageMetaHVTypeVMware)} + case "Ironic": + return map[string]any{"hypervisor_type": string(api.NovaImageMetaHVTypeBaremetal)} + default: + return nil + } +} + func (c *FailoverReservationController) queryHypervisorsFromScheduler(ctx context.Context, vm reservations.VM, allHypervisors []string, pipeline string, resSpec resolvedReservationSpec, intent v1alpha1.SchedulingIntent, opts scheduling.Options) ([]string, error) { logger := LoggerFromContext(ctx) @@ -93,6 +125,7 @@ func (c *FailoverReservationController) queryHypervisorsFromScheduler(ctx contex "_nova_check_type": string(intent), api.HintKeyResourceGroup: resSpec.ResourceGroup(vm.FlavorName), }, + ImageProperties: imagePropertiesFromFlavor(flavorExtraSpecs), } logger.V(1).Info("scheduling failover reservation", @@ -226,6 +259,7 @@ func (c *FailoverReservationController) validateVMViaSchedulerEvacuation( Pipeline: inferFailoverPipeline(flavorExtraSpecs), AvailabilityZone: vm.AvailabilityZone, SchedulerHints: map[string]any{"_nova_check_type": string(api.EvacuateIntent)}, + ImageProperties: imagePropertiesFromFlavor(flavorExtraSpecs), } logger.V(1).Info("validating VM via scheduler evacuation", diff --git a/internal/scheduling/reservations/failover/reservation_scheduling_test.go b/internal/scheduling/reservations/failover/reservation_scheduling_test.go index 6ae2ab820..bc51c6213 100644 --- a/internal/scheduling/reservations/failover/reservation_scheduling_test.go +++ b/internal/scheduling/reservations/failover/reservation_scheduling_test.go @@ -572,3 +572,64 @@ func TestFailoverSchedulerOptions(t *testing.T) { }) } } + +// ============================================================================ +// Test: imagePropertiesFromFlavor +// ============================================================================ + +func TestImagePropertiesFromFlavor(t *testing.T) { + tests := []struct { + name string + extraSpecs map[string]string + want map[string]any + }{ + { + name: "qemu (lowercase) -> kvm img_hv_type", + extraSpecs: map[string]string{"capabilities:hypervisor_type": "qemu"}, + want: map[string]any{"img_hv_type": "kvm"}, + }, + { + name: "QEMU (uppercase) -> kvm img_hv_type", + extraSpecs: map[string]string{"capabilities:hypervisor_type": "QEMU"}, + want: map[string]any{"img_hv_type": "kvm"}, + }, + { + name: "CH -> kvm img_hv_type", + extraSpecs: map[string]string{"capabilities:hypervisor_type": "CH"}, + want: map[string]any{"img_hv_type": "kvm"}, + }, + { + name: "VMware -> hypervisor_type=vmware", + extraSpecs: map[string]string{"capabilities:hypervisor_type": "VMware vCenter Server"}, + want: map[string]any{"hypervisor_type": "vmware"}, + }, + { + name: "Ironic -> hypervisor_type=baremetal", + extraSpecs: map[string]string{"capabilities:hypervisor_type": "Ironic"}, + want: map[string]any{"hypervisor_type": "baremetal"}, + }, + { + name: "unspecified -> nil", + extraSpecs: map[string]string{}, + want: nil, + }, + { + name: "unknown value -> nil", + extraSpecs: map[string]string{"capabilities:hypervisor_type": "SomeFuturisticThing"}, + want: nil, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := imagePropertiesFromFlavor(tt.extraSpecs) + if len(got) != len(tt.want) { + t.Fatalf("length mismatch: got %v want %v", got, tt.want) + } + for k, v := range tt.want { + if got[k] != v { + t.Errorf("key %q: got %v want %v", k, got[k], v) + } + } + }) + } +} diff --git a/internal/scheduling/reservations/scheduler_client.go b/internal/scheduling/reservations/scheduler_client.go index fb3ef269b..a03f9cd72 100644 --- a/internal/scheduling/reservations/scheduler_client.go +++ b/internal/scheduling/reservations/scheduler_client.go @@ -79,6 +79,10 @@ type ScheduleReservationRequest struct { // SchedulerHints are hints passed to the scheduler pipeline. // Used to set _nova_check_type for evacuation intent detection. SchedulerHints map[string]any + // ImageProperties are Glance-style image properties (e.g., "img_hv_type", "hypervisor_type") + // forwarded on Spec.Data.Image.Data.Properties.Data. Optional; when empty the outgoing + // request leaves the image field at its zero value. Used by FilterImagePropertiesStep. + ImageProperties map[string]any } // ScheduleReservationResponse contains the result of scheduling a reservation. @@ -143,6 +147,13 @@ func (c *SchedulerClient) ScheduleReservation(ctx context.Context, req ScheduleR // Disk is currently not considered. }, }, + Image: api.NovaObject[api.NovaImageMeta]{ + Data: api.NovaImageMeta{ + Properties: api.NovaObject[map[string]any]{ + Data: req.ImageProperties, + }, + }, + }, }, }, } 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) + } +}