feat(scheduling): backend-driven HCP placement

backend/pkg/app/backend.go

@@ -442,7 +442,7 @@ func (b *Backend) runBackendControllersUnderLeaderElection(ctx context.Context,
 	clusterVersionMetricsController := metrics.NewController(
 		"ClusterVersionMetrics", serviceProviderClusterInformer, clusterVersionMetricsHandler)
 
-	clusterInfoHandler := metrics.NewClusterInfoMetricsHandler(b.options.MetricsRegisterer)
+	clusterInfoHandler := metrics.NewClusterInfoMetricsHandler(b.options.MetricsRegisterer, clusterLister)
 	clusterInfoMetricsController := metrics.NewController(
 		"ClusterInfoMetrics", serviceProviderClusterInformer, clusterInfoHandler)
 
@@ -881,6 +881,18 @@ func (b *Backend) runBackendControllersUnderLeaderElection(ctx context.Context,
 		backendInformers,
 		unionKubeApplierInformers,
 	)
+	placementController := clusterplacement.NewPlacementController(
+		b.options.ResourcesDBClient,
+		b.options.FleetDBClient,
+		managementClusterLister,
+		backendInformers,
+		unionKubeApplierInformers,
+	)
+	pendingCleanupController := clusterplacement.NewPendingCleanupController(
+		b.options.FleetDBClient,
+		serviceProviderClusterLister,
+		fleetInformers,
+	)
 
 	nodePoolClusterServiceCreateController := nodepoolcreation.NewNodePoolClusterServiceCreateController(
 		b.options.ResourcesDBClient,
@@ -952,6 +964,7 @@ func (b *Backend) runBackendControllersUnderLeaderElection(ctx context.Context,
 	clusterClusterServiceCreateController := clustercreation.NewClusterClusterServiceCreateController(
 		b.options.ResourcesDBClient,
 		b.options.ClustersServiceClient,
+		managementClusterLister,
 		backendInformers,
 	)
 
@@ -1127,6 +1140,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)     // multiple workers are safe: optimistic writes to per-MC scheduling docs make assignments to different management clusters non-conflicting, and same-cluster keys are serialized by the workqueue
+				go pendingCleanupController.Run(ctx, 1) // single worker: sweeps pending reservations per management cluster
 				go cosmosMigrationController.Run(ctx, 5)
 				go virtualMachineResourceSKUsCachedReaderController.Run(ctx, 20)
 				go backupScheduleController.Run(ctx, 20)

backend/pkg/controllers/cluster/creation/cluster_cluster_service_create_controller.go

@@ -29,31 +29,35 @@ 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)
 
 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/<stampIdentifier>/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
+}

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{

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
 	}