diff --git a/backend/pkg/app/backend.go b/backend/pkg/app/backend.go index dce9bb4ec86..f96cc422765 100644 --- a/backend/pkg/app/backend.go +++ b/backend/pkg/app/backend.go @@ -881,6 +881,21 @@ func (b *Backend) runBackendControllersUnderLeaderElection(ctx context.Context, backendInformers, unionKubeApplierInformers, ) + _, managementClusterSchedulingLister := fleetInformers.ManagementClusterSchedulings() + placementController := clusterplacement.NewPlacementController( + b.options.ResourcesDBClient, + b.options.FleetDBClient, + b.options.ClustersServiceClient, + managementClusterLister, + managementClusterSchedulingLister, + backendInformers, + unionKubeApplierInformers, + ) + pendingCleanupController := clusterplacement.NewPendingCleanupController( + b.options.FleetDBClient, + serviceProviderClusterLister, + fleetInformers, + ) nodePoolClusterServiceCreateController := nodepoolcreation.NewNodePoolClusterServiceCreateController( b.options.ResourcesDBClient, @@ -952,6 +967,7 @@ func (b *Backend) runBackendControllersUnderLeaderElection(ctx context.Context, clusterClusterServiceCreateController := clustercreation.NewClusterClusterServiceCreateController( b.options.ResourcesDBClient, b.options.ClustersServiceClient, + managementClusterLister, backendInformers, ) @@ -1127,6 +1143,8 @@ func (b *Backend) runBackendControllersUnderLeaderElection(ctx context.Context, go externalAuthMetricsController.Run(ctx, 1) go clusterInfoMetricsController.Run(ctx, 1) go placementSyncController.Run(ctx, 20) + go placementController.Run(ctx, 20) + go pendingCleanupController.Run(ctx, 5) go cosmosMigrationController.Run(ctx, 5) go virtualMachineResourceSKUsCachedReaderController.Run(ctx, 20) go backupScheduleController.Run(ctx, 20) diff --git a/backend/pkg/controllers/cluster/creation/cluster_cluster_service_create_controller.go b/backend/pkg/controllers/cluster/creation/cluster_cluster_service_create_controller.go index 2fbbd3ab154..12f8670ec2e 100644 --- a/backend/pkg/controllers/cluster/creation/cluster_cluster_service_create_controller.go +++ b/backend/pkg/controllers/cluster/creation/cluster_cluster_service_create_controller.go @@ -29,15 +29,17 @@ import ( "github.com/Azure/ARO-HCP/internal/database/cosmosstorage/cosmosstorageutils" "github.com/Azure/ARO-HCP/internal/database/informers/coreinformers" "github.com/Azure/ARO-HCP/internal/database/listers/corelisters" + "github.com/Azure/ARO-HCP/internal/database/listers/fleetlisters" "github.com/Azure/ARO-HCP/internal/ocm" "github.com/Azure/ARO-HCP/internal/utils" ) type clusterClusterServiceCreateSyncer struct { - resourcesDBClient corecosmosstorage.ResourcesDBClient - clusterLister corelisters.ClusterLister - subscriptionLister corelisters.SubscriptionLister - clustersServiceClient ocm.ClusterServiceClientSpec + resourcesDBClient corecosmosstorage.ResourcesDBClient + clusterLister corelisters.ClusterLister + subscriptionLister corelisters.SubscriptionLister + managementClusterLister fleetlisters.ManagementClusterLister + clustersServiceClient ocm.ClusterServiceClientSpec } var _ controllerutils.ClusterSyncer = (*clusterClusterServiceCreateSyncer)(nil) @@ -45,15 +47,17 @@ var _ controllerutils.ClusterSyncer = (*clusterClusterServiceCreateSyncer)(nil) func NewClusterClusterServiceCreateController( resourcesDBClient corecosmosstorage.ResourcesDBClient, clustersServiceClient ocm.ClusterServiceClientSpec, + managementClusterLister fleetlisters.ManagementClusterLister, backendInformers coreinformers.BackendInformers, ) controllerutils.Controller { _, clusterLister := backendInformers.Clusters() _, subscriptionLister := backendInformers.Subscriptions() syncer := &clusterClusterServiceCreateSyncer{ - resourcesDBClient: resourcesDBClient, - clusterLister: clusterLister, - subscriptionLister: subscriptionLister, - clustersServiceClient: clustersServiceClient, + resourcesDBClient: resourcesDBClient, + clusterLister: clusterLister, + subscriptionLister: subscriptionLister, + managementClusterLister: managementClusterLister, + clustersServiceClient: clustersServiceClient, } return controllerutils.NewClusterWatchingController( @@ -131,6 +135,17 @@ func (c *clusterClusterServiceCreateSyncer) SyncOnce(ctx context.Context, key co } if csCluster == nil { + // Placement gate: the scheduler records its chosen management cluster on + // ServiceProviderCluster.Spec.ManagementClusterResourceID, which + // createClusterServiceCluster needs to pin the Cluster Service provision + // shard. If placement has not been resolved yet, do not create the CS + // cluster and do not return an error: erroring would re-enqueue and churn + // the workqueue. Return nil instead — the ServiceProviderCluster update the + // PlacementController makes once placement lands re-triggers this cluster. + if existingServiceProviderCluster.Spec.ManagementClusterResourceID == nil { + logger.Info("ServiceProviderCluster has no Spec.ManagementClusterResourceID yet; deferring Cluster Service cluster creation until placement is resolved") + return nil + } csCluster, err = c.createClusterServiceCluster(ctx, cluster, existingServiceProviderCluster, tenantID) if err != nil { return utils.TrackError(fmt.Errorf("failed to create cluster in CS: %w", err)) @@ -235,10 +250,18 @@ func (c *clusterClusterServiceCreateSyncer) csClustersMatchingClusterByAzureInfo func (c *clusterClusterServiceCreateSyncer) createClusterServiceCluster(ctx context.Context, cluster *coreapi.HCPOpenShiftCluster, serviceProviderCluster *coreapi.ServiceProviderCluster, tenantID string) (*arohcpv1alpha1.Cluster, error) { logger := utils.LoggerFromContext(ctx) + provisionShardID, err := c.provisionShardID(ctx, serviceProviderCluster) + if err != nil { + return nil, utils.TrackError(err) + } + csClusterBuilder, err := ocm.BuildCSCluster(cluster.ID, tenantID, cluster, nil, nil, serviceProviderCluster) if err != nil { return nil, utils.TrackError(fmt.Errorf("failed to build CS cluster: %w", err)) } + // Pin the CS provision shard for the scheduler-selected management cluster via + // the SDK builder method. + csClusterBuilder.ProvisionShardID(provisionShardID) clusterServiceUID := cluster.ServiceProviderProperties.PendingClusterServiceID.ClusterID() logger.Info("Creating cluster in Cluster Service", "version", serviceProviderCluster.Spec.ControlPlaneVersion.DesiredVersion.String()) @@ -253,3 +276,36 @@ func (c *clusterClusterServiceCreateSyncer) createClusterServiceCluster(ctx cont return result, nil } + +// provisionShardID resolves the Cluster Service provision shard ID for the +// management cluster the scheduler pinned on +// ServiceProviderCluster.Spec.ManagementClusterResourceID. The caller sets it on +// the CS cluster via ClusterBuilder.ProvisionShardID so the new CS cluster is +// created on the correct provision shard. +func (c *clusterClusterServiceCreateSyncer) provisionShardID(ctx context.Context, serviceProviderCluster *coreapi.ServiceProviderCluster) (string, error) { + managementClusterResourceID := serviceProviderCluster.Spec.ManagementClusterResourceID + if managementClusterResourceID == nil { + return "", fmt.Errorf("ServiceProviderCluster has no Spec.ManagementClusterResourceID; placement is not resolved") + } + // A management cluster is a singleton within a stamp, so its resource ID is + // .../stamps//managementClusters/default and the lister is + // keyed by the stamp identifier (the parent segment's name). + if managementClusterResourceID.Parent == nil { + return "", fmt.Errorf("management cluster resource ID %q has no parent stamp", managementClusterResourceID.String()) + } + stampIdentifier := managementClusterResourceID.Parent.Name + + managementCluster, err := c.managementClusterLister.Get(ctx, stampIdentifier) + if cosmosstorageutils.IsNotFoundError(err) { + return "", fmt.Errorf("management cluster %q not found", managementClusterResourceID.String()) + } + if err != nil { + return "", utils.TrackError(fmt.Errorf("failed to get management cluster %q: %w", managementClusterResourceID.String(), err)) + } + + if managementCluster.Status.ClusterServiceProvisionShardID == nil { + return "", fmt.Errorf("management cluster %q has no ClusterServiceProvisionShardID", managementClusterResourceID.String()) + } + + return managementCluster.Status.ClusterServiceProvisionShardID.ID(), nil +} diff --git a/backend/pkg/controllers/cluster/creation/cluster_cluster_service_create_controller_test.go b/backend/pkg/controllers/cluster/creation/cluster_cluster_service_create_controller_test.go index 27c9dc0783e..8549cb05225 100644 --- a/backend/pkg/controllers/cluster/creation/cluster_cluster_service_create_controller_test.go +++ b/backend/pkg/controllers/cluster/creation/cluster_cluster_service_create_controller_test.go @@ -34,10 +34,12 @@ import ( "github.com/Azure/ARO-HCP/backend/pkg/utils/controllerutils" "github.com/Azure/ARO-HCP/internal/api/coreapi" + "github.com/Azure/ARO-HCP/internal/api/fleetapi" "github.com/Azure/ARO-HCP/internal/api/metadataapi" "github.com/Azure/ARO-HCP/internal/apitesting/coreapitesting" "github.com/Azure/ARO-HCP/internal/database/cosmosstoragetesting/corecosmosstoragetesting" "github.com/Azure/ARO-HCP/internal/database/listertesting/corelistertesting" + "github.com/Azure/ARO-HCP/internal/database/listertesting/fleetlistertesting" "github.com/Azure/ARO-HCP/internal/ocm" "github.com/Azure/ARO-HCP/internal/utils" ) @@ -52,8 +54,30 @@ const ( testClusterUID = "00000000-0000-0000-0000-000000000000" // testManagedResourceGroup must match what coreapitesting.MinimumValidClusterTestCase() sets. testManagedResourceGroup = "testManagedResourceGroup" + // testStampIdentifier / testProvisionShardID drive the management-cluster + // placement + provision-shard-pinning fixtures. + testStampIdentifier = "1" + testProvisionShardID = "shard-abc123" ) +// testManagementClusterResourceID returns the resource ID of the placed management cluster. +func testManagementClusterResourceID() *azcorearm.ResourceID { + return metadataapi.Must(fleetapi.ToManagementClusterResourceID(testStampIdentifier)) +} + +// newTestManagementCluster returns a management cluster carrying the CS provision +// shard used by the provision-shard-pinning tests. +func newTestManagementCluster() *fleetapi.ManagementCluster { + resourceID := testManagementClusterResourceID() + return &fleetapi.ManagementCluster{ + CosmosMetadata: coreapi.CosmosMetadata{ResourceID: resourceID, PartitionKey: testStampIdentifier}, + ResourceID: resourceID, + Status: fleetapi.ManagementClusterStatus{ + ClusterServiceProvisionShardID: ptr.To(metadataapi.Must(metadataapi.NewInternalID("/api/aro_hcp/v1alpha1/provision_shards/" + testProvisionShardID))), + }, + } +} + // testClusterResourceID builds the ARM resource ID for the test cluster. func testClusterResourceID() *azcorearm.ResourceID { return metadataapi.Must(azcorearm.ParseResourceID( @@ -128,6 +152,7 @@ func TestClusterClusterServiceCreate_SyncOnce(t *testing.T) { listCluster *coreapi.HCPOpenShiftCluster // cluster seeded into the lister (nil = not found) dbCluster *coreapi.HCPOpenShiftCluster // cluster stored in the DB existingServiceProviderCluster *coreapi.ServiceProviderCluster // nil = not pre-seeded; controller get-or-creates + managementClusters []*fleetapi.ManagementCluster // seeded into the fleet lister setupMockCS func(ctrl *gomock.Controller) ocm.ClusterServiceClientSpec expectError bool verifyDB func(t *testing.T, ctx context.Context, db *corecosmosstoragetesting.MockResourcesDBClient) @@ -142,7 +167,9 @@ func TestClusterClusterServiceCreate_SyncOnce(t *testing.T) { }), existingServiceProviderCluster: newTestSPC(func(spc *coreapi.ServiceProviderCluster) { spc.Spec.ControlPlaneVersion.DesiredVersion = desiredVersion + spc.Spec.ManagementClusterResourceID = testManagementClusterResourceID() }), + managementClusters: []*fleetapi.ManagementCluster{newTestManagementCluster()}, setupMockCS: func(ctrl *gomock.Controller) ocm.ClusterServiceClientSpec { mockCS := ocm.NewMockClusterServiceClientSpec(ctrl) mockCS.EXPECT(). @@ -154,6 +181,7 @@ func TestClusterClusterServiceCreate_SyncOnce(t *testing.T) { built, buildErr := builder.Build() require.NoError(t, buildErr) assert.Equal(t, pendingClusterServiceID.ID(), built.ID(), "PostCluster should use the final segment of PendingClusterServiceID") + assert.Equal(t, testProvisionShardID, built.ProvisionShardID(), "PostCluster should pin the provision shard from the placed management cluster") csCluster, err := arohcpv1alpha1.NewCluster(). ID(pendingClusterServiceID.ID()). HREF(testClusterServiceIDStr). @@ -222,6 +250,38 @@ func TestClusterClusterServiceCreate_SyncOnce(t *testing.T) { assert.Nil(t, cluster.ServiceProviderProperties.ClusterServiceID) }, }, + { + name: "defer creation when placement intent (Spec.ManagementClusterResourceID) is not resolved", + listCluster: newTestCluster(func(c *coreapi.HCPOpenShiftCluster) { + c.ServiceProviderProperties.PendingClusterServiceID = &pendingClusterServiceID + }), + dbCluster: newTestCluster(func(c *coreapi.HCPOpenShiftCluster) { + c.ServiceProviderProperties.PendingClusterServiceID = &pendingClusterServiceID + }), + existingServiceProviderCluster: newTestSPC(func(spc *coreapi.ServiceProviderCluster) { + spc.Spec.ControlPlaneVersion.DesiredVersion = desiredVersion + // Spec.ManagementClusterResourceID intentionally left nil: placement not resolved. + }), + setupMockCS: func(ctrl *gomock.Controller) ocm.ClusterServiceClientSpec { + mockCS := ocm.NewMockClusterServiceClientSpec(ctrl) + // findAROHCPClusterByAzureInfo still runs (before the placement gate) + // and finds no existing CS cluster; PostCluster must NOT be called + // because the gate returns nil before creation. + mockCS.EXPECT(). + ListClusters(gomock.Any()). + Return(ocm.NewSimpleClusterListIterator(nil, nil)) + return mockCS + }, + expectError: false, + verifyDB: func(t *testing.T, ctx context.Context, db *corecosmosstoragetesting.MockResourcesDBClient) { + cluster, err := db.HCPClusters(testSubscriptionID, testResourceGroupName).Get(ctx, testClusterName) + require.NoError(t, err) + // No CS cluster created: ClusterServiceID stays nil and the pending + // ID is preserved for the next attempt once placement lands. + assert.Nil(t, cluster.ServiceProviderProperties.ClusterServiceID) + assert.NotNil(t, cluster.ServiceProviderProperties.PendingClusterServiceID) + }, + }, { name: "adopts existing Cluster Service cluster for Azure resource", listCluster: newTestCluster(func(c *coreapi.HCPOpenShiftCluster) { @@ -232,7 +292,9 @@ func TestClusterClusterServiceCreate_SyncOnce(t *testing.T) { }), existingServiceProviderCluster: newTestSPC(func(spc *coreapi.ServiceProviderCluster) { spc.Spec.ControlPlaneVersion.DesiredVersion = desiredVersion + spc.Spec.ManagementClusterResourceID = testManagementClusterResourceID() }), + managementClusters: []*fleetapi.ManagementCluster{newTestManagementCluster()}, setupMockCS: func(ctrl *gomock.Controller) ocm.ClusterServiceClientSpec { mockCS := ocm.NewMockClusterServiceClientSpec(ctrl) // Build the CS cluster with Azure fields matching the test cluster so it @@ -284,10 +346,11 @@ func TestClusterClusterServiceCreate_SyncOnce(t *testing.T) { listerClusters = []*coreapi.HCPOpenShiftCluster{tt.listCluster} } syncer := &clusterClusterServiceCreateSyncer{ - resourcesDBClient: mockDB, - clusterLister: &corelistertesting.SliceClusterLister{Clusters: listerClusters}, - subscriptionLister: &corelistertesting.SliceSubscriptionLister{Subscriptions: []*coreapi.Subscription{subscription}}, - clustersServiceClient: mockCS, + resourcesDBClient: mockDB, + clusterLister: &corelistertesting.SliceClusterLister{Clusters: listerClusters}, + subscriptionLister: &corelistertesting.SliceSubscriptionLister{Subscriptions: []*coreapi.Subscription{subscription}}, + managementClusterLister: &fleetlistertesting.SliceManagementClusterLister{ManagementClusters: tt.managementClusters}, + clustersServiceClient: mockCS, } key := controllerutils.HCPClusterKey{ diff --git a/backend/pkg/controllers/cluster/creation/cluster_pending_cluster_service_id_assign_controller.go b/backend/pkg/controllers/cluster/creation/cluster_pending_cluster_service_id_assign_controller.go index 13389a7d726..80dfbf6e228 100644 --- a/backend/pkg/controllers/cluster/creation/cluster_pending_cluster_service_id_assign_controller.go +++ b/backend/pkg/controllers/cluster/creation/cluster_pending_cluster_service_id_assign_controller.go @@ -31,8 +31,9 @@ import ( ) type clusterPendingClusterServiceIDAssignSyncer struct { - clusterLister corelisters.ClusterLister - resourcesDBClient corecosmosstorage.ResourcesDBClient + clusterLister corelisters.ClusterLister + serviceProviderClusterLister corelisters.ServiceProviderClusterLister + resourcesDBClient corecosmosstorage.ResourcesDBClient } var _ controllerutils.ClusterSyncer = (*clusterPendingClusterServiceIDAssignSyncer)(nil) @@ -41,9 +42,11 @@ const ClusterPendingClusterServiceIDAssignControllerName = "ClusterPendingCluste func NewClusterPendingClusterServiceIDAssignController(resourcesDBClient corecosmosstorage.ResourcesDBClient, backendInformers coreinformers.BackendInformers) controllerutils.Controller { _, clusterLister := backendInformers.Clusters() + _, serviceProviderClusterLister := backendInformers.ServiceProviderClusters() syncer := &clusterPendingClusterServiceIDAssignSyncer{ - clusterLister: clusterLister, - resourcesDBClient: resourcesDBClient, + clusterLister: clusterLister, + serviceProviderClusterLister: serviceProviderClusterLister, + resourcesDBClient: resourcesDBClient, } return controllerutils.NewClusterWatchingController( @@ -56,11 +59,19 @@ func NewClusterPendingClusterServiceIDAssignController(resourcesDBClient corecos ) } -func (c *clusterPendingClusterServiceIDAssignSyncer) needsWork(cluster *coreapi.HCPOpenShiftCluster) bool { +// needsWork reports whether a PendingClusterServiceID should be assigned. In +// addition to the cluster not yet having a (pending or resolved) Cluster Service +// ID and not being deleted, placement must already be resolved: the +// ServiceProviderCluster must have Spec.ManagementClusterResourceID set by the +// PlacementController. This gates Cluster Service creation on a management +// cluster having been chosen first. +func (c *clusterPendingClusterServiceIDAssignSyncer) needsWork(cluster *coreapi.HCPOpenShiftCluster, serviceProviderCluster *coreapi.ServiceProviderCluster) bool { return cluster.ServiceProviderProperties.DeletionTimestamp == nil && cluster.ServiceProviderProperties.PendingClusterServiceID == nil && (cluster.ServiceProviderProperties.ClusterServiceID == nil || - len(cluster.ServiceProviderProperties.ClusterServiceID.String()) == 0) + len(cluster.ServiceProviderProperties.ClusterServiceID.String()) == 0) && + serviceProviderCluster != nil && + serviceProviderCluster.Spec.ManagementClusterResourceID != nil } func (c *clusterPendingClusterServiceIDAssignSyncer) SyncOnce(ctx context.Context, key controllerutils.HCPClusterKey) error { @@ -74,7 +85,16 @@ func (c *clusterPendingClusterServiceIDAssignSyncer) SyncOnce(ctx context.Contex return utils.TrackError(err) } - if !c.needsWork(cluster) { + serviceProviderCluster, err := c.serviceProviderClusterLister.Get(ctx, key.SubscriptionID, key.ResourceGroupName, key.HCPClusterName) + if cosmosstorageutils.IsNotFoundError(err) { + // Placement has not produced a ServiceProviderCluster yet; wait. + return nil + } + if err != nil { + return utils.TrackError(err) + } + + if !c.needsWork(cluster, serviceProviderCluster) { return nil } diff --git a/backend/pkg/controllers/cluster/creation/cluster_pending_cluster_service_id_assign_controller_test.go b/backend/pkg/controllers/cluster/creation/cluster_pending_cluster_service_id_assign_controller_test.go index fb286e15d80..aaa97214a17 100644 --- a/backend/pkg/controllers/cluster/creation/cluster_pending_cluster_service_id_assign_controller_test.go +++ b/backend/pkg/controllers/cluster/creation/cluster_pending_cluster_service_id_assign_controller_test.go @@ -36,16 +36,26 @@ import ( func TestClusterPendingClusterServiceIDAssign_SyncOnce(t *testing.T) { clusterInternalID := metadataapi.Must(metadataapi.NewInternalID(testClusterServiceIDStr)) + // placedSPC is a ServiceProviderCluster whose Spec.ManagementClusterResourceID + // is set (placement resolved by the PlacementController). + placedSPC := func() *coreapi.ServiceProviderCluster { + return newTestSPC(func(spc *coreapi.ServiceProviderCluster) { + spc.Spec.ManagementClusterResourceID = testManagementClusterResourceID() + }) + } + tests := []struct { name string listCluster *coreapi.HCPOpenShiftCluster + listSPC *coreapi.ServiceProviderCluster // seeded into the SPC lister (nil = not found) dbCluster *coreapi.HCPOpenShiftCluster expectError bool verifyDB func(t *testing.T, ctx context.Context, db *corecosmosstoragetesting.MockResourcesDBClient) }{ { - name: "assigns PendingClusterServiceID when both IDs are nil", + name: "assigns PendingClusterServiceID when placement resolved and both IDs nil", listCluster: newTestCluster(), + listSPC: placedSPC(), dbCluster: newTestCluster(), expectError: false, verifyDB: func(t *testing.T, ctx context.Context, db *corecosmosstoragetesting.MockResourcesDBClient) { @@ -56,11 +66,36 @@ func TestClusterPendingClusterServiceIDAssign_SyncOnce(t *testing.T) { assert.Len(t, cluster.ServiceProviderProperties.PendingClusterServiceID.ID(), 32) }, }, + { + name: "skip when placement not resolved (Spec.ManagementClusterResourceID nil)", + listCluster: newTestCluster(), + listSPC: newTestSPC(), // Spec.ManagementClusterResourceID nil + dbCluster: newTestCluster(), + expectError: false, + verifyDB: func(t *testing.T, ctx context.Context, db *corecosmosstoragetesting.MockResourcesDBClient) { + cluster, err := db.HCPClusters(testSubscriptionID, testResourceGroupName).Get(ctx, testClusterName) + require.NoError(t, err) + assert.Nil(t, cluster.ServiceProviderProperties.PendingClusterServiceID) + }, + }, + { + name: "skip when ServiceProviderCluster not found", + listCluster: newTestCluster(), + listSPC: nil, + dbCluster: newTestCluster(), + expectError: false, + verifyDB: func(t *testing.T, ctx context.Context, db *corecosmosstoragetesting.MockResourcesDBClient) { + cluster, err := db.HCPClusters(testSubscriptionID, testResourceGroupName).Get(ctx, testClusterName) + require.NoError(t, err) + assert.Nil(t, cluster.ServiceProviderProperties.PendingClusterServiceID) + }, + }, { name: "skip when PendingClusterServiceID already set", listCluster: newTestCluster(func(c *coreapi.HCPOpenShiftCluster) { c.ServiceProviderProperties.PendingClusterServiceID = &clusterInternalID }), + listSPC: placedSPC(), dbCluster: newTestCluster(func(c *coreapi.HCPOpenShiftCluster) { c.ServiceProviderProperties.PendingClusterServiceID = &clusterInternalID }), @@ -77,6 +112,7 @@ func TestClusterPendingClusterServiceIDAssign_SyncOnce(t *testing.T) { listCluster: newTestCluster(func(c *coreapi.HCPOpenShiftCluster) { c.ServiceProviderProperties.ClusterServiceID = &clusterInternalID }), + listSPC: placedSPC(), dbCluster: newTestCluster(func(c *coreapi.HCPOpenShiftCluster) { c.ServiceProviderProperties.ClusterServiceID = &clusterInternalID }), @@ -93,6 +129,7 @@ func TestClusterPendingClusterServiceIDAssign_SyncOnce(t *testing.T) { now := metav1.Now() c.ServiceProviderProperties.DeletionTimestamp = &now }), + listSPC: placedSPC(), dbCluster: newTestCluster(func(c *coreapi.HCPOpenShiftCluster) { now := metav1.Now() c.ServiceProviderProperties.DeletionTimestamp = &now @@ -107,6 +144,7 @@ func TestClusterPendingClusterServiceIDAssign_SyncOnce(t *testing.T) { { name: "skip when cluster not found in lister", listCluster: nil, + listSPC: placedSPC(), dbCluster: newTestCluster(), expectError: false, verifyDB: func(t *testing.T, ctx context.Context, db *corecosmosstoragetesting.MockResourcesDBClient) { @@ -129,9 +167,14 @@ func TestClusterPendingClusterServiceIDAssign_SyncOnce(t *testing.T) { if tt.listCluster != nil { listerClusters = []*coreapi.HCPOpenShiftCluster{tt.listCluster} } + var listerSPCs []*coreapi.ServiceProviderCluster + if tt.listSPC != nil { + listerSPCs = []*coreapi.ServiceProviderCluster{tt.listSPC} + } syncer := &clusterPendingClusterServiceIDAssignSyncer{ - resourcesDBClient: mockDB, - clusterLister: &corelistertesting.SliceClusterLister{Clusters: listerClusters}, + resourcesDBClient: mockDB, + clusterLister: &corelistertesting.SliceClusterLister{Clusters: listerClusters}, + serviceProviderClusterLister: &corelistertesting.SliceServiceProviderClusterLister{ServiceProviderClusters: listerSPCs}, } key := controllerutils.HCPClusterKey{ diff --git a/backend/pkg/controllers/cluster/placement/pending_cleanup_controller.go b/backend/pkg/controllers/cluster/placement/pending_cleanup_controller.go new file mode 100644 index 00000000000..2efc4c4b20a --- /dev/null +++ b/backend/pkg/controllers/cluster/placement/pending_cleanup_controller.go @@ -0,0 +1,174 @@ +// Copyright 2026 Microsoft Corporation +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package placement + +import ( + "context" + "fmt" + "strings" + "time" + + azcorearm "github.com/Azure/azure-sdk-for-go/sdk/azcore/arm" + + "github.com/Azure/ARO-HCP/backend/pkg/utils/controllerutils" + controllerutil "github.com/Azure/ARO-HCP/internal/controllerutils" + "github.com/Azure/ARO-HCP/internal/database/cosmosstorage/cosmosstorageutils" + "github.com/Azure/ARO-HCP/internal/database/cosmosstorage/fleetcosmosstorage" + "github.com/Azure/ARO-HCP/internal/database/informers/fleetinformers" + "github.com/Azure/ARO-HCP/internal/database/listers/corelisters" + "github.com/Azure/ARO-HCP/internal/database/listers/fleetlisters" + "github.com/Azure/ARO-HCP/internal/utils" +) + +// PendingCleanupControllerName is the single logical name for this controller. +const PendingCleanupControllerName = "PendingCleanup" + +const pendingCleanupResyncPeriod = 10 * time.Minute + +// pendingCleanupSyncer periodically sweeps each management cluster's +// ManagementClusterScheduling.Status.PendingAssignedClusters and removes stale +// reservations. For each entry it determines the referenced cluster's effective +// placement — the observed Status.ManagementClusterResourceID when set (Cluster +// Service reality), otherwise the Spec intent — and: +// - keeps it when the effective placement points at this management cluster; +// - keeps it when the effective placement is nil (placement still in progress); +// - removes it when the effective placement points at a different management +// cluster, or when the ServiceProviderCluster no longer exists. +// +// CapacityReportingController removes reservations once the HCP is observed +// (Ready/NotReady); this controller handles the reservations that never get +// observed (e.g. placement retried onto a different management cluster, or the +// cluster was deleted before it showed up). +type pendingCleanupSyncer struct { + serviceProviderClusterLister corelisters.ServiceProviderClusterLister + managementClusterSchedulingLister fleetlisters.ManagementClusterSchedulingLister + fleetDBClient fleetcosmosstorage.FleetDBClient +} + +var _ controllerutils.ManagementClusterSyncer = (*pendingCleanupSyncer)(nil) + +// NewPendingCleanupController creates the management-cluster-keyed controller +// that garbage-collects stale PendingAssignedClusters reservations. +func NewPendingCleanupController( + fleetDBClient fleetcosmosstorage.FleetDBClient, + serviceProviderClusterLister corelisters.ServiceProviderClusterLister, + fleetInformers fleetinformers.FleetInformers, +) controllerutils.Controller { + _, managementClusterSchedulingLister := fleetInformers.ManagementClusterSchedulings() + syncer := &pendingCleanupSyncer{ + serviceProviderClusterLister: serviceProviderClusterLister, + managementClusterSchedulingLister: managementClusterSchedulingLister, + fleetDBClient: fleetDBClient, + } + + return controllerutils.NewManagementClusterWatchingController( + PendingCleanupControllerName, + fleetDBClient, + fleetInformers, + pendingCleanupResyncPeriod, + syncer, + ) +} + +// CooldownChecker returns nil: the resync period governs the sweep cadence. +func (c *pendingCleanupSyncer) CooldownChecker() controllerutil.CooldownChecker { + return nil +} + +// SyncOnce sweeps one management cluster's PendingAssignedClusters list. On a +// transient failure it returns an error so the workqueue retries with backoff. +func (c *pendingCleanupSyncer) SyncOnce(ctx context.Context, key controllerutils.ManagementClusterKey) error { + logger := utils.LoggerFromContext(ctx) + + managementClusterResourceID := key.GetResourceID() + + // Read the scheduling document from the informer cache (not a live CRUD Get). + // The cached copy carries the etag that guards the optimistic Replace below, so + // a stale cache can only produce a write conflict (412) that re-enqueues the + // key — never a lost update. + existing, err := c.managementClusterSchedulingLister.Get(ctx, key.StampIdentifier) + if cosmosstorageutils.IsNotFoundError(err) { + return nil + } + if err != nil { + return utils.TrackError(fmt.Errorf("failed to get scheduling document for management cluster %q from cache: %w", key.StampIdentifier, err)) + } + if len(existing.Status.PendingAssignedClusters) == 0 { + return nil + } + + kept := make([]*azcorearm.ResourceID, 0, len(existing.Status.PendingAssignedClusters)) + removed := 0 + for _, pending := range existing.Status.PendingAssignedClusters { + keep, err := c.shouldKeepPending(ctx, pending, managementClusterResourceID) + if err != nil { + return err + } + if keep { + kept = append(kept, pending) + } else { + removed++ + } + } + if removed == 0 { + return nil + } + + updated := existing.DeepCopy() + if len(kept) == 0 { + updated.Status.PendingAssignedClusters = nil + } else { + updated.Status.PendingAssignedClusters = kept + } + + // The write path stays a live, etag-guarded Replace via the CRUD client; the + // base document (and its etag) came from the cache read above. + schedulingCRUD := c.fleetDBClient.Stamps().ManagementClusters(key.StampIdentifier).Scheduling() + if _, err := schedulingCRUD.Replace(ctx, updated, nil); err != nil { + return utils.TrackError(fmt.Errorf("failed to update scheduling document for management cluster %q: %w", key.StampIdentifier, err)) + } + logger.Info("cleaned up stale pending assignments", "removed", removed, "remaining", len(kept)) + return nil +} + +// shouldKeepPending decides whether a single pending reservation entry (a +// cluster ARM resource ID) should be retained on managementClusterResourceID. +func (c *pendingCleanupSyncer) shouldKeepPending(ctx context.Context, pending, managementClusterResourceID *azcorearm.ResourceID) (bool, error) { + if pending == nil { + return false, nil + } + + serviceProviderCluster, err := c.serviceProviderClusterLister.Get(ctx, pending.SubscriptionID, pending.ResourceGroupName, pending.Name) + if cosmosstorageutils.IsNotFoundError(err) { + // The ServiceProviderCluster no longer exists: drop the reservation. + return false, nil + } + if err != nil { + return false, utils.TrackError(fmt.Errorf("failed to get ServiceProviderCluster for pending assignment %q: %w", pending.String(), err)) + } + + // The observed placement (Status) is Cluster Service reality and takes + // precedence; fall back to the scheduler intent (Spec) when Status is unset. + effectivePlacement := serviceProviderCluster.Status.ManagementClusterResourceID + if effectivePlacement == nil { + effectivePlacement = serviceProviderCluster.Spec.ManagementClusterResourceID + } + if effectivePlacement == nil { + // Placement still in progress: keep the reservation so capacity stays held. + return true, nil + } + // Keep only when the effective placement still points at this management cluster. + return strings.EqualFold(effectivePlacement.String(), managementClusterResourceID.String()), nil +} diff --git a/backend/pkg/controllers/cluster/placement/pending_cleanup_controller_test.go b/backend/pkg/controllers/cluster/placement/pending_cleanup_controller_test.go new file mode 100644 index 00000000000..26550f8eafc --- /dev/null +++ b/backend/pkg/controllers/cluster/placement/pending_cleanup_controller_test.go @@ -0,0 +1,158 @@ +// Copyright 2026 Microsoft Corporation +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package placement + +import ( + "context" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + azcorearm "github.com/Azure/azure-sdk-for-go/sdk/azcore/arm" + + "github.com/Azure/ARO-HCP/backend/pkg/utils/controllerutils" + "github.com/Azure/ARO-HCP/internal/api/coreapi" + "github.com/Azure/ARO-HCP/internal/api/fleetapi" + "github.com/Azure/ARO-HCP/internal/api/metadataapi" + "github.com/Azure/ARO-HCP/internal/database/cosmosstoragetesting/fleetcosmosstoragetesting" + "github.com/Azure/ARO-HCP/internal/database/listertesting/corelistertesting" + "github.com/Azure/ARO-HCP/internal/database/listertesting/fleetlistertesting" +) + +func pendingClusterResourceID(name string) *azcorearm.ResourceID { + return metadataapi.Must(azcorearm.ParseResourceID( + "/subscriptions/" + testClusterSubscriptionID + + "/resourceGroups/" + testClusterResourceGroup + + "/providers/Microsoft.RedHatOpenShift/hcpOpenShiftClusters/" + name)) +} + +// spcForCluster builds a ServiceProviderCluster for a cluster name with the given +// Spec (intent) and Status (observed) placements. Pass nil for either to leave it unset. +func spcForCluster(name string, specPlacement, statusPlacement *azcorearm.ResourceID) *coreapi.ServiceProviderCluster { + clusterRID := pendingClusterResourceID(name) + spcRID := metadataapi.Must(azcorearm.ParseResourceID( + clusterRID.String() + "/" + coreapi.ServiceProviderClusterResourceTypeName + "/" + coreapi.ServiceProviderClusterResourceName)) + spc := &coreapi.ServiceProviderCluster{ + CosmosMetadata: coreapi.CosmosMetadata{ResourceID: spcRID, PartitionKey: strings.ToLower(testClusterSubscriptionID)}, + } + if specPlacement != nil { + spc.Spec.ManagementClusterResourceID = specPlacement + } + if statusPlacement != nil { + spc.Status.ManagementClusterResourceID = statusPlacement + } + return spc +} + +func pendingStrings(ids []*azcorearm.ResourceID) []string { + out := make([]string, 0, len(ids)) + for _, id := range ids { + out = append(out, strings.ToLower(id.String())) + } + return out +} + +func TestPendingCleanupSyncer_SyncOnce(t *testing.T) { + ctx := context.Background() + + const thisStamp = "1" + thisMC := metadataapi.Must(fleetapi.ToManagementClusterResourceID(thisStamp)) + otherMC := metadataapi.Must(fleetapi.ToManagementClusterResourceID("2")) + + // Effective placement = Status (CS reality) when set, else Spec. + // a: Status here, Spec nil => keep + // b: Status other, Spec here (Status wins) => remove + // c: Status nil, Spec here (Spec fallback) => keep + // d: Status nil, Spec nil (in progress) => keep + // e: Status other => remove + // f: SPC missing => remove + serviceProviderClusters := []*coreapi.ServiceProviderCluster{ + spcForCluster("a", nil, thisMC), + spcForCluster("b", thisMC, otherMC), + spcForCluster("c", thisMC, nil), + spcForCluster("d", nil, nil), + spcForCluster("e", nil, otherMC), + } + spcLister := &corelistertesting.SliceServiceProviderClusterLister{ServiceProviderClusters: serviceProviderClusters} + + pending := []*azcorearm.ResourceID{ + pendingClusterResourceID("a"), + pendingClusterResourceID("b"), + pendingClusterResourceID("c"), + pendingClusterResourceID("d"), + pendingClusterResourceID("e"), + pendingClusterResourceID("f"), + } + fleetDB := fleetcosmosstoragetesting.NewMockFleetDBClient() + doc := &fleetapi.ManagementClusterScheduling{ + CosmosMetadata: coreapi.CosmosMetadata{ + ResourceID: metadataapi.Must(fleetapi.ToManagementClusterSchedulingResourceID(thisStamp)), + PartitionKey: thisStamp, + }, + Status: fleetapi.ManagementClusterSchedulingStatus{PendingAssignedClusters: pending}, + } + created, err := fleetDB.Stamps().ManagementClusters(thisStamp).Scheduling().Create(ctx, doc, nil) + require.NoError(t, err) + + // The syncer reads the scheduling document from the informer-cache lister + // (seeded with the created doc so its etag matches the fleet DB for the + // Replace write path). + schedulingLister := &fleetlistertesting.SliceManagementClusterSchedulingLister{Schedulings: []*fleetapi.ManagementClusterScheduling{created}} + syncer := &pendingCleanupSyncer{serviceProviderClusterLister: spcLister, managementClusterSchedulingLister: schedulingLister, fleetDBClient: fleetDB} + require.NoError(t, syncer.SyncOnce(ctx, controllerutils.ManagementClusterKey{StampIdentifier: thisStamp})) + + updated, err := fleetDB.Stamps().ManagementClusters(thisStamp).Scheduling().Get(ctx, fleetapi.SchedulingResourceName) + require.NoError(t, err) + + kept := pendingStrings(updated.Status.PendingAssignedClusters) + assert.ElementsMatch(t, []string{ + strings.ToLower(pendingClusterResourceID("a").String()), + strings.ToLower(pendingClusterResourceID("c").String()), + strings.ToLower(pendingClusterResourceID("d").String()), + }, kept, "keep entries whose effective placement (Status first, else Spec) points here or is still nil") +} + +func TestPendingCleanupSyncer_SyncOnce_NoChangeWhenAllValid(t *testing.T) { + ctx := context.Background() + + const thisStamp = "1" + thisMC := metadataapi.Must(fleetapi.ToManagementClusterResourceID(thisStamp)) + + spcLister := &corelistertesting.SliceServiceProviderClusterLister{ + ServiceProviderClusters: []*coreapi.ServiceProviderCluster{spcForCluster("a", nil, thisMC)}, + } + + fleetDB := fleetcosmosstoragetesting.NewMockFleetDBClient() + doc := &fleetapi.ManagementClusterScheduling{ + CosmosMetadata: coreapi.CosmosMetadata{ + ResourceID: metadataapi.Must(fleetapi.ToManagementClusterSchedulingResourceID(thisStamp)), + PartitionKey: thisStamp, + }, + Status: fleetapi.ManagementClusterSchedulingStatus{PendingAssignedClusters: []*azcorearm.ResourceID{pendingClusterResourceID("a")}}, + } + created, err := fleetDB.Stamps().ManagementClusters(thisStamp).Scheduling().Create(ctx, doc, nil) + require.NoError(t, err) + + schedulingLister := &fleetlistertesting.SliceManagementClusterSchedulingLister{Schedulings: []*fleetapi.ManagementClusterScheduling{created}} + syncer := &pendingCleanupSyncer{serviceProviderClusterLister: spcLister, managementClusterSchedulingLister: schedulingLister, fleetDBClient: fleetDB} + require.NoError(t, syncer.SyncOnce(ctx, controllerutils.ManagementClusterKey{StampIdentifier: thisStamp})) + + updated, err := fleetDB.Stamps().ManagementClusters(thisStamp).Scheduling().Get(ctx, fleetapi.SchedulingResourceName) + require.NoError(t, err) + require.Len(t, updated.Status.PendingAssignedClusters, 1) + assert.Equal(t, strings.ToLower(pendingClusterResourceID("a").String()), strings.ToLower(updated.Status.PendingAssignedClusters[0].String())) +} diff --git a/backend/pkg/controllers/cluster/placement/placement_controller.go b/backend/pkg/controllers/cluster/placement/placement_controller.go new file mode 100644 index 00000000000..5ee668bf6e4 --- /dev/null +++ b/backend/pkg/controllers/cluster/placement/placement_controller.go @@ -0,0 +1,548 @@ +// Copyright 2026 Microsoft Corporation +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package placement + +import ( + "context" + "errors" + "fmt" + "net/http" + "strings" + "time" + + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + azcorearm "github.com/Azure/azure-sdk-for-go/sdk/azcore/arm" + + ocmerrors "github.com/openshift-online/ocm-sdk-go/errors" + + "github.com/Azure/ARO-HCP/backend/pkg/utils/controllerutils" + "github.com/Azure/ARO-HCP/internal/api/coreapi" + "github.com/Azure/ARO-HCP/internal/api/fleetapi" + "github.com/Azure/ARO-HCP/internal/api/metadataapi" + "github.com/Azure/ARO-HCP/internal/database/cosmosstorage/corecosmosstorage" + "github.com/Azure/ARO-HCP/internal/database/cosmosstorage/cosmosstorageutils" + "github.com/Azure/ARO-HCP/internal/database/cosmosstorage/fleetcosmosstorage" + "github.com/Azure/ARO-HCP/internal/database/informers/coreinformers" + "github.com/Azure/ARO-HCP/internal/database/listers/corelisters" + "github.com/Azure/ARO-HCP/internal/database/listers/fleetlisters" + unionkubeapplierinformers "github.com/Azure/ARO-HCP/internal/database/unioninformers/kubeapplier" + "github.com/Azure/ARO-HCP/internal/kuberesources" + "github.com/Azure/ARO-HCP/internal/ocm" + "github.com/Azure/ARO-HCP/internal/utils" +) + +// PlacementControllerName is the single logical name for this controller. It is +// used for the workqueue name (a Prometheus label), the context controller name, +// and log values so metrics, ctx, and log fields never drift. +const PlacementControllerName = "Placement" + +// swiftNICsPerHCP is the number of SWIFT NICs a single HostedControlPlane +// consumes. Every HCP needs exactly 3 (except clusters created with the 2024 +// API version, which use fewer — but that version is being removed). Using 3 as +// a flat per-HCP cost is a conservative approximation that never overbooks a +// management cluster's swift-NIC capacity. +const swiftNICsPerHCP int64 = 3 + +// placementSyncer selects the management cluster a newly-created HCP should be +// scheduled onto and records that intent on ServiceProviderCluster.Spec. +// ManagementClusterResourceID. Status.ManagementClusterResourceID (the observed +// placement) continues to be written by ManagementClusterPlacementSync. +type placementSyncer struct { + serviceProviderClusterLister corelisters.ServiceProviderClusterLister + clusterLister corelisters.ClusterLister + managementClusterLister fleetlisters.ManagementClusterLister + managementClusterSchedulingLister fleetlisters.ManagementClusterSchedulingLister + cosmosClient corecosmosstorage.ResourcesDBClient + fleetDBClient fleetcosmosstorage.FleetDBClient + clusterServiceClient ocm.ClusterServiceClientSpec +} + +var _ controllerutils.ClusterSyncer = (*placementSyncer)(nil) + +// NewPlacementController creates the scheduling controller that resolves initial +// placement for a HostedControlPlane by choosing an eligible management cluster +// with sufficient swift-NIC capacity and writing it to +// ServiceProviderCluster.Spec.ManagementClusterResourceID. +func NewPlacementController( + cosmosClient corecosmosstorage.ResourcesDBClient, + fleetDBClient fleetcosmosstorage.FleetDBClient, + clusterServiceClient ocm.ClusterServiceClientSpec, + managementClusterLister fleetlisters.ManagementClusterLister, + managementClusterSchedulingLister fleetlisters.ManagementClusterSchedulingLister, + informers coreinformers.BackendInformers, + kubeApplierInformers *unionkubeapplierinformers.UnionKubeApplierInformers, +) controllerutils.Controller { + _, serviceProviderClusterLister := informers.ServiceProviderClusters() + _, clusterLister := informers.Clusters() + + syncer := &placementSyncer{ + serviceProviderClusterLister: serviceProviderClusterLister, + clusterLister: clusterLister, + managementClusterLister: managementClusterLister, + managementClusterSchedulingLister: managementClusterSchedulingLister, + cosmosClient: cosmosClient, + fleetDBClient: fleetDBClient, + clusterServiceClient: clusterServiceClient, + } + + return controllerutils.NewClusterWatchingController( + PlacementControllerName, + cosmosClient, + informers, + kubeApplierInformers, + 5*time.Minute, // Check every 5 minutes + syncer, + ) +} + +// needsWork reports whether the ServiceProviderCluster still needs its +// Spec.ManagementClusterResourceID (scheduler intent) resolved. There is work +// whenever Spec is nil: either a fresh capacity-aware selection (Spec and Status +// both nil) or a rollout backfill from the observed Status placement (Spec nil, +// Status set). +func (c *placementSyncer) needsWork(serviceProviderCluster *coreapi.ServiceProviderCluster) bool { + return serviceProviderCluster.Spec.ManagementClusterResourceID == nil +} + +// SyncOnce resolves placement for a single HCP cluster and records it on +// ServiceProviderCluster.Spec.ManagementClusterResourceID. +func (c *placementSyncer) SyncOnce(ctx context.Context, key controllerutils.HCPClusterKey) error { + logger := utils.LoggerFromContext(ctx) + + // Reads come from the informer caches; every Cosmos write is an optimistic, + // etag-guarded Replace. The ServiceProviderCluster is read from its lister + // (here for the needsWork gate and again in setSpecPlacement as the + // etag-carrying base of the Replace) and the per-management-cluster scheduling + // documents are read from their lister for capacity scoring. The write path is + // the only place that reads live: reservePendingAssignment read-modify-writes + // the scheduling document via the CRUD client. A stale cached read can only + // lose the race with a 412 conflict that re-enqueues the key — never a lost + // update — and capacity reserved against a now-obsolete decision is reclaimed + // by the PendingCleanupController. + serviceProviderCluster, err := c.serviceProviderClusterLister.Get(ctx, key.SubscriptionID, key.ResourceGroupName, key.HCPClusterName) + if cosmosstorageutils.IsNotFoundError(err) { + logger.V(1).Info("ServiceProviderCluster not found in cache, skipping") + return nil + } + if err != nil { + return utils.TrackError(fmt.Errorf("failed to get ServiceProviderCluster from cache: %w", err)) + } + if !c.needsWork(serviceProviderCluster) { + logger.V(1).Info("ServiceProviderCluster already has Spec.ManagementClusterResourceID, skipping") + return nil + } + + // Old records: the HCP was already placed by ManagementClusterPlacementSync + // (Status.ManagementClusterResourceID mirrors the Cluster Service placement) + // before the scheduler-intent Spec field existed, so Spec is nil while Status + // is set. Backfill Spec from the observed Status placement rather than + // fresh-scheduling, so downstream Cluster Service creation adopts the existing + // placement instead of selecting a possibly different management cluster. + if serviceProviderCluster.Status.ManagementClusterResourceID != nil { + if err := c.setSpecPlacement(ctx, key, serviceProviderCluster.Status.ManagementClusterResourceID); err != nil { + return err + } + logger.Info("backfilled management cluster placement intent from observed status", + "managementClusterID", serviceProviderCluster.Status.ManagementClusterResourceID.String()) + return nil + } + + // Rollout race: an old record created by a prior backend version can have a + // Cluster Service ID assigned — and a placement already decided by Cluster + // Service — while both Spec and Status ManagementClusterResourceID are still + // nil (the observed-placement mirror, ManagementClusterPlacementSync, has not + // caught up yet). Fresh-selecting here could pick a different management + // cluster than the one Cluster Service already committed to. Instead, ask + // Cluster Service where it placed the cluster (by the pending CS ID) and + // backfill Spec from that. This targeted live Cluster Service read only runs + // for this migration edge case; new records never reach it because + // PendingClusterServiceID assignment is gated on Spec being resolved first. + if chosen, handled, err := c.backfillFromClusterService(ctx, key); err != nil { + return err + } else if handled { + if chosen == nil { + // A pending CS ID exists but Cluster Service has not reported a placement + // yet: defer rather than fresh-select, to avoid diverging from the + // placement Cluster Service will eventually report. + logger.Info("cluster has a pending Cluster Service ID but Cluster Service has not reported a placement yet; deferring placement") + return nil + } + if err := c.setSpecPlacement(ctx, key, chosen); err != nil { + return err + } + logger.Info("backfilled management cluster placement intent from Cluster Service", "managementClusterID", chosen.String()) + return nil + } + + // Fresh capacity-aware selection: gather candidate management clusters paired + // with their scheduling documents, then let selectByCapacity perform all + // candidate elimination and choose the emptiest eligible one. + candidates, err := c.gatherSchedulingCandidates(ctx) + if err != nil { + return utils.TrackError(fmt.Errorf("failed to gather scheduling candidates for %s: %w", key.HCPClusterName, err)) + } + chosen, err := selectByCapacity(candidates) + if err != nil { + return utils.TrackError(fmt.Errorf("failed to select management cluster for %s: %w", key.HCPClusterName, err)) + } + + // Reserve capacity on the chosen management cluster before recording the + // placement intent, so concurrent decisions do not overbook it. A crash + // between the reservation and the Spec write is safe: the reservation is + // preserved while Spec is nil and a re-run may pick the same or a different + // management cluster (stale reservations are cleaned up later). + clusterResourceID := key.GetResourceID() + if err := c.reservePendingAssignment(ctx, chosen, clusterResourceID); err != nil { + return err + } + if err := c.setSpecPlacement(ctx, key, chosen); err != nil { + return err + } + logger.Info("assigned management cluster placement", "managementClusterID", chosen.String()) + return nil +} + +// backfillFromClusterService handles the rollout-race migration edge case. When +// the cluster document carries a PendingClusterServiceID (a placement already +// decided by a prior backend version / Cluster Service), it resolves the +// Cluster-Service-reported provision shard back to a management cluster resource +// ID. +// +// It returns handled=true when the caller must NOT fresh-select: either the +// placement was resolved (chosen set to the management cluster resource ID) or +// Cluster Service knows the cluster but has not reported a placement yet (chosen +// nil — the caller should defer to avoid diverging from the placement Cluster +// Service will eventually report). +// +// It returns handled=false when the caller SHOULD fresh-select: there is no +// pending CS ID, or Cluster Service returns 404 (not found) for the pending CS +// ID. A 404 means the pending ID never became a real cluster — an older backend +// recorded PendingClusterServiceID and then crashed or lost leadership before +// creating the cluster in Cluster Service — so there is no committed placement to +// preserve and a fresh capacity-aware selection is safe. Every other error is +// transient and is returned so the workqueue retries. +func (c *placementSyncer) backfillFromClusterService(ctx context.Context, key controllerutils.HCPClusterKey) (chosen *azcorearm.ResourceID, handled bool, err error) { + cluster, err := c.clusterLister.Get(ctx, key.SubscriptionID, key.ResourceGroupName, key.HCPClusterName) + if cosmosstorageutils.IsNotFoundError(err) { + return nil, false, nil + } + if err != nil { + return nil, false, utils.TrackError(fmt.Errorf("failed to get cluster from cache: %w", err)) + } + pendingClusterServiceID := cluster.ServiceProviderProperties.PendingClusterServiceID + if pendingClusterServiceID == nil { + return nil, false, nil + } + + chosen, err = c.resolvePlacementFromClusterService(ctx, *pendingClusterServiceID) + if err != nil { + // A 404 from Cluster Service means this pending Cluster Service ID never + // became a real cluster (an older backend recorded PendingClusterServiceID + // then crashed / lost leadership before creating it in Cluster Service). + // There is no committed placement to diverge from, so fall through to a + // fresh capacity-aware selection instead of deferring forever. Any other + // error is transient — propagate it so the workqueue retries. + var ocmError *ocmerrors.Error + if errors.As(err, &ocmError) && ocmError.Status() == http.StatusNotFound { + utils.LoggerFromContext(ctx).Info("pending Cluster Service ID has no cluster in Cluster Service (404); proceeding with fresh placement", + "clusterServiceID", pendingClusterServiceID.String()) + return nil, false, nil + } + return nil, true, err + } + return chosen, true, nil +} + +// resolvePlacementFromClusterService asks Cluster Service where it already placed +// a cluster (by its Cluster Service ID) and maps the reported provision shard +// back to a management cluster resource ID. It returns (nil, nil) when Cluster +// Service has not yet reported a provision shard, or when no known management +// cluster matches it yet — in both cases the caller should retry later rather +// than fresh-select. +func (c *placementSyncer) resolvePlacementFromClusterService(ctx context.Context, clusterServiceID metadataapi.InternalID) (*azcorearm.ResourceID, error) { + csShard, err := c.clusterServiceClient.GetClusterProvisionShard(ctx, clusterServiceID) + if err != nil { + return nil, utils.TrackError(fmt.Errorf("failed to get provision shard from Cluster Service for %q: %w", clusterServiceID.String(), err)) + } + if len(csShard.HREF()) == 0 { + return nil, nil // provision shard not yet allocated by Cluster Service + } + provisionShardID, err := metadataapi.NewInternalID(csShard.HREF()) + if err != nil { + return nil, utils.TrackError(fmt.Errorf("failed to parse provision shard href %q: %w", csShard.HREF(), err)) + } + managementCluster, err := c.managementClusterLister.GetByCSProvisionShardID(ctx, provisionShardID.ID()) + if cosmosstorageutils.IsNotFoundError(err) { + return nil, nil // provision shard not yet mapped to a known management cluster + } + if err != nil { + return nil, utils.TrackError(fmt.Errorf("failed to resolve provision shard %q to a management cluster: %w", provisionShardID.ID(), err)) + } + return managementCluster.ResourceID, nil +} + +// schedulingCandidate pairs a management cluster with its scheduling document. +// scheduling is nil when the management cluster has no stamp identifier or no +// scheduling document in the cache yet; selectByCapacity records that as an +// elimination reason. +type schedulingCandidate struct { + managementCluster *fleetapi.ManagementCluster + scheduling *fleetapi.ManagementClusterScheduling +} + +// gatherSchedulingCandidates lists management clusters and pairs each with its +// scheduling document read from the informer cache. It performs no elimination — +// that is selectByCapacity's job — so every non-nil management cluster is +// returned, with a nil scheduling document when none is cached. +func (c *placementSyncer) gatherSchedulingCandidates(ctx context.Context) ([]schedulingCandidate, error) { + managementClusters, err := c.managementClusterLister.List(ctx) + if err != nil { + return nil, utils.TrackError(fmt.Errorf("failed to list management clusters: %w", err)) + } + + candidates := make([]schedulingCandidate, 0, len(managementClusters)) + for _, managementCluster := range managementClusters { + if managementCluster == nil || managementCluster.ResourceID == nil { + continue + } + candidate := schedulingCandidate{managementCluster: managementCluster} + // The scheduling document is a singleton child fetched by the parent stamp + // identifier. Without a stamp identifier there is nothing to fetch; leave + // scheduling nil and let selectByCapacity report the reason. + if stampIdentifier := managementCluster.GetStampIdentifier(); stampIdentifier != "" { + scheduling, err := c.managementClusterSchedulingLister.Get(ctx, stampIdentifier) + switch { + case cosmosstorageutils.IsNotFoundError(err): + // No capacity data cached yet: leave scheduling nil. + case err != nil: + return nil, utils.TrackError(fmt.Errorf("failed to get scheduling document for management cluster %q from cache: %w", managementCluster.ResourceID.String(), err)) + default: + candidate.scheduling = scheduling + } + } + candidates = append(candidates, candidate) + } + return candidates, nil +} + +// managementClusterCandidate pairs an eligible management cluster's resource ID +// with its computed available swift-NIC capacity. +type managementClusterCandidate struct { + resourceID *azcorearm.ResourceID + available int64 +} + +// selectByCapacity is a pure function that performs ALL candidate elimination and +// capacity-based selection in one place, so the scheduling decision lives in a +// single unit-testable function rather than being split across the reconcile. +// +// From the gathered (management cluster, scheduling document) pairs it eliminates +// ineligible candidates — recording a human-readable reason for each — then, among +// the candidates whose available swift-NIC capacity is at least swiftNICsPerHCP, +// returns the one with the HIGHEST available capacity (spread: place each new HCP +// on the emptiest management cluster so load is distributed evenly rather than +// concentrated). Ties are broken deterministically by the lowest resource ID +// string so the selection is stable. +// +// When nothing fits it returns an error that enumerates why every candidate was +// eliminated, so the decision can be debugged from the error alone. +func selectByCapacity(candidates []schedulingCandidate) (*azcorearm.ResourceID, error) { + var chosen managementClusterCandidate + found := false + var eliminated []string + + for _, candidate := range candidates { + managementCluster := candidate.managementCluster + if managementCluster == nil || managementCluster.ResourceID == nil { + continue + } + id := managementCluster.ResourceID.String() + if reason := ineligibilityReason(managementCluster, candidate.scheduling); reason != "" { + eliminated = append(eliminated, fmt.Sprintf("%s: %s", id, reason)) + continue + } + available := computeAvailableSwiftNICs(candidate.scheduling) + if available < swiftNICsPerHCP { + eliminated = append(eliminated, fmt.Sprintf("%s: insufficient swift-NIC capacity (available %d, need %d)", id, available, swiftNICsPerHCP)) + continue + } + + fit := managementClusterCandidate{resourceID: managementCluster.ResourceID, available: available} + switch { + case !found: + chosen = fit + found = true + case fit.available > chosen.available: + chosen = fit + case fit.available == chosen.available && fit.resourceID.String() < chosen.resourceID.String(): + chosen = fit + } + } + + if !found { + // Only append the elimination reasons when there are any; otherwise the + // message would end with a dangling ": " (e.g. zero candidates, or every + // candidate skipped for a nil ResourceID before a reason was recorded). + if len(eliminated) == 0 { + return nil, fmt.Errorf("no eligible management cluster with at least %d available swift NICs among %d candidate(s)", + swiftNICsPerHCP, len(candidates)) + } + return nil, fmt.Errorf("no eligible management cluster with at least %d available swift NICs among %d candidate(s): %s", + swiftNICsPerHCP, len(candidates), strings.Join(eliminated, "; ")) + } + return chosen.resourceID, nil +} + +// ineligibilityReason returns a human-readable reason a management cluster cannot +// accept a new HCP, or "" when it is eligible. A candidate is eligible only when +// it is Schedulable, Ready, has a stamp identifier, and has an observed scheduling +// document (the source of its swift-NIC capacity). +func ineligibilityReason(managementCluster *fleetapi.ManagementCluster, scheduling *fleetapi.ManagementClusterScheduling) string { + if managementCluster.Spec.SchedulingPolicy != fleetapi.ManagementClusterSchedulingPolicySchedulable { + return fmt.Sprintf("scheduling policy is %q, not %q", managementCluster.Spec.SchedulingPolicy, fleetapi.ManagementClusterSchedulingPolicySchedulable) + } + if !meta.IsStatusConditionTrue(managementCluster.Status.Conditions, string(fleetapi.ManagementClusterConditionReady)) { + return "management cluster is not Ready" + } + if managementCluster.GetStampIdentifier() == "" { + return "management cluster has no stamp identifier" + } + if scheduling == nil { + return "no scheduling/capacity data available" + } + return "" +} + +// computeAvailableSwiftNICs returns the swift-NIC capacity still available on a +// management cluster: +// +// available = ScaleCeiling.Capacity[swift-nic] +// - ObservedResources.Usage[swift-nic] +// - len(NotReadyResourceIDs) * swiftNICsPerHCP +// - len(PendingAssignedClusters) * swiftNICsPerHCP +// +// Ready HCPs are already reflected in Usage, so they are not reserved again. +// NotReady HCPs may not yet consume their NICs, so each reserves swiftNICsPerHCP. +// Pending (just-scheduled, not-yet-observed) HCPs likewise reserve +// swiftNICsPerHCP. Capacity is bounded against the ScaleCeiling (max node count) +// so the estimate reflects the worst case. Empty/nil list entries do not +// correspond to a real HCP and are not counted toward the reservation. +func computeAvailableSwiftNICs(scheduling *fleetapi.ManagementClusterScheduling) int64 { + ceiling := swiftNICCount(scheduling.Status.ScaleCeiling.Capacity) + usage := swiftNICCount(scheduling.Status.ObservedResources.Usage) + notReady := countNonEmpty(scheduling.Status.NotReadyResourceIDs) * swiftNICsPerHCP + pending := countNonNilResourceIDs(scheduling.Status.PendingAssignedClusters) * swiftNICsPerHCP + return ceiling - usage - notReady - pending +} + +// countNonNilResourceIDs counts the non-nil entries of a resource ID slice; a +// nil entry does not correspond to a real HCP and must not reserve capacity. +func countNonNilResourceIDs(ids []*azcorearm.ResourceID) int64 { + var count int64 + for _, id := range ids { + if id != nil { + count++ + } + } + return count +} + +// countNonEmpty counts the non-empty entries of a string slice. +func countNonEmpty(values []string) int64 { + var count int64 + for _, value := range values { + if value != "" { + count++ + } + } + return count +} + +// swiftNICCount returns the swift-NIC quantity in a ResourceList as an int64, +// or 0 when the resource is absent. +func swiftNICCount(resources corev1.ResourceList) int64 { + quantity, ok := resources[kuberesources.SwiftNICResourceName] + if !ok { + return 0 + } + return quantity.Value() +} + +// reservePendingAssignment adds clusterResourceID to the chosen management +// cluster's PendingAssignedClusters list (idempotently). On a write conflict it +// returns an error so the workqueue retries the whole reconcile with backoff. +func (c *placementSyncer) reservePendingAssignment(ctx context.Context, managementClusterResourceID, clusterResourceID *azcorearm.ResourceID) error { + if managementClusterResourceID.Parent == nil { + return utils.TrackError(fmt.Errorf("management cluster resource ID %q has no parent stamp", managementClusterResourceID.String())) + } + stampIdentifier := managementClusterResourceID.Parent.Name + schedulingCRUD := c.fleetDBClient.Stamps().ManagementClusters(stampIdentifier).Scheduling() + + existing, err := schedulingCRUD.Get(ctx, fleetapi.SchedulingResourceName) + if err != nil { + return utils.TrackError(fmt.Errorf("failed to get scheduling document for management cluster %q: %w", managementClusterResourceID.String(), err)) + } + want := strings.ToLower(clusterResourceID.String()) + for _, pending := range existing.Status.PendingAssignedClusters { + if pending != nil && strings.ToLower(pending.String()) == want { + return nil // already reserved + } + } + + updated := existing.DeepCopy() + updated.Status.PendingAssignedClusters = append(updated.Status.PendingAssignedClusters, coreapi.DeepCopyResourceID(clusterResourceID)) + if _, err := schedulingCRUD.Replace(ctx, updated, nil); err != nil { + return utils.TrackError(fmt.Errorf("failed to reserve pending assignment on management cluster %q: %w", managementClusterResourceID.String(), err)) + } + return nil +} + +// setSpecPlacement records the chosen management cluster on +// ServiceProviderCluster.Spec.ManagementClusterResourceID and stamps +// Spec.ManagementClusterPlacementTime with the moment of placement. The base +// document is read from the informer cache (not a live Cosmos Get); the cached +// copy carries the etag that guards the optimistic Replace, so a stale cache can +// only produce a write conflict — never a lost update. On such a conflict it +// returns an error so the workqueue retries the whole reconcile with backoff. +func (c *placementSyncer) setSpecPlacement(ctx context.Context, key controllerutils.HCPClusterKey, chosen *azcorearm.ResourceID) error { + existing, err := c.serviceProviderClusterLister.Get(ctx, key.SubscriptionID, key.ResourceGroupName, key.HCPClusterName) + if cosmosstorageutils.IsNotFoundError(err) { + return nil + } + if err != nil { + return utils.TrackError(fmt.Errorf("failed to get ServiceProviderCluster from cache: %w", err)) + } + if !c.needsWork(existing) { + return nil + } + + replacement := existing.DeepCopy() + replacement.Spec.ManagementClusterResourceID = coreapi.DeepCopyResourceID(chosen) + // Stamp the placement time atomically with the intent, but only on first + // placement: preserve any existing timestamp across re-writes/backfills so it + // marks the moment of placement rather than the latest reconcile. + if replacement.Spec.ManagementClusterPlacementTime == nil { + now := metav1.Now() + replacement.Spec.ManagementClusterPlacementTime = &now + } + spcCRUD := c.cosmosClient.ServiceProviderClusters(key.SubscriptionID, key.ResourceGroupName, key.HCPClusterName) + if _, err := spcCRUD.Replace(ctx, replacement, nil); err != nil { + return utils.TrackError(fmt.Errorf("failed to update ServiceProviderCluster placement: %w", err)) + } + return nil +} diff --git a/backend/pkg/controllers/cluster/placement/placement_controller_test.go b/backend/pkg/controllers/cluster/placement/placement_controller_test.go new file mode 100644 index 00000000000..fae4ef3cb14 --- /dev/null +++ b/backend/pkg/controllers/cluster/placement/placement_controller_test.go @@ -0,0 +1,596 @@ +// Copyright 2026 Microsoft Corporation +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package placement + +import ( + "context" + "fmt" + "net/http" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.uber.org/mock/gomock" + + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/api/resource" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + azcorearm "github.com/Azure/azure-sdk-for-go/sdk/azcore/arm" + + arohcpv1alpha1 "github.com/openshift-online/ocm-sdk-go/arohcp/v1alpha1" + ocmerrors "github.com/openshift-online/ocm-sdk-go/errors" + + "github.com/Azure/ARO-HCP/backend/pkg/utils/controllerutils" + "github.com/Azure/ARO-HCP/internal/api/coreapi" + "github.com/Azure/ARO-HCP/internal/api/fleetapi" + "github.com/Azure/ARO-HCP/internal/api/metadataapi" + "github.com/Azure/ARO-HCP/internal/database/cosmosstoragetesting/corecosmosstoragetesting" + "github.com/Azure/ARO-HCP/internal/database/cosmosstoragetesting/fleetcosmosstoragetesting" + "github.com/Azure/ARO-HCP/internal/database/listertesting/corelistertesting" + "github.com/Azure/ARO-HCP/internal/database/listertesting/fleetlistertesting" + "github.com/Azure/ARO-HCP/internal/kuberesources" + "github.com/Azure/ARO-HCP/internal/ocm" +) + +// mcForStamp builds an eligible/ineligible ManagementCluster for a stamp. +func mcForStamp(stamp string, schedulable, ready bool) *fleetapi.ManagementCluster { + resourceID := metadataapi.Must(fleetapi.ToManagementClusterResourceID(stamp)) + policy := fleetapi.ManagementClusterSchedulingPolicyUnschedulable + if schedulable { + policy = fleetapi.ManagementClusterSchedulingPolicySchedulable + } + readyStatus := metav1.ConditionFalse + if ready { + readyStatus = metav1.ConditionTrue + } + return &fleetapi.ManagementCluster{ + CosmosMetadata: coreapi.CosmosMetadata{ResourceID: resourceID, PartitionKey: strings.ToLower(stamp)}, + ResourceID: resourceID, + Spec: fleetapi.ManagementClusterSpec{SchedulingPolicy: policy}, + Status: fleetapi.ManagementClusterStatus{ + Conditions: []metav1.Condition{{Type: string(fleetapi.ManagementClusterConditionReady), Status: readyStatus, Reason: "Test"}}, + }, + } +} + +// swiftResourceList returns a ResourceList with the given swift-NIC quantity, or +// nil when count < 0 (to model absent capacity data). +func swiftResourceList(count int64) corev1.ResourceList { + if count < 0 { + return nil + } + return corev1.ResourceList{kuberesources.SwiftNICResourceName: *resource.NewQuantity(count, resource.DecimalSI)} +} + +// dummyResourceIDs builds n distinct HCP-cluster resource IDs (for NotReady / +// Pending list length only; the exact values do not matter for capacity math). +func dummyResourceIDs(n int) []*azcorearm.ResourceID { + ids := make([]*azcorearm.ResourceID, 0, n) + for i := 0; i < n; i++ { + ids = append(ids, metadataapi.Must(azcorearm.ParseResourceID( + fmt.Sprintf("/subscriptions/%s/resourceGroups/%s/providers/Microsoft.RedHatOpenShift/hcpOpenShiftClusters/pending-%d", + testClusterSubscriptionID, testClusterResourceGroup, i)))) + } + return ids +} + +func schedulingDoc(stamp string, ceiling, usage, notReady, pending int64) *fleetapi.ManagementClusterScheduling { + resourceID := metadataapi.Must(fleetapi.ToManagementClusterSchedulingResourceID(stamp)) + notReadyIDs := make([]string, notReady) + for i := range notReadyIDs { + notReadyIDs[i] = fmt.Sprintf("/subscriptions/x/resourceGroups/y/providers/Microsoft.RedHatOpenShift/hcpOpenShiftClusters/nr-%s-%d", stamp, i) + } + return &fleetapi.ManagementClusterScheduling{ + CosmosMetadata: coreapi.CosmosMetadata{ResourceID: resourceID, PartitionKey: strings.ToLower(stamp)}, + Status: fleetapi.ManagementClusterSchedulingStatus{ + ObservedResources: fleetapi.ObservedResources{Usage: swiftResourceList(usage)}, + ScaleCeiling: fleetapi.ScaleCeiling{Capacity: swiftResourceList(ceiling)}, + NotReadyResourceIDs: notReadyIDs, + PendingAssignedClusters: dummyResourceIDs(int(pending)), + }, + } +} + +func TestComputeAvailableSwiftNICs(t *testing.T) { + tests := []struct { + name string + ceiling int64 + usage int64 + notReady int64 + pending int64 + expected int64 + }{ + {name: "empty capacity data => 0", ceiling: -1, usage: -1, expected: 0}, + {name: "ceiling only", ceiling: 9, expected: 9}, + {name: "usage subtracted", ceiling: 9, usage: 3, expected: 6}, + {name: "notReady eats slack (3 each)", ceiling: 9, usage: 0, notReady: 2, expected: 3}, + {name: "pending reserved (3 each)", ceiling: 9, usage: 0, pending: 2, expected: 3}, + {name: "all combined", ceiling: 30, usage: 6, notReady: 2, pending: 1, expected: 30 - 6 - 6 - 3}, + {name: "can go negative when overcommitted", ceiling: 3, usage: 0, notReady: 2, expected: 3 - 6}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + doc := schedulingDoc("s", tc.ceiling, tc.usage, tc.notReady, tc.pending) + assert.Equal(t, tc.expected, computeAvailableSwiftNICs(doc)) + }) + } +} + +func TestComputeAvailableSwiftNICs_IgnoresNilAndEmptyEntries(t *testing.T) { + doc := &fleetapi.ManagementClusterScheduling{ + Status: fleetapi.ManagementClusterSchedulingStatus{ + ScaleCeiling: fleetapi.ScaleCeiling{Capacity: swiftResourceList(9)}, + // One nil entry (must not reserve) + one real entry (reserves 3). + PendingAssignedClusters: []*azcorearm.ResourceID{ + nil, + metadataapi.Must(fleetapi.ToManagementClusterResourceID("x")), + }, + // One empty string (must not count) + one real entry (reserves 3). + NotReadyResourceIDs: []string{"", "/subscriptions/s/resourceGroups/rg/providers/Microsoft.RedHatOpenShift/hcpOpenShiftClusters/nr"}, + }, + } + // 9 - usage(0) - notReady(1*3) - pending(1*3) = 3 + assert.Equal(t, int64(3), computeAvailableSwiftNICs(doc)) +} + +// eligibleCandidate builds a schedulingCandidate for an eligible (schedulable + +// ready) management cluster with a scheduling document exposing `available` +// swift NICs (ceiling=available, no usage/notReady/pending). +func eligibleCandidate(stamp string, available int64) schedulingCandidate { + return schedulingCandidate{ + managementCluster: mcForStamp(stamp, true, true), + scheduling: schedulingDoc(stamp, available, 0, 0, 0), + } +} + +// TestSelectByCapacity exercises the single, pure elimination+selection function: +// all candidate elimination (eligibility AND capacity) now lives here, so the +// cases cover both ineligibility reasons and capacity-based spread/tie-breaking. +func TestSelectByCapacity(t *testing.T) { + rid := func(stamp string) *azcorearm.ResourceID { + return metadataapi.Must(fleetapi.ToManagementClusterResourceID(stamp)) + } + + tests := []struct { + name string + candidates []schedulingCandidate + expectedStamp string // "" => expect error + expectError bool + errContains string // substring the error must enumerate (elimination reason) + }{ + {name: "no candidates - error", candidates: nil, expectError: true}, + { + name: "not schedulable - eliminated with reason", + candidates: []schedulingCandidate{{managementCluster: mcForStamp("1", false, true), scheduling: schedulingDoc("1", 9, 0, 0, 0)}}, + expectError: true, + errContains: "scheduling policy", + }, + { + name: "not ready - eliminated with reason", + candidates: []schedulingCandidate{{managementCluster: mcForStamp("1", true, false), scheduling: schedulingDoc("1", 9, 0, 0, 0)}}, + expectError: true, + errContains: "not Ready", + }, + { + name: "no scheduling data - eliminated with reason", + candidates: []schedulingCandidate{{managementCluster: mcForStamp("1", true, true), scheduling: nil}}, + expectError: true, + errContains: "no scheduling/capacity data", + }, + { + name: "eligible but below threshold - eliminated with reason", + candidates: []schedulingCandidate{eligibleCandidate("1", 2)}, + expectError: true, + errContains: "insufficient swift-NIC capacity", + }, + {name: "single fit", candidates: []schedulingCandidate{eligibleCandidate("1", 3)}, expectedStamp: "1"}, + {name: "exactly at threshold fits", candidates: []schedulingCandidate{eligibleCandidate("1", 3)}, expectedStamp: "1"}, + { + name: "highest available among fits (spread load)", + candidates: []schedulingCandidate{eligibleCandidate("1", 9), eligibleCandidate("2", 3), eligibleCandidate("3", 6)}, + expectedStamp: "1", + }, + { + name: "skips those below threshold, picks highest fitting", + candidates: []schedulingCandidate{eligibleCandidate("1", 2), eligibleCandidate("2", 5), eligibleCandidate("3", 4)}, + expectedStamp: "2", + }, + { + name: "tie on available - lowest resource ID wins (order independent)", + candidates: []schedulingCandidate{eligibleCandidate("3", 3), eligibleCandidate("1", 3), eligibleCandidate("2", 3)}, + expectedStamp: "1", + }, + { + name: "mix of ineligible and eligible - picks the eligible fit", + candidates: []schedulingCandidate{ + {managementCluster: mcForStamp("1", true, false), scheduling: schedulingDoc("1", 9, 0, 0, 0)}, // not ready + eligibleCandidate("2", 3), + }, + expectedStamp: "2", + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + chosen, err := selectByCapacity(tc.candidates) + if tc.expectError { + require.Error(t, err) + assert.Nil(t, chosen) + if tc.errContains != "" { + assert.Contains(t, err.Error(), tc.errContains, "error should enumerate the elimination reason") + } + return + } + require.NoError(t, err) + require.NotNil(t, chosen) + assert.Equal(t, rid(tc.expectedStamp).String(), chosen.String()) + }) + } +} + +func TestPlacementSyncer_SyncOnce_Backfill(t *testing.T) { + ctx := context.Background() + + existing := newTestSPC(func(spc *coreapi.ServiceProviderCluster) { + spc.Status.ManagementClusterResourceID = testMgmtClusterResourceID() + }) + + mockDB := corecosmosstoragetesting.NewMockResourcesDBClient() + spcCRUD := mockDB.ServiceProviderClusters(testClusterSubscriptionID, testClusterResourceGroup, testClusterName) + created, err := spcCRUD.Create(ctx, existing, nil) + require.NoError(t, err) + + // No eligible MC and no fleet capacity data: proves backfill does NOT run + // fresh selection (which would fail). + syncer := &placementSyncer{ + serviceProviderClusterLister: &corelistertesting.SliceServiceProviderClusterLister{ServiceProviderClusters: []*coreapi.ServiceProviderCluster{created}}, + managementClusterLister: &fleetlistertesting.SliceManagementClusterLister{}, + cosmosClient: mockDB, + fleetDBClient: fleetcosmosstoragetesting.NewMockFleetDBClient(), + } + + key := controllerutils.HCPClusterKey{SubscriptionID: testClusterSubscriptionID, ResourceGroupName: testClusterResourceGroup, HCPClusterName: testClusterName} + require.NoError(t, syncer.SyncOnce(ctx, key)) + + updated, err := spcCRUD.Get(ctx, coreapi.ServiceProviderClusterResourceName) + require.NoError(t, err) + require.NotNil(t, updated.Spec.ManagementClusterResourceID, "Spec must be backfilled from Status") + assert.Equal(t, testMgmtClusterResourceID().String(), updated.Spec.ManagementClusterResourceID.String()) + require.NotNil(t, updated.Spec.ManagementClusterPlacementTime, "placement time must be recorded on backfill") +} + +func TestPlacementSyncer_SyncOnce_FreshSelection(t *testing.T) { + ctx := context.Background() + + existing := newTestSPC() // Spec and Status both nil + + mockDB := corecosmosstoragetesting.NewMockResourcesDBClient() + spcCRUD := mockDB.ServiceProviderClusters(testClusterSubscriptionID, testClusterResourceGroup, testClusterName) + created, err := spcCRUD.Create(ctx, existing, nil) + require.NoError(t, err) + + // Two eligible management clusters; stamp "1" has less available capacity + // (3) than stamp "2" (6), so spread (highest-available) must choose "2". + // The scheduling documents are read from the informer-cache lister for + // scoring; the fleet DB holds them too for the reservation write path. + sched1 := schedulingDoc("1", 6, 3, 0, 0) + sched2 := schedulingDoc("2", 6, 0, 0, 0) + fleetDB := fleetcosmosstoragetesting.NewMockFleetDBClient() + _, err = fleetDB.Stamps().ManagementClusters("1").Scheduling().Create(ctx, sched1, nil) + require.NoError(t, err) + _, err = fleetDB.Stamps().ManagementClusters("2").Scheduling().Create(ctx, sched2, nil) + require.NoError(t, err) + + syncer := &placementSyncer{ + serviceProviderClusterLister: &corelistertesting.SliceServiceProviderClusterLister{ServiceProviderClusters: []*coreapi.ServiceProviderCluster{created}}, + // No cluster in cache => no PendingClusterServiceID => fresh selection. + clusterLister: &corelistertesting.SliceClusterLister{}, + managementClusterLister: &fleetlistertesting.SliceManagementClusterLister{ManagementClusters: []*fleetapi.ManagementCluster{ + mcForStamp("1", true, true), + mcForStamp("2", true, true), + }}, + managementClusterSchedulingLister: &fleetlistertesting.SliceManagementClusterSchedulingLister{Schedulings: []*fleetapi.ManagementClusterScheduling{sched1, sched2}}, + cosmosClient: mockDB, + fleetDBClient: fleetDB, + } + + key := controllerutils.HCPClusterKey{SubscriptionID: testClusterSubscriptionID, ResourceGroupName: testClusterResourceGroup, HCPClusterName: testClusterName} + require.NoError(t, syncer.SyncOnce(ctx, key)) + + // Spec set to the emptier eligible MC (stamp "2") per spread selection. + updated, err := spcCRUD.Get(ctx, coreapi.ServiceProviderClusterResourceName) + require.NoError(t, err) + require.NotNil(t, updated.Spec.ManagementClusterResourceID) + assert.Equal(t, metadataapi.Must(fleetapi.ToManagementClusterResourceID("2")).String(), updated.Spec.ManagementClusterResourceID.String()) + require.NotNil(t, updated.Spec.ManagementClusterPlacementTime, "placement time must be recorded atomically with the placement intent") + + // Pending reservation recorded on the chosen MC's scheduling doc. + scheduling, err := fleetDB.Stamps().ManagementClusters("2").Scheduling().Get(ctx, fleetapi.SchedulingResourceName) + require.NoError(t, err) + require.Len(t, scheduling.Status.PendingAssignedClusters, 1) + assert.Equal(t, strings.ToLower(key.GetResourceID().String()), strings.ToLower(scheduling.Status.PendingAssignedClusters[0].String())) +} + +func TestPlacementSyncer_SyncOnce_NoCapacityFails(t *testing.T) { + ctx := context.Background() + + existing := newTestSPC() + mockDB := corecosmosstoragetesting.NewMockResourcesDBClient() + spcCRUD := mockDB.ServiceProviderClusters(testClusterSubscriptionID, testClusterResourceGroup, testClusterName) + created, err := spcCRUD.Create(ctx, existing, nil) + require.NoError(t, err) + + // Eligible MC but no scheduling doc in the cache => ineligible => no fit => error. + syncer := &placementSyncer{ + serviceProviderClusterLister: &corelistertesting.SliceServiceProviderClusterLister{ServiceProviderClusters: []*coreapi.ServiceProviderCluster{created}}, + clusterLister: &corelistertesting.SliceClusterLister{}, + managementClusterLister: &fleetlistertesting.SliceManagementClusterLister{ManagementClusters: []*fleetapi.ManagementCluster{mcForStamp("1", true, true)}}, + managementClusterSchedulingLister: &fleetlistertesting.SliceManagementClusterSchedulingLister{}, + cosmosClient: mockDB, + fleetDBClient: fleetcosmosstoragetesting.NewMockFleetDBClient(), + } + + key := controllerutils.HCPClusterKey{SubscriptionID: testClusterSubscriptionID, ResourceGroupName: testClusterResourceGroup, HCPClusterName: testClusterName} + require.Error(t, syncer.SyncOnce(ctx, key)) + + updated, err := spcCRUD.Get(ctx, coreapi.ServiceProviderClusterResourceName) + require.NoError(t, err) + assert.Nil(t, updated.Spec.ManagementClusterResourceID) +} + +// TestPlacementSyncer_SyncOnce_PlacementSource is the tabular test for the three +// placement sources SyncOnce chooses between when Spec is unset: +// - Status already set => backfill Spec from observed Status +// - both nil + PendingClusterServiceID set => backfill Spec from Cluster Service +// (rollout-race migration path; NOT a fresh capacity selection) +// - both nil + no PendingClusterServiceID => fresh capacity selection +// +// It also covers the defer case: a pending CS ID exists but Cluster Service has +// not reported a provision shard yet (Spec must stay nil, no error); and the +// Cluster Service error cases for a pending CS ID: a 404 (the pending ID never +// became a real cluster) falls through to a FRESH selection, while transient and +// other non-404 errors are returned so the workqueue retries. +func TestPlacementSyncer_SyncOnce_PlacementSource(t *testing.T) { + pendingCSID := metadataapi.Must(metadataapi.NewInternalID(testClusterServiceIDStr)) + + // freshMC is an eligible management cluster (distinct from the CS-mapped mc1) + // used only by the fresh-selection case. + const freshStamp = "fresh-mc" + freshMCResourceID := metadataapi.Must(fleetapi.ToManagementClusterResourceID(freshStamp)) + + tests := []struct { + name string + spc *coreapi.ServiceProviderCluster + cluster *coreapi.HCPOpenShiftCluster // nil => not present in cache + managementClusters []*fleetapi.ManagementCluster + schedulings []*fleetapi.ManagementClusterScheduling + csShard *arohcpv1alpha1.ProvisionShard + csError error + expectCSCall bool + expectedSpec string // "" => nil + expectError bool + }{ + { + name: "status set - backfill from status (no CS call, no fresh select)", + spc: newTestSPC(func(spc *coreapi.ServiceProviderCluster) { + spc.Status.ManagementClusterResourceID = testMgmtClusterResourceID() + }), + expectCSCall: false, + expectedSpec: testMgmtClusterResourceID().String(), + }, + { + name: "both nil + pending CS ID - backfill from Cluster Service (not fresh select)", + spc: newTestSPC(), + cluster: newTestHCPCluster(func(c *coreapi.HCPOpenShiftCluster) { + c.ServiceProviderProperties.ClusterServiceID = nil + c.ServiceProviderProperties.PendingClusterServiceID = &pendingCSID + }), + managementClusters: []*fleetapi.ManagementCluster{newTestManagementCluster()}, + csShard: metadataapi.Must(arohcpv1alpha1.NewProvisionShard(). + HREF(testProvisionShardHREF(testProvisionShardIDStr)). + Build()), + expectCSCall: true, + expectedSpec: testMgmtClusterResourceID().String(), + }, + { + name: "both nil + pending CS ID + shard not allocated - defer (no fresh select, no error)", + spc: newTestSPC(), + cluster: newTestHCPCluster(func(c *coreapi.HCPOpenShiftCluster) { + c.ServiceProviderProperties.ClusterServiceID = nil + c.ServiceProviderProperties.PendingClusterServiceID = &pendingCSID + }), + managementClusters: []*fleetapi.ManagementCluster{newTestManagementCluster()}, + csShard: metadataapi.Must(arohcpv1alpha1.NewProvisionShard().Build()), // empty HREF + expectCSCall: true, + expectedSpec: "", + }, + { + name: "both nil + no pending CS ID - fresh selection", + spc: newTestSPC(), + cluster: newTestHCPCluster(func(c *coreapi.HCPOpenShiftCluster) { + c.ServiceProviderProperties.ClusterServiceID = nil + }), + managementClusters: []*fleetapi.ManagementCluster{mcForStamp(freshStamp, true, true)}, + schedulings: []*fleetapi.ManagementClusterScheduling{schedulingDoc(freshStamp, 6, 0, 0, 0)}, + expectCSCall: false, + expectedSpec: freshMCResourceID.String(), + }, + { + // The pending CS ID is stale: an older backend recorded it then crashed + // before creating the cluster in Cluster Service, so CS returns 404. + // There is no committed placement to preserve, so SyncOnce must fall + // through to a FRESH capacity-aware selection (picks the eligible MC) + // rather than deferring forever. + name: "both nil + pending CS ID + CS 404 - fresh selection (stale pending ID)", + spc: newTestSPC(), + cluster: newTestHCPCluster(func(c *coreapi.HCPOpenShiftCluster) { + c.ServiceProviderProperties.ClusterServiceID = nil + c.ServiceProviderProperties.PendingClusterServiceID = &pendingCSID + }), + managementClusters: []*fleetapi.ManagementCluster{mcForStamp(freshStamp, true, true)}, + schedulings: []*fleetapi.ManagementClusterScheduling{schedulingDoc(freshStamp, 6, 0, 0, 0)}, + csError: metadataapi.Must(ocmerrors.NewError().Status(http.StatusNotFound).Build()), + expectCSCall: true, + expectedSpec: freshMCResourceID.String(), + }, + { + // A transient (non-404) Cluster Service error must NOT fresh-select: it + // is returned so the workqueue retries. An eligible MC is present to + // prove the error short-circuits before fresh selection (no placement + // is written). + name: "both nil + pending CS ID + CS transient error - return error (retry, no placement)", + spc: newTestSPC(), + cluster: newTestHCPCluster(func(c *coreapi.HCPOpenShiftCluster) { + c.ServiceProviderProperties.ClusterServiceID = nil + c.ServiceProviderProperties.PendingClusterServiceID = &pendingCSID + }), + managementClusters: []*fleetapi.ManagementCluster{mcForStamp(freshStamp, true, true)}, + schedulings: []*fleetapi.ManagementClusterScheduling{schedulingDoc(freshStamp, 6, 0, 0, 0)}, + csError: fmt.Errorf("connection refused"), + expectCSCall: true, + expectedSpec: "", + expectError: true, + }, + { + // A non-404 Cluster Service HTTP error (e.g. 500) must also be returned + // for retry: only a 404 means the cluster was never created and is safe + // to fresh-select. + name: "both nil + pending CS ID + CS non-404 error - return error (only 404 falls through)", + spc: newTestSPC(), + cluster: newTestHCPCluster(func(c *coreapi.HCPOpenShiftCluster) { + c.ServiceProviderProperties.ClusterServiceID = nil + c.ServiceProviderProperties.PendingClusterServiceID = &pendingCSID + }), + managementClusters: []*fleetapi.ManagementCluster{mcForStamp(freshStamp, true, true)}, + schedulings: []*fleetapi.ManagementClusterScheduling{schedulingDoc(freshStamp, 6, 0, 0, 0)}, + csError: metadataapi.Must(ocmerrors.NewError().Status(http.StatusInternalServerError).Build()), + expectCSCall: true, + expectedSpec: "", + expectError: true, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + ctx := context.Background() + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + mockDB := corecosmosstoragetesting.NewMockResourcesDBClient() + spcCRUD := mockDB.ServiceProviderClusters(testClusterSubscriptionID, testClusterResourceGroup, testClusterName) + created, err := spcCRUD.Create(ctx, tc.spc, nil) + require.NoError(t, err) + + clusters := []*coreapi.HCPOpenShiftCluster{} + if tc.cluster != nil { + clusters = append(clusters, tc.cluster) + } + + // The fleet DB backs the reservation write path (fresh selection only); + // seed it with the same scheduling docs used for scoring. + fleetDB := fleetcosmosstoragetesting.NewMockFleetDBClient() + for _, s := range tc.schedulings { + _, err := fleetDB.Stamps().ManagementClusters(s.PartitionKey).Scheduling().Create(ctx, s, nil) + require.NoError(t, err) + } + + mockCSClient := ocm.NewMockClusterServiceClientSpec(ctrl) + if tc.expectCSCall { + mockCSClient.EXPECT(). + GetClusterProvisionShard(gomock.Any(), pendingCSID). + Return(tc.csShard, tc.csError) + } + + syncer := &placementSyncer{ + serviceProviderClusterLister: &corelistertesting.SliceServiceProviderClusterLister{ServiceProviderClusters: []*coreapi.ServiceProviderCluster{created}}, + clusterLister: &corelistertesting.SliceClusterLister{Clusters: clusters}, + managementClusterLister: &fleetlistertesting.SliceManagementClusterLister{ManagementClusters: tc.managementClusters}, + managementClusterSchedulingLister: &fleetlistertesting.SliceManagementClusterSchedulingLister{Schedulings: tc.schedulings}, + cosmosClient: mockDB, + fleetDBClient: fleetDB, + clusterServiceClient: mockCSClient, + } + + key := controllerutils.HCPClusterKey{SubscriptionID: testClusterSubscriptionID, ResourceGroupName: testClusterResourceGroup, HCPClusterName: testClusterName} + err = syncer.SyncOnce(ctx, key) + if tc.expectError { + require.Error(t, err) + } else { + require.NoError(t, err) + } + + updated, err := spcCRUD.Get(ctx, coreapi.ServiceProviderClusterResourceName) + require.NoError(t, err) + if tc.expectedSpec != "" { + require.NotNil(t, updated.Spec.ManagementClusterResourceID) + assert.Equal(t, tc.expectedSpec, updated.Spec.ManagementClusterResourceID.String()) + require.NotNil(t, updated.Spec.ManagementClusterPlacementTime, "placement time must be recorded when a placement is written") + } else { + assert.Nil(t, updated.Spec.ManagementClusterResourceID, "no placement should be written") + } + }) + } +} + +// TestPlacementSyncer_setSpecPlacement_Timestamp covers the placement-time stamp: +// it is set on first placement and preserved (not overwritten) when a timestamp +// already exists, so it marks the moment of placement rather than the latest sync. +func TestPlacementSyncer_setSpecPlacement_Timestamp(t *testing.T) { + ctx := context.Background() + chosen := testMgmtClusterResourceID() + key := controllerutils.HCPClusterKey{SubscriptionID: testClusterSubscriptionID, ResourceGroupName: testClusterResourceGroup, HCPClusterName: testClusterName} + preExisting := metav1.NewTime(time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC)) + + tests := []struct { + name string + existingPlacement *metav1.Time + // wantPreserved, when non-nil, is the timestamp the write must keep intact; + // when nil the write must populate a fresh (non-nil) timestamp. + wantPreserved *metav1.Time + }{ + {name: "sets placement time on first placement", existingPlacement: nil, wantPreserved: nil}, + {name: "preserves an existing placement time across re-write", existingPlacement: &preExisting, wantPreserved: &preExisting}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + existing := newTestSPC(func(spc *coreapi.ServiceProviderCluster) { + spc.Spec.ManagementClusterPlacementTime = tc.existingPlacement + }) + mockDB := corecosmosstoragetesting.NewMockResourcesDBClient() + spcCRUD := mockDB.ServiceProviderClusters(testClusterSubscriptionID, testClusterResourceGroup, testClusterName) + created, err := spcCRUD.Create(ctx, existing, nil) + require.NoError(t, err) + + syncer := &placementSyncer{ + serviceProviderClusterLister: &corelistertesting.SliceServiceProviderClusterLister{ServiceProviderClusters: []*coreapi.ServiceProviderCluster{created}}, + cosmosClient: mockDB, + } + require.NoError(t, syncer.setSpecPlacement(ctx, key, chosen)) + + updated, err := spcCRUD.Get(ctx, coreapi.ServiceProviderClusterResourceName) + require.NoError(t, err) + require.NotNil(t, updated.Spec.ManagementClusterResourceID) + require.NotNil(t, updated.Spec.ManagementClusterPlacementTime, "placement time must be set") + if tc.wantPreserved != nil { + assert.Equal(t, tc.wantPreserved.Unix(), updated.Spec.ManagementClusterPlacementTime.Unix(), "existing placement time must be preserved") + } + }) + } +} diff --git a/backend/pkg/controllers/metrics/cluster_info_metrics_handler.go b/backend/pkg/controllers/metrics/cluster_info_metrics_handler.go index a5fbb98975e..d59d753a0bf 100644 --- a/backend/pkg/controllers/metrics/cluster_info_metrics_handler.go +++ b/backend/pkg/controllers/metrics/cluster_info_metrics_handler.go @@ -26,20 +26,36 @@ import ( ) type clusterInfoMetricsHandler struct { - clusterInfo *prometheus.GaugeVec + clusterInfo *prometheus.GaugeVec + placementTime *prometheus.GaugeVec } -// NewClusterInfoMetricsHandler creates a metrics handler that emits a -// backend_cluster_info gauge for each cluster, labeled with its management -// cluster placement. Use PromQL joins to combine with other per-cluster metrics. +// NewClusterInfoMetricsHandler creates a metrics handler that emits, per cluster: +// +// - backend_cluster_info: an info gauge (value always 1) carrying the cluster's +// resource ID, subscription ID, and observed management-cluster placement +// (management_cluster_resource_id, mirrored from +// ServiceProviderCluster.Status.ManagementClusterResourceID). Use PromQL joins +// to combine it with other per-cluster metrics. +// +// - backend_cluster_placement_time_seconds: a kube-state-metrics-style gauge +// emitted only once the scheduler has recorded placement intent +// (ServiceProviderCluster.Spec.ManagementClusterResourceID is set). Its value +// is the unix timestamp (seconds) at which placement was recorded +// (Spec.ManagementClusterPlacementTime) — a stable timestamp, not a duration. +// Compute time-to-placement in PromQL against the cluster's creation timestamp. func NewClusterInfoMetricsHandler(registerer prometheus.Registerer) Handler[*coreapi.ServiceProviderCluster] { handler := &clusterInfoMetricsHandler{ clusterInfo: prometheus.NewGaugeVec(prometheus.GaugeOpts{ Name: "backend_cluster_info", Help: "Info metric for clusters. Value is always 1.", }, []string{"resource_id", "subscription_id", "management_cluster_resource_id"}), + placementTime: prometheus.NewGaugeVec(prometheus.GaugeOpts{ + Name: "backend_cluster_placement_time_seconds", + Help: "Unix timestamp (seconds) at which the scheduler recorded a management-cluster placement (Spec.ManagementClusterPlacementTime). Emitted per cluster once placement intent is set; kube-state-metrics style — compute time-to-placement in PromQL against the cluster's creation timestamp.", + }, []string{"resource_id", "subscription_id"}), } - registerer.MustRegister(handler.clusterInfo) + registerer.MustRegister(handler.clusterInfo, handler.placementTime) return handler } @@ -58,6 +74,19 @@ func (h *clusterInfoMetricsHandler) Sync(_ context.Context, serviceProviderClust "subscription_id": subscriptionID, "management_cluster_resource_id": managementClusterResourceID, }).Set(1.0) + + // Placement time (kube-state-metrics style): expose the timestamp at which the + // scheduler recorded placement intent (Spec.ManagementClusterPlacementTime) as + // unix seconds. Clear any prior series first so an unplaced cluster — or one + // without a recorded placement timestamp — carries no stale series. + h.placementTime.DeletePartialMatch(prometheus.Labels{"resource_id": resourceID}) + if serviceProviderCluster.Spec.ManagementClusterResourceID == nil || serviceProviderCluster.Spec.ManagementClusterPlacementTime == nil { + return + } + h.placementTime.With(prometheus.Labels{ + "resource_id": resourceID, + "subscription_id": subscriptionID, + }).Set(float64(serviceProviderCluster.Spec.ManagementClusterPlacementTime.Unix())) } func (h *clusterInfoMetricsHandler) Delete(key string) { @@ -69,6 +98,7 @@ func (h *clusterInfoMetricsHandler) Delete(key string) { return } h.clusterInfo.DeletePartialMatch(prometheus.Labels{"resource_id": resourceID}) + h.placementTime.DeletePartialMatch(prometheus.Labels{"resource_id": resourceID}) } func clusterResourceIDFromServiceProviderCluster(serviceProviderCluster *coreapi.ServiceProviderCluster) *azcorearm.ResourceID { diff --git a/backend/pkg/controllers/metrics/cluster_info_metrics_handler_test.go b/backend/pkg/controllers/metrics/cluster_info_metrics_handler_test.go index 128be9c5044..f943df3a3b0 100644 --- a/backend/pkg/controllers/metrics/cluster_info_metrics_handler_test.go +++ b/backend/pkg/controllers/metrics/cluster_info_metrics_handler_test.go @@ -17,13 +17,17 @@ package metrics import ( "context" "fmt" + "strconv" "strings" "testing" + "time" "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/testutil" "github.com/stretchr/testify/require" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + azcorearm "github.com/Azure/azure-sdk-for-go/sdk/azcore/arm" "github.com/Azure/ARO-HCP/internal/api/coreapi" @@ -34,6 +38,7 @@ func TestClusterInfoMetricsHandler(t *testing.T) { clusterResourceID := metadataapi.Must(azcorearm.ParseResourceID("/subscriptions/sub-1/resourceGroups/rg/providers/Microsoft.RedHatOpenShift/hcpOpenShiftClusters/cluster-1")) spcResourceID := metadataapi.Must(azcorearm.ParseResourceID(clusterResourceID.String() + "/serviceProviderClusters/default")) mcResourceID := metadataapi.Must(azcorearm.ParseResourceID("/providers/microsoft.redhatopenshift/stamps/1/managementclusters/default")) + mcResourceIDOther := metadataapi.Must(azcorearm.ParseResourceID("/providers/microsoft.redhatopenshift/stamps/2/managementclusters/default")) tests := []struct { name string @@ -41,9 +46,12 @@ func TestClusterInfoMetricsHandler(t *testing.T) { expectedMetrics string }{ { - name: "emits cluster info with management cluster resource ID", + name: "observed placement reported from status", spc: &coreapi.ServiceProviderCluster{ CosmosMetadata: coreapi.CosmosMetadata{ResourceID: spcResourceID}, + Spec: coreapi.ServiceProviderClusterSpec{ + ManagementClusterResourceID: mcResourceID, + }, Status: coreapi.ServiceProviderClusterStatus{ ManagementClusterResourceID: mcResourceID, }, @@ -54,7 +62,7 @@ backend_cluster_info{management_cluster_resource_id="%s",resource_id="%s",subscr `, resourceIDMetricLabel(mcResourceID), resourceIDMetricLabel(clusterResourceID), subscriptionIDMetricLabel(clusterResourceID)), }, { - name: "emits empty management cluster resource ID when not placed", + name: "no observed placement, empty management cluster resource ID", spc: &coreapi.ServiceProviderCluster{ CosmosMetadata: coreapi.CosmosMetadata{ResourceID: spcResourceID}, Status: coreapi.ServiceProviderClusterStatus{}, @@ -64,6 +72,22 @@ backend_cluster_info{management_cluster_resource_id="%s",resource_id="%s",subscr backend_cluster_info{management_cluster_resource_id="",resource_id="%s",subscription_id="%s"} 1 `, resourceIDMetricLabel(clusterResourceID), subscriptionIDMetricLabel(clusterResourceID)), }, + { + name: "management cluster resource id mirrors the observed status placement", + spc: &coreapi.ServiceProviderCluster{ + CosmosMetadata: coreapi.CosmosMetadata{ResourceID: spcResourceID}, + Spec: coreapi.ServiceProviderClusterSpec{ + ManagementClusterResourceID: mcResourceID, + }, + Status: coreapi.ServiceProviderClusterStatus{ + ManagementClusterResourceID: mcResourceIDOther, + }, + }, + expectedMetrics: fmt.Sprintf(`# HELP backend_cluster_info Info metric for clusters. Value is always 1. +# TYPE backend_cluster_info gauge +backend_cluster_info{management_cluster_resource_id="%s",resource_id="%s",subscription_id="%s"} 1 +`, resourceIDMetricLabel(mcResourceIDOther), resourceIDMetricLabel(clusterResourceID), subscriptionIDMetricLabel(clusterResourceID)), + }, } for _, tt := range tests { @@ -92,7 +116,7 @@ func TestClusterInfoMetricsHandler_DeleteCleansUp(t *testing.T) { require.NoError(t, testutil.GatherAndCompare(reg, strings.NewReader(""), "backend_cluster_info")) } -func TestClusterInfoMetricsHandler_UpdatesOnPlacementChange(t *testing.T) { +func TestClusterInfoMetricsHandler_UpdatesOnManagementClusterChange(t *testing.T) { spcResourceID := metadataapi.Must(azcorearm.ParseResourceID("/subscriptions/sub-1/resourceGroups/rg/providers/Microsoft.RedHatOpenShift/hcpOpenShiftClusters/cluster-1/serviceProviderClusters/default")) clusterResourceID := spcResourceID.Parent mc1 := metadataapi.Must(azcorearm.ParseResourceID("/providers/microsoft.redhatopenshift/stamps/1/managementclusters/default")) @@ -101,18 +125,122 @@ func TestClusterInfoMetricsHandler_UpdatesOnPlacementChange(t *testing.T) { reg := prometheus.NewRegistry() handler := NewClusterInfoMetricsHandler(reg) + // First observed on mc1. handler.Sync(context.Background(), &coreapi.ServiceProviderCluster{ CosmosMetadata: coreapi.CosmosMetadata{ResourceID: spcResourceID}, + Spec: coreapi.ServiceProviderClusterSpec{ManagementClusterResourceID: mc1}, Status: coreapi.ServiceProviderClusterStatus{ManagementClusterResourceID: mc1}, }) + // Observed placement moves to mc2. handler.Sync(context.Background(), &coreapi.ServiceProviderCluster{ CosmosMetadata: coreapi.CosmosMetadata{ResourceID: spcResourceID}, + Spec: coreapi.ServiceProviderClusterSpec{ManagementClusterResourceID: mc1}, Status: coreapi.ServiceProviderClusterStatus{ManagementClusterResourceID: mc2}, }) + // Only the current series survives: DeletePartialMatch on resource_id clears + // the stale management_cluster_resource_id=mc1 series so no duplicate lingers + // after the label value changes. expected := fmt.Sprintf(`# HELP backend_cluster_info Info metric for clusters. Value is always 1. # TYPE backend_cluster_info gauge backend_cluster_info{management_cluster_resource_id="%s",resource_id="%s",subscription_id="%s"} 1 `, resourceIDMetricLabel(mc2), resourceIDMetricLabel(clusterResourceID), subscriptionIDMetricLabel(clusterResourceID)) require.NoError(t, testutil.GatherAndCompare(reg, strings.NewReader(expected), "backend_cluster_info")) } + +func TestClusterInfoMetricsHandler_PlacementTime(t *testing.T) { + clusterResourceID := metadataapi.Must(azcorearm.ParseResourceID("/subscriptions/sub-1/resourceGroups/rg/providers/Microsoft.RedHatOpenShift/hcpOpenShiftClusters/cluster-1")) + spcResourceID := metadataapi.Must(azcorearm.ParseResourceID(clusterResourceID.String() + "/serviceProviderClusters/default")) + mcResourceID := metadataapi.Must(azcorearm.ParseResourceID("/providers/microsoft.redhatopenshift/stamps/1/managementclusters/default")) + + // A real placement timestamp; the gauge value is its unix-seconds representation. + placedAt := metav1.Date(2026, 1, 1, 12, 2, 0, 0, time.UTC) + // Format the expected value exactly as the prometheus text exposition does + // (strconv 'g', -1, 64) so large unix timestamps compare correctly. + placedValue := strconv.FormatFloat(float64(placedAt.Unix()), 'g', -1, 64) + + placementTimeHeader := `# HELP backend_cluster_placement_time_seconds Unix timestamp (seconds) at which the scheduler recorded a management-cluster placement (Spec.ManagementClusterPlacementTime). Emitted per cluster once placement intent is set; kube-state-metrics style — compute time-to-placement in PromQL against the cluster's creation timestamp. +# TYPE backend_cluster_placement_time_seconds gauge +` + + tests := []struct { + name string + spec coreapi.ServiceProviderClusterSpec + expected string + }{ + { + name: "emitted as the placement unix timestamp when intent and timestamp are set", + spec: coreapi.ServiceProviderClusterSpec{ + ManagementClusterResourceID: mcResourceID, + ManagementClusterPlacementTime: &placedAt, + }, + expected: placementTimeHeader + fmt.Sprintf(`backend_cluster_placement_time_seconds{resource_id="%s",subscription_id="%s"} %s +`, resourceIDMetricLabel(clusterResourceID), subscriptionIDMetricLabel(clusterResourceID), placedValue), + }, + { + name: "not emitted while placement intent is unset", + spec: coreapi.ServiceProviderClusterSpec{ManagementClusterPlacementTime: &placedAt}, + expected: "", + }, + { + name: "not emitted when the placement timestamp is nil", + spec: coreapi.ServiceProviderClusterSpec{ManagementClusterResourceID: mcResourceID}, + expected: "", + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + reg := prometheus.NewRegistry() + handler := NewClusterInfoMetricsHandler(reg) + handler.Sync(context.Background(), &coreapi.ServiceProviderCluster{ + CosmosMetadata: coreapi.CosmosMetadata{ResourceID: spcResourceID}, + Spec: tc.spec, + }) + require.NoError(t, testutil.GatherAndCompare(reg, strings.NewReader(tc.expected), "backend_cluster_placement_time_seconds")) + }) + } +} + +func TestClusterInfoMetricsHandler_PlacementTimeClearedWhenIntentRemoved(t *testing.T) { + clusterResourceID := metadataapi.Must(azcorearm.ParseResourceID("/subscriptions/sub-1/resourceGroups/rg/providers/Microsoft.RedHatOpenShift/hcpOpenShiftClusters/cluster-1")) + spcResourceID := metadataapi.Must(azcorearm.ParseResourceID(clusterResourceID.String() + "/serviceProviderClusters/default")) + mcResourceID := metadataapi.Must(azcorearm.ParseResourceID("/providers/microsoft.redhatopenshift/stamps/1/managementclusters/default")) + placedAt := metav1.Date(2026, 1, 1, 12, 2, 0, 0, time.UTC) + + reg := prometheus.NewRegistry() + handler := NewClusterInfoMetricsHandler(reg) + + // Placed with a timestamp: the series is emitted. + handler.Sync(context.Background(), &coreapi.ServiceProviderCluster{ + CosmosMetadata: coreapi.CosmosMetadata{ResourceID: spcResourceID}, + Spec: coreapi.ServiceProviderClusterSpec{ + ManagementClusterResourceID: mcResourceID, + ManagementClusterPlacementTime: &placedAt, + }, + }) + // Intent cleared: the placement-time series must be removed. + handler.Sync(context.Background(), &coreapi.ServiceProviderCluster{ + CosmosMetadata: coreapi.CosmosMetadata{ResourceID: spcResourceID}, + }) + require.NoError(t, testutil.GatherAndCompare(reg, strings.NewReader(""), "backend_cluster_placement_time_seconds")) +} + +func TestClusterInfoMetricsHandler_PlacementTimeDeletedOnDelete(t *testing.T) { + clusterResourceID := metadataapi.Must(azcorearm.ParseResourceID("/subscriptions/sub-1/resourceGroups/rg/providers/Microsoft.RedHatOpenShift/hcpOpenShiftClusters/cluster-1")) + spcResourceID := metadataapi.Must(azcorearm.ParseResourceID(clusterResourceID.String() + "/serviceProviderClusters/default")) + mcResourceID := metadataapi.Must(azcorearm.ParseResourceID("/providers/microsoft.redhatopenshift/stamps/1/managementclusters/default")) + placedAt := metav1.Date(2026, 1, 1, 12, 2, 0, 0, time.UTC) + + reg := prometheus.NewRegistry() + handler := NewClusterInfoMetricsHandler(reg) + handler.Sync(context.Background(), &coreapi.ServiceProviderCluster{ + CosmosMetadata: coreapi.CosmosMetadata{ResourceID: spcResourceID}, + Spec: coreapi.ServiceProviderClusterSpec{ + ManagementClusterResourceID: mcResourceID, + ManagementClusterPlacementTime: &placedAt, + }, + }) + handler.Delete(strings.ToLower(spcResourceID.String())) + require.NoError(t, testutil.GatherAndCompare(reg, strings.NewReader(""), "backend_cluster_placement_time_seconds")) +} diff --git a/docs/cosmos-data-flow.md b/docs/cosmos-data-flow.md index 8accd67cd4b..ed1e46c32ca 100644 --- a/docs/cosmos-data-flow.md +++ b/docs/cosmos-data-flow.md @@ -485,7 +485,8 @@ which performs a **transactional batch** to atomically update the operation and | | Object | Fields | |---|--------|--------| | Read | `HCPOpenShiftCluster` | | -| Read | `ServiceProviderCluster` | | +| Read | `ServiceProviderCluster` | | +| Read | `ManagementCluster` | | | Read | `Subscription` | | | Read | Cluster Service | | | **Write** | **`HCPOpenShiftCluster`** | | @@ -967,20 +968,51 @@ No Cosmos writes. Posts `NodePoolUpgradePolicy` to Cluster Service. ### Other Controllers +#### PlacementController + +**File:** [placement_controller.go](../backend/pkg/controllers/cluster/placement/placement_controller.go) +**Trigger:** Cluster informer, 5-minute resync (20 workers) +**Gate (needsWork on ServiceProviderCluster):** +- `ServiceProviderCluster.Spec.ManagementClusterResourceID` == nil + +Resolves the scheduler's *desired* placement (`Spec.ManagementClusterResourceID`). When the HCP was already placed by ManagementClusterPlacementSync (`Status.ManagementClusterResourceID` set) but Spec is still nil, it backfills Spec from Status (rollout, no re-scheduling). When instead both Spec and Status are nil but the HCP already carries a `PendingClusterServiceID` (a rollout-race record created by a prior backend version), it asks Cluster Service where that cluster was placed (`GetClusterProvisionShard` → provision shard → the matching `ManagementCluster` by `Status.ClusterServiceProvisionShardID`) and backfills Spec from that already-decided placement rather than fresh-scheduling — a migration-only targeted Cluster Service read; it defers (no write) when Cluster Service has not yet reported a shard. Otherwise it selects an eligible management cluster (`ManagementCluster.Spec.SchedulingPolicy == Schedulable` AND Ready condition True) with sufficient SWIFT-NIC capacity, where `available = ScaleCeiling.Capacity[swift-nic] - ObservedResources.Usage[swift-nic] - (non-empty NotReadyResourceIDs)*3 - (non-nil PendingAssignedClusters)*3` (empty-string / nil entries reserve 0), an HCP fits when `available >= 3`, and among fitting clusters it chooses the highest-available eligible cluster (spread load evenly across management clusters), breaking ties by lowest resource ID. All candidate elimination (eligibility and capacity) happens in one place (`selectByCapacity`), which surfaces the per-candidate elimination reasons in its error. The `ManagementCluster` and `ManagementClusterScheduling` reads come from the informer caches; it reserves capacity on the chosen cluster (a live etag-guarded read-modify-write) before recording the intent, and stamps `Spec.ManagementClusterPlacementTime` with the current time atomically with `Spec.ManagementClusterResourceID` on first placement (preserved across later re-writes/backfills). + +| | Object | Fields | +|---|--------|--------| +| Read | `ServiceProviderCluster` | | +| Read | `HCPOpenShiftCluster` (rollout-race only) | | +| Read | Cluster Service (rollout-race only) | | +| Read | `ManagementCluster` (all) | | +| Read | `ManagementClusterScheduling` (per eligible MC) | | +| **Write** | **`ManagementClusterScheduling`** | | +| **Write** | **`ServiceProviderCluster`** | | + +#### PendingCleanupController + +**File:** [pending_cleanup_controller.go](../backend/pkg/controllers/cluster/placement/pending_cleanup_controller.go) +**Trigger:** ManagementCluster informer, 10-minute resync (5 workers) + +Garbage-collects stale entries from each management cluster's `Status.PendingAssignedClusters`. Each entry's *effective* placement is the referenced ServiceProviderCluster's `Status.ManagementClusterResourceID` (Cluster Service reality) when set, falling back to `Spec.ManagementClusterResourceID` only when Status is unset. An entry is kept when that effective placement points at this management cluster, or is still nil (placement in progress); it is removed when the effective placement points at a different management cluster or the ServiceProviderCluster no longer exists. Reservations that become observed (present in `ReadyResourceIDs`/`NotReadyResourceIDs`) are cleared by CapacityReportingController instead. + +| | Object | Fields | +|---|--------|--------| +| Read | `ManagementClusterScheduling` | | +| Read | `ServiceProviderCluster` (per pending entry) | | +| **Write** | **`ManagementClusterScheduling`** | | + #### ManagementClusterPlacementSync **File:** [management_cluster_placement_sync.go](../backend/pkg/controllers/cluster/placement/management_cluster_placement_sync.go) **Trigger:** Cluster informer, 5-minute resync -**Gate (needsWork on ServiceProviderCluster):** -- `ServiceProviderCluster.Status.ManagementClusterResourceID` == nil +Records the observed placement (`Status.ManagementClusterResourceID`) from the Cluster Service provision shard. Cluster Service is queried **only while the observed placement is unknown** (`Status` unset); once a shard has been observed and recorded, the CS lookup is skipped on subsequent syncs. | | Object | Fields | |---|--------|--------| -| Read | `ServiceProviderCluster` | | -| Read | `HCPOpenShiftCluster` | | -| Read | Cluster Service | | +| Read | `ServiceProviderCluster` | | +| Read | `HCPOpenShiftCluster` | | +| Read | Cluster Service | | | Read | `ManagementCluster` | | -| **Write** | **`ServiceProviderCluster`** | | +| **Write** | **`ServiceProviderCluster`** | | #### BackfillClusterUID @@ -1473,13 +1505,29 @@ Single writer today (`RequirementsValid` only). Single writer, but read by `ClusterClusterServiceCreate` (gate), `OperationClusterUpdate`, and `TriggerControlPlaneUpgrade`. +### `ServiceProviderCluster.Spec.ManagementClusterResourceID` + +| Actor | When | +|-------|------| +| [PlacementController](#placementcontroller) | Sets the scheduler's placement intent: backfilled from `Status.ManagementClusterResourceID` when already placed, backfilled from Cluster Service (via `PendingClusterServiceID`) for rollout-race records, otherwise the capacity-selected eligible management cluster | + +This is the *desired* placement (scheduler intent), owned solely by the PlacementController. It is read by `ClusterPendingClusterServiceIDAssign` (gate: must be non-nil before a pending CS ID is assigned) and `ClusterClusterServiceCreate` (resolves the placed management cluster to pin the CS provision shard). + +### `ServiceProviderCluster.Spec.ManagementClusterPlacementTime` + +| Actor | When | +|-------|------| +| [PlacementController](#placementcontroller) | Stamps the current time when placement intent is first recorded (atomically with `Spec.ManagementClusterResourceID`); preserved across later re-writes/backfills | + +Marks the moment of placement (not the latest reconcile), owned solely by the PlacementController. Exposed kube-state-metrics style as the `backend_cluster_placement_time_seconds` gauge — a unix timestamp (seconds), not a duration. Time-to-placement is computed in PromQL against the cluster's creation timestamp (`placement_ts - creation_ts`). + ### `ServiceProviderCluster.Status.ManagementClusterResourceID` | Actor | When | |-------|------| -| [ManagementClusterPlacementSync](#managementclusterplacementsync) | Sets from CS provision shard | +| [ManagementClusterPlacementSync](#managementclusterplacementsync) | Resolves from the CS provision shard when unset (the CS lookup is skipped once the shard has been observed) | -Single writer, but gates `CreateClusterScopedReadDesires` and deletion cleanup. +This is the *observed* placement. It gates `CreateClusterScopedReadDesires` and deletion cleanup, and seeds `PlacementController`'s rollout backfill. ### `ServiceProviderCluster.Status.HostedClusterNamespace` diff --git a/fleet/pkg/controllers/capacityreporting/capacity_reporting_controller.go b/fleet/pkg/controllers/capacityreporting/capacity_reporting_controller.go index 2c30b6192d2..e79f36a8386 100644 --- a/fleet/pkg/controllers/capacityreporting/capacity_reporting_controller.go +++ b/fleet/pkg/controllers/capacityreporting/capacity_reporting_controller.go @@ -24,12 +24,15 @@ import ( "context" "encoding/json" "fmt" + "strings" "time" "k8s.io/apimachinery/pkg/api/meta" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/client-go/tools/cache" + azcorearm "github.com/Azure/azure-sdk-for-go/sdk/azcore/arm" + fleetcontrollers "github.com/Azure/ARO-HCP/fleet/pkg/controllers/base" "github.com/Azure/ARO-HCP/internal/api/fleetapi" "github.com/Azure/ARO-HCP/internal/api/kubeapplierapi" @@ -110,7 +113,7 @@ func (s *capacityReportingSyncer) SyncOnce(ctx context.Context, key fleetcontrol capacity := ComputeObservedResources(report) condition := evaluateCapacityCondition(report) - return s.persistObservedResources(ctx, key.StampIdentifier, capacity, condition) + return s.persistObservedResources(ctx, key.StampIdentifier, capacity, report.Status.HostedControlPlanes.ReadyResourceIDs, report.Status.HostedControlPlanes.NotReadyResourceIDs, condition) } func evaluateCapacityCondition(report *capacityreportv1alpha1.CapacityReport) metav1.Condition { @@ -129,7 +132,7 @@ func evaluateCapacityCondition(report *capacityreportv1alpha1.CapacityReport) me } } -func (s *capacityReportingSyncer) persistObservedResources(ctx context.Context, stampIdentifier string, observed fleetapi.ObservedResources, condition metav1.Condition) error { +func (s *capacityReportingSyncer) persistObservedResources(ctx context.Context, stampIdentifier string, observed fleetapi.ObservedResources, readyResourceIDs, notReadyResourceIDs []string, condition metav1.Condition) error { existing, err := fleetcosmosstorage.GetOrCreateManagementClusterScheduling(ctx, s.fleetDBClient, stampIdentifier) if err != nil { return err @@ -140,12 +143,51 @@ func (s *capacityReportingSyncer) persistObservedResources(ctx context.Context, updated := existing.DeepCopy() meta.SetStatusCondition(&updated.Status.Conditions, condition) updated.Status.ObservedResources = observed + // Mirror the ready/not-ready HCP resource IDs from the CapacityReport CR. + updated.Status.ReadyResourceIDs = readyResourceIDs + updated.Status.NotReadyResourceIDs = notReadyResourceIDs + // Observation-based cleanup: drop pending reservations that are now observed + // (present in Ready ∪ NotReady). Their swift-NIC capacity is accounted for by + // the observed data, so the transient reservation is no longer needed. + updated.Status.PendingAssignedClusters = dropObservedPendingAssignments(updated.Status.PendingAssignedClusters, readyResourceIDs, notReadyResourceIDs) if _, err := schedulingCRUD.Replace(ctx, updated, nil); err != nil { return utils.TrackError(err) } return nil } +// dropObservedPendingAssignments returns pending reservations minus any whose +// cluster resource ID now appears in the observed ready or not-ready sets. Nil +// entries are dropped. It returns nil when nothing remains; because the field is +// tagged omitempty, a nil slice is omitted from the serialized document rather +// than encoded as an empty array. +func dropObservedPendingAssignments(pending []*azcorearm.ResourceID, readyResourceIDs, notReadyResourceIDs []string) []*azcorearm.ResourceID { + if len(pending) == 0 { + return nil + } + observed := make(map[string]struct{}, len(readyResourceIDs)+len(notReadyResourceIDs)) + for _, id := range readyResourceIDs { + observed[strings.ToLower(id)] = struct{}{} + } + for _, id := range notReadyResourceIDs { + observed[strings.ToLower(id)] = struct{}{} + } + kept := make([]*azcorearm.ResourceID, 0, len(pending)) + for _, entry := range pending { + if entry == nil { + continue + } + if _, ok := observed[strings.ToLower(entry.String())]; ok { + continue // now observed → drop the reservation + } + kept = append(kept, entry) + } + if len(kept) == 0 { + return nil + } + return kept +} + // GetCapacityReport reads and unmarshals the CapacityReport from the // ReadDesire lister. func GetCapacityReport(ctx context.Context, readDesireLister kubeapplierlisters.ReadDesireLister, stampIdentifier string) (*capacityreportv1alpha1.CapacityReport, error) { diff --git a/fleet/pkg/controllers/capacityreporting/controller_test.go b/fleet/pkg/controllers/capacityreporting/controller_test.go index c834a83a87d..6b4e7cda3d5 100644 --- a/fleet/pkg/controllers/capacityreporting/controller_test.go +++ b/fleet/pkg/controllers/capacityreporting/controller_test.go @@ -18,6 +18,7 @@ import ( "context" "encoding/json" "net/http" + "strings" "testing" "github.com/stretchr/testify/assert" @@ -34,6 +35,7 @@ import ( "github.com/Azure/azure-sdk-for-go/sdk/data/azcosmos" fleetcontrollers "github.com/Azure/ARO-HCP/fleet/pkg/controllers/base" + "github.com/Azure/ARO-HCP/internal/api/coreapi" "github.com/Azure/ARO-HCP/internal/api/fleetapi" "github.com/Azure/ARO-HCP/internal/api/kubeapplierapi" "github.com/Azure/ARO-HCP/internal/api/metadataapi" @@ -211,6 +213,119 @@ func TestSyncOnce_WritesCapacityToScheduling(t *testing.T) { assert.Equal(t, "DataCollected", condition.Reason, "condition reason") } +func TestSyncOnce_MirrorsReadyAndNotReadyResourceIDs(t *testing.T) { + ctx := context.Background() + + readyIDs := []string{ + "/subscriptions/00000000-0000-0000-0000-000000000000/resourceGroups/rg1/providers/Microsoft.RedHatOpenShift/hcpOpenShiftClusters/ready-a", + "/subscriptions/00000000-0000-0000-0000-000000000000/resourceGroups/rg1/providers/Microsoft.RedHatOpenShift/hcpOpenShiftClusters/ready-b", + } + notReadyIDs := []string{ + "/subscriptions/00000000-0000-0000-0000-000000000000/resourceGroups/rg1/providers/Microsoft.RedHatOpenShift/hcpOpenShiftClusters/notready-c", + } + + report := &capacityreportv1alpha1.CapacityReport{ + Status: capacityreportv1alpha1.CapacityReportStatus{ + HostedControlPlanes: capacityreportv1alpha1.HostedControlPlanes{ + ReadyResourceIDs: readyIDs, + NotReadyResourceIDs: notReadyIDs, + }, + Conditions: []metav1.Condition{ + { + Type: capacityreportv1alpha1.ConditionTypeReportCurrent, + Status: metav1.ConditionTrue, + }, + }, + }, + } + + desire := buildTestReadDesire(report) + lister := &kubeapplierlistertesting.SliceReadDesireLister{ + Desires: []*kubeapplierapi.ReadDesire{desire}, + } + + fleetDB := fleetcosmosstoragetesting.NewMockFleetDBClient() + + syncer := &capacityReportingSyncer{ + fleetDBClient: fleetDB, + readDesireLister: lister, + } + + err := syncer.SyncOnce(ctx, testKey()) + require.NoError(t, err) + + schedulingCRUD := fleetDB.Stamps().ManagementClusters(testStampIdentifier).Scheduling() + scheduling, err := schedulingCRUD.Get(ctx, fleetapi.SchedulingResourceName) + require.NoError(t, err) + + assert.Equal(t, readyIDs, scheduling.Status.ReadyResourceIDs, "Status.ReadyResourceIDs must mirror the CapacityReport") + assert.Equal(t, notReadyIDs, scheduling.Status.NotReadyResourceIDs, "Status.NotReadyResourceIDs must mirror the CapacityReport") +} + +func TestDropObservedPendingAssignments(t *testing.T) { + ridA := metadataapi.Must(azcorearm.ParseResourceID("/subscriptions/00000000-0000-0000-0000-000000000000/resourceGroups/rg/providers/Microsoft.RedHatOpenShift/hcpOpenShiftClusters/a")) + ridB := metadataapi.Must(azcorearm.ParseResourceID("/subscriptions/00000000-0000-0000-0000-000000000000/resourceGroups/rg/providers/Microsoft.RedHatOpenShift/hcpOpenShiftClusters/b")) + ridC := metadataapi.Must(azcorearm.ParseResourceID("/subscriptions/00000000-0000-0000-0000-000000000000/resourceGroups/rg/providers/Microsoft.RedHatOpenShift/hcpOpenShiftClusters/c")) + pending := []*azcorearm.ResourceID{ridA, ridB, ridC} + + // A observed via Ready, C observed via NotReady, B still pending. + kept := dropObservedPendingAssignments(pending, []string{ridA.String()}, []string{ridC.String()}) + require.Len(t, kept, 1) + assert.Equal(t, strings.ToLower(ridB.String()), strings.ToLower(kept[0].String())) + + // None observed -> unchanged. + assert.Len(t, dropObservedPendingAssignments(pending, nil, nil), 3) + + // All observed -> nil. + assert.Nil(t, dropObservedPendingAssignments(pending, []string{ridA.String(), ridB.String()}, []string{ridC.String()})) + + // Empty pending -> nil (both nil and empty-non-nil inputs). + assert.Nil(t, dropObservedPendingAssignments(nil, []string{ridA.String()}, nil)) + assert.Nil(t, dropObservedPendingAssignments([]*azcorearm.ResourceID{}, []string{ridA.String()}, nil)) +} + +func TestSyncOnce_DropsObservedPendingAssignments(t *testing.T) { + ctx := context.Background() + + ridObserved := metadataapi.Must(azcorearm.ParseResourceID("/subscriptions/00000000-0000-0000-0000-000000000000/resourceGroups/rg/providers/Microsoft.RedHatOpenShift/hcpOpenShiftClusters/observed")) + ridStillPending := metadataapi.Must(azcorearm.ParseResourceID("/subscriptions/00000000-0000-0000-0000-000000000000/resourceGroups/rg/providers/Microsoft.RedHatOpenShift/hcpOpenShiftClusters/pending")) + + report := &capacityreportv1alpha1.CapacityReport{ + Status: capacityreportv1alpha1.CapacityReportStatus{ + HostedControlPlanes: capacityreportv1alpha1.HostedControlPlanes{ + ReadyResourceIDs: []string{ridObserved.String()}, + }, + Conditions: []metav1.Condition{ + {Type: capacityreportv1alpha1.ConditionTypeReportCurrent, Status: metav1.ConditionTrue}, + }, + }, + } + desire := buildTestReadDesire(report) + lister := &kubeapplierlistertesting.SliceReadDesireLister{Desires: []*kubeapplierapi.ReadDesire{desire}} + + fleetDB := fleetcosmosstoragetesting.NewMockFleetDBClient() + // Pre-seed the scheduling doc with two pending reservations: one that the + // report now observes (Ready) and one that is still pending. + _, err := fleetDB.Stamps().ManagementClusters(testStampIdentifier).Scheduling().Create(ctx, &fleetapi.ManagementClusterScheduling{ + CosmosMetadata: coreapi.CosmosMetadata{ + ResourceID: metadataapi.Must(fleetapi.ToManagementClusterSchedulingResourceID(testStampIdentifier)), + PartitionKey: testStampIdentifier, + }, + Status: fleetapi.ManagementClusterSchedulingStatus{ + PendingAssignedClusters: []*azcorearm.ResourceID{ridObserved, ridStillPending}, + }, + }, nil) + require.NoError(t, err) + + syncer := &capacityReportingSyncer{fleetDBClient: fleetDB, readDesireLister: lister} + require.NoError(t, syncer.SyncOnce(ctx, testKey())) + + scheduling, err := fleetDB.Stamps().ManagementClusters(testStampIdentifier).Scheduling().Get(ctx, fleetapi.SchedulingResourceName) + require.NoError(t, err) + require.Len(t, scheduling.Status.PendingAssignedClusters, 1, "observed pending reservation should be dropped") + assert.Equal(t, strings.ToLower(ridStillPending.String()), strings.ToLower(scheduling.Status.PendingAssignedClusters[0].String())) +} + // --- Test doubles for conflict-on-create scenario --- // conflictOnCreateDBClients implements KubeApplierDBClients, returning a diff --git a/internal/api/coreapi/types_serviceprovider_cluster.go b/internal/api/coreapi/types_serviceprovider_cluster.go index 15c4104511e..9e328b83842 100644 --- a/internal/api/coreapi/types_serviceprovider_cluster.go +++ b/internal/api/coreapi/types_serviceprovider_cluster.go @@ -114,6 +114,29 @@ type ServiceProviderClusterSpec struct { // BackupScheduleState is the desired backup scheduling state: Enabled or Disabled. // Default is Enabled. Set to Disabled via Admin API to pause scheduled backups. BackupScheduleState BackupScheduleState `json:"backupScheduleState,omitempty"` + + // ManagementClusterResourceID is the resource ID of the management cluster the + // scheduler has selected for this HCP. This is the scheduler's intent (desired + // placement): nil means placement has not been resolved yet. Downstream + // controllers (cluster creation gating and Cluster Service provision-shard + // pinning) rely on it once set. + // + // It is set once by the PlacementController and is not otherwise mutated. + // Written by: PlacementController + ManagementClusterResourceID *azcorearm.ResourceID `json:"managementClusterResourceID,omitempty"` + + // ManagementClusterPlacementTime records when the PlacementController first + // resolved placement for this HCP — i.e. the moment ManagementClusterResourceID + // was set. It is written once, atomically with the ManagementClusterResourceID + // intent, and preserved across any later re-write/backfill so it marks the time + // of placement rather than the latest reconcile. Nil until the cluster is placed. + // + // It is exposed kube-state-metrics style as the + // backend_cluster_placement_time_seconds gauge (a unix timestamp, not a + // duration); time-to-placement is computed in PromQL against the cluster's + // creation timestamp. + // Written by: PlacementController + ManagementClusterPlacementTime *metav1.Time `json:"managementClusterPlacementTime,omitempty"` } // ServiceProviderClusterSpecVersion contains the desired version information. diff --git a/internal/api/coreapi/zz_generated.deepcopy.go b/internal/api/coreapi/zz_generated.deepcopy.go index 3adb35cc8b0..9639ae0fec5 100644 --- a/internal/api/coreapi/zz_generated.deepcopy.go +++ b/internal/api/coreapi/zz_generated.deepcopy.go @@ -2213,6 +2213,14 @@ func (in *ServiceProviderClusterSpec) DeepCopyInto(out *ServiceProviderClusterSp *out = new(string) **out = **in } + if in.ManagementClusterResourceID != nil { + in, out := &in.ManagementClusterResourceID, &out.ManagementClusterResourceID + *out = DeepCopyResourceID(*in) + } + if in.ManagementClusterPlacementTime != nil { + in, out := &in.ManagementClusterPlacementTime, &out.ManagementClusterPlacementTime + *out = (*in).DeepCopy() + } return } diff --git a/internal/api/fleetapi/types_management_cluster_scheduling.go b/internal/api/fleetapi/types_management_cluster_scheduling.go index 3091f3a7c08..34bd62882cf 100644 --- a/internal/api/fleetapi/types_management_cluster_scheduling.go +++ b/internal/api/fleetapi/types_management_cluster_scheduling.go @@ -18,6 +18,8 @@ import ( corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + azcorearm "github.com/Azure/azure-sdk-for-go/sdk/azcore/arm" + "github.com/Azure/ARO-HCP/internal/api/coreapi" ) @@ -64,11 +66,42 @@ type ManagementClusterSchedulingStatus struct { // Written by: CapacityReportingController. ObservedResources ObservedResources `json:"observedResources"` + // ReadyResourceIDs lists the ARM resource IDs of the HCPs whose + // HostedCluster on this management cluster is ready (Available condition + // True), mirrored verbatim from the CapacityReport CR's + // Status.HostedControlPlanes.ReadyResourceIDs. + // + // +optional + // Written by: CapacityReportingController. + ReadyResourceIDs []string `json:"readyResourceIDs,omitempty"` + + // NotReadyResourceIDs lists the ARM resource IDs of the HCPs whose + // HostedCluster on this management cluster exists but is not ready + // (Available condition not True or missing), mirrored verbatim from the + // CapacityReport CR's Status.HostedControlPlanes.NotReadyResourceIDs. + // + // +optional + // Written by: CapacityReportingController. + NotReadyResourceIDs []string `json:"notReadyResourceIDs,omitempty"` + // ScaleCeiling holds projected capacity limits derived from AKS agent // pool configuration and SKU data. // // Written by: ScaleCeilingReportingController. ScaleCeiling ScaleCeiling `json:"scaleCeiling"` + + // PendingAssignedClusters holds the ARM resource IDs of HCPs the scheduler + // has just placed on this management cluster but whose HostedCluster is not + // yet observed in the CapacityReport (i.e. not yet in ReadyResourceIDs or + // NotReadyResourceIDs). Each pending entry reserves swift-NIC capacity so + // concurrent placement decisions do not overbook a management cluster before + // the workload shows up in the observed capacity data. Entries are removed + // once observed (by CapacityReportingController) or when the reservation + // becomes stale (by PendingCleanupController). + // + // +optional + // Written by: PlacementController, CapacityReportingController, PendingCleanupController. + PendingAssignedClusters []*azcorearm.ResourceID `json:"pendingAssignedClusters,omitempty"` } // ObservedResources reports the observed worker resource state of a diff --git a/internal/api/fleetapi/zz_generated.deepcopy.go b/internal/api/fleetapi/zz_generated.deepcopy.go index d8387d07fd7..dc301c9aa2c 100644 --- a/internal/api/fleetapi/zz_generated.deepcopy.go +++ b/internal/api/fleetapi/zz_generated.deepcopy.go @@ -20,6 +20,7 @@ package fleetapi import ( + azcorearm "github.com/Azure/azure-sdk-for-go/sdk/azcore/arm" corev1 "k8s.io/api/core/v1" v1 "k8s.io/apimachinery/pkg/apis/meta/v1" runtime "k8s.io/apimachinery/pkg/runtime" @@ -262,7 +263,27 @@ func (in *ManagementClusterSchedulingStatus) DeepCopyInto(out *ManagementCluster } } in.ObservedResources.DeepCopyInto(&out.ObservedResources) + if in.ReadyResourceIDs != nil { + in, out := &in.ReadyResourceIDs, &out.ReadyResourceIDs + *out = make([]string, len(*in)) + copy(*out, *in) + } + if in.NotReadyResourceIDs != nil { + in, out := &in.NotReadyResourceIDs, &out.NotReadyResourceIDs + *out = make([]string, len(*in)) + copy(*out, *in) + } in.ScaleCeiling.DeepCopyInto(&out.ScaleCeiling) + if in.PendingAssignedClusters != nil { + in, out := &in.PendingAssignedClusters, &out.PendingAssignedClusters + *out = make([]*azcorearm.ResourceID, len(*in)) + for i := range *in { + if (*in)[i] != nil { + in, out := &(*in)[i], &(*out)[i] + *out = coreapi.DeepCopyResourceID(*in) + } + } + } return } diff --git a/internal/database/cosmosstorage/fleetcosmosstorage/fleet_client.go b/internal/database/cosmosstorage/fleetcosmosstorage/fleet_client.go index 28a48230130..5b043de757f 100644 --- a/internal/database/cosmosstorage/fleetcosmosstorage/fleet_client.go +++ b/internal/database/cosmosstorage/fleetcosmosstorage/fleet_client.go @@ -60,6 +60,7 @@ type ManagementClustersCRUD interface { type FleetGlobalListers interface { Stamps() cosmosstorageutils.GlobalLister[fleetapi.Stamp] ManagementClusters() cosmosstorageutils.GlobalLister[fleetapi.ManagementCluster] + ManagementClusterSchedulings() cosmosstorageutils.GlobalLister[fleetapi.ManagementClusterScheduling] } type cosmosFleetDBClient struct { @@ -181,3 +182,10 @@ func (g *cosmosFleetGlobalListers) ManagementClusters() cosmosstorageutils.Globa ResourceTypes: []azcorearm.ResourceType{fleetapi.ManagementClusterResourceType}, } } + +func (g *cosmosFleetGlobalListers) ManagementClusterSchedulings() cosmosstorageutils.GlobalLister[fleetapi.ManagementClusterScheduling] { + return &cosmosstorageutils.CosmosGlobalLister[fleetapi.ManagementClusterScheduling, cosmosstorageutils.GenericDocument[fleetapi.ManagementClusterScheduling]]{ + ContainerClient: g.container, + ResourceTypes: []azcorearm.ResourceType{fleetapi.ManagementClusterSchedulingResourceType}, + } +} diff --git a/internal/database/cosmosstoragetesting/fleetcosmosstoragetesting/mock_fleet_client.go b/internal/database/cosmosstoragetesting/fleetcosmosstoragetesting/mock_fleet_client.go index 2b699f0823f..0f7ad96d7a3 100644 --- a/internal/database/cosmosstoragetesting/fleetcosmosstoragetesting/mock_fleet_client.go +++ b/internal/database/cosmosstoragetesting/fleetcosmosstoragetesting/mock_fleet_client.go @@ -319,3 +319,10 @@ func (g *mockFleetGlobalListers) ManagementClusters() cosmosstorageutils.GlobalL []azcorearm.ResourceType{fleetapi.ManagementClusterResourceType}, ) } + +func (g *mockFleetGlobalListers) ManagementClusterSchedulings() cosmosstorageutils.GlobalLister[fleetapi.ManagementClusterScheduling] { + return corecosmosstoragetesting.NewMockGlobalLister[fleetapi.ManagementClusterScheduling, cosmosstorageutils.GenericDocument[fleetapi.ManagementClusterScheduling]]( + g.client, + []azcorearm.ResourceType{fleetapi.ManagementClusterSchedulingResourceType}, + ) +} diff --git a/internal/database/informers/fleetinformers/fleet_informers.go b/internal/database/informers/fleetinformers/fleet_informers.go index b821d4a0e99..ca590e76fdc 100644 --- a/internal/database/informers/fleetinformers/fleet_informers.go +++ b/internal/database/informers/fleetinformers/fleet_informers.go @@ -29,8 +29,9 @@ import ( ) const ( - StampRelistDuration = 2 * time.Minute - ManagementClusterRelistDuration = 2 * time.Minute + StampRelistDuration = 2 * time.Minute + ManagementClusterRelistDuration = 2 * time.Minute + ManagementClusterSchedulingRelistDuration = 2 * time.Minute ) // NewStampInformer creates an unstarted SharedIndexInformer for stamps @@ -91,3 +92,32 @@ func NewManagementClusterInformerWithRelistDuration(lister cosmosstorageutils.Gl }, ) } + +// NewManagementClusterSchedulingInformer creates an unstarted SharedIndexInformer +// for management cluster scheduling documents with the default relist duration. +func NewManagementClusterSchedulingInformer(lister cosmosstorageutils.GlobalLister[fleetapi.ManagementClusterScheduling], cosmosClient cosmosstorageutils.ChangeFeedClient) cache.SharedIndexInformer { + return NewManagementClusterSchedulingInformerWithRelistDuration(lister, cosmosClient, ManagementClusterSchedulingRelistDuration) +} + +// NewManagementClusterSchedulingInformerWithRelistDuration creates an unstarted +// SharedIndexInformer for management cluster scheduling documents with a +// configurable relist duration. +func NewManagementClusterSchedulingInformerWithRelistDuration(lister cosmosstorageutils.GlobalLister[fleetapi.ManagementClusterScheduling], cosmosClient cosmosstorageutils.ChangeFeedClient, relistDuration time.Duration) cache.SharedIndexInformer { + lw := informerutils.NewChangeFeedListWatcher[fleetapi.ManagementClusterScheduling, *fleetapi.ManagementClusterScheduling, cosmosstorageutils.GenericDocument[fleetapi.ManagementClusterScheduling]]( + []azcorearm.ResourceType{fleetapi.ManagementClusterSchedulingResourceType}, + utilsclock.RealClock{}, + lister, + cosmosClient, + relistDuration, + "fleet", + ) + + return cache.NewSharedIndexInformerWithOptions( + &informerutils.ListWatchWithoutWatchListSemantics{ListWatch: lw.ToListWatch()}, + &fleetapi.ManagementClusterScheduling{}, + cache.SharedIndexInformerOptions{ + ResyncPeriod: 1 * time.Hour, + ObjectDescription: "ManagementClusterScheduling", + }, + ) +} diff --git a/internal/database/informers/fleetinformers/fleet_types.go b/internal/database/informers/fleetinformers/fleet_types.go index de08dd7f04c..5112a1fcc62 100644 --- a/internal/database/informers/fleetinformers/fleet_types.go +++ b/internal/database/informers/fleetinformers/fleet_types.go @@ -33,14 +33,17 @@ import ( type FleetInformers interface { Stamps() (cache.SharedIndexInformer, fleetlisters.StampLister) ManagementClusters() (cache.SharedIndexInformer, fleetlisters.ManagementClusterLister) + ManagementClusterSchedulings() (cache.SharedIndexInformer, fleetlisters.ManagementClusterSchedulingLister) RunWithContext(ctx context.Context) } type fleetInformers struct { - stampInformer cache.SharedIndexInformer - stampLister fleetlisters.StampLister - managementClusterInformer cache.SharedIndexInformer - managementClusterLister fleetlisters.ManagementClusterLister + stampInformer cache.SharedIndexInformer + stampLister fleetlisters.StampLister + managementClusterInformer cache.SharedIndexInformer + managementClusterLister fleetlisters.ManagementClusterLister + managementClusterSchedulingInformer cache.SharedIndexInformer + managementClusterSchedulingLister fleetlisters.ManagementClusterSchedulingLister } func (f *fleetInformers) Stamps() (cache.SharedIndexInformer, fleetlisters.StampLister) { @@ -51,6 +54,10 @@ func (f *fleetInformers) ManagementClusters() (cache.SharedIndexInformer, fleetl return f.managementClusterInformer, f.managementClusterLister } +func (f *fleetInformers) ManagementClusterSchedulings() (cache.SharedIndexInformer, fleetlisters.ManagementClusterSchedulingLister) { + return f.managementClusterSchedulingInformer, f.managementClusterSchedulingLister +} + // NewFleetInformers creates FleetInformers with default relist durations. func NewFleetInformers(ctx context.Context, globalListers fleetcosmosstorage.FleetGlobalListers, fleetDBClient fleetcosmosstorage.FleetDBClient) FleetInformers { ret := &fleetInformers{} @@ -58,6 +65,8 @@ func NewFleetInformers(ctx context.Context, globalListers fleetcosmosstorage.Fle ret.stampLister = fleetlisters.NewStampLister(ret.stampInformer.GetIndexer()) ret.managementClusterInformer = NewManagementClusterInformer(globalListers.ManagementClusters(), fleetDBClient) ret.managementClusterLister = fleetlisters.NewManagementClusterLister(ret.managementClusterInformer.GetIndexer()) + ret.managementClusterSchedulingInformer = NewManagementClusterSchedulingInformer(globalListers.ManagementClusterSchedulings(), fleetDBClient) + ret.managementClusterSchedulingLister = fleetlisters.NewManagementClusterSchedulingLister(ret.managementClusterSchedulingInformer.GetIndexer()) return ret } @@ -84,6 +93,13 @@ func (f *fleetInformers) RunWithContext(ctx context.Context) { f.managementClusterInformer.RunWithContext(ctx) }() + wg.Add(1) + go func() { + defer utilruntime.HandleCrash() + defer wg.Done() + f.managementClusterSchedulingInformer.RunWithContext(ctx) + }() + <-ctx.Done() wg.Wait() } diff --git a/internal/database/listers/fleetlisters/management_cluster_scheduling_lister.go b/internal/database/listers/fleetlisters/management_cluster_scheduling_lister.go new file mode 100644 index 00000000000..66efcb6e5bb --- /dev/null +++ b/internal/database/listers/fleetlisters/management_cluster_scheduling_lister.go @@ -0,0 +1,55 @@ +// Copyright 2026 Microsoft Corporation +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package fleetlisters + +import ( + "context" + + "k8s.io/client-go/tools/cache" + + "github.com/Azure/ARO-HCP/internal/api/fleetapi" + "github.com/Azure/ARO-HCP/internal/database/listers/listerutils" +) + +// ManagementClusterSchedulingLister lists and gets management cluster scheduling +// documents from an informer's indexer. A scheduling document is a singleton +// child of a management cluster, so it is keyed (and fetched) by the parent +// stamp identifier. +type ManagementClusterSchedulingLister interface { + List(ctx context.Context) ([]*fleetapi.ManagementClusterScheduling, error) + Get(ctx context.Context, stampIdentifier string) (*fleetapi.ManagementClusterScheduling, error) +} + +type informerBasedManagementClusterSchedulingLister struct { + indexer cache.Indexer +} + +// NewManagementClusterSchedulingLister creates a ManagementClusterSchedulingLister +// from a SharedIndexInformer's indexer. +func NewManagementClusterSchedulingLister(indexer cache.Indexer) ManagementClusterSchedulingLister { + return &informerBasedManagementClusterSchedulingLister{ + indexer: indexer, + } +} + +func (l *informerBasedManagementClusterSchedulingLister) List(ctx context.Context) ([]*fleetapi.ManagementClusterScheduling, error) { + return listerutils.ListAll[fleetapi.ManagementClusterScheduling](l.indexer) +} + +// Get retrieves a single management cluster scheduling document by stamp identifier. +func (l *informerBasedManagementClusterSchedulingLister) Get(ctx context.Context, stampIdentifier string) (*fleetapi.ManagementClusterScheduling, error) { + key := fleetapi.ToManagementClusterSchedulingResourceIDString(stampIdentifier) + return listerutils.GetByKey[fleetapi.ManagementClusterScheduling](l.indexer, key) +} diff --git a/internal/database/listertesting/fleetlistertesting/slice_listers.go b/internal/database/listertesting/fleetlistertesting/slice_listers.go index b15a322b710..0babc7aaaa0 100644 --- a/internal/database/listertesting/fleetlistertesting/slice_listers.go +++ b/internal/database/listertesting/fleetlistertesting/slice_listers.go @@ -13,7 +13,7 @@ // limitations under the License. // Package fleetlistertesting provides slice-backed test implementations of the -// fleet listers (Stamp, ManagementCluster). +// fleet listers (Stamp, ManagementCluster, ManagementClusterScheduling). package fleetlistertesting import ( @@ -84,3 +84,25 @@ func (l *SliceManagementClusterLister) GetByCSProvisionShardID(ctx context.Conte return nil, fmt.Errorf("expected at most 1 management cluster for CS provision shard ID %q, got %d", shardID, len(matches)) } } + +// SliceManagementClusterSchedulingLister implements +// fleetlisters.ManagementClusterSchedulingLister backed by a slice. +type SliceManagementClusterSchedulingLister struct { + Schedulings []*fleetapi.ManagementClusterScheduling +} + +var _ fleetlisters.ManagementClusterSchedulingLister = &SliceManagementClusterSchedulingLister{} + +func (l *SliceManagementClusterSchedulingLister) List(ctx context.Context) ([]*fleetapi.ManagementClusterScheduling, error) { + return l.Schedulings, nil +} + +func (l *SliceManagementClusterSchedulingLister) Get(ctx context.Context, stampIdentifier string) (*fleetapi.ManagementClusterScheduling, error) { + key := fleetapi.ToManagementClusterSchedulingResourceIDString(stampIdentifier) + for _, s := range l.Schedulings { + if s.ResourceID != nil && strings.EqualFold(s.ResourceID.String(), key) { + return s, nil + } + } + return nil, cosmosstorageutils.NewNotFoundError() +}