feat: add HCP scheduling PlacementController (phase 1)

backend/pkg/app/backend.go

@@ -881,6 +881,12 @@ func (b *Backend) runBackendControllersUnderLeaderElection(ctx context.Context,
 		backendInformers,
 		unionKubeApplierInformers,
 	)
+	placementController := clusterplacement.NewPlacementController(
+		b.options.ResourcesDBClient,
+		managementClusterLister,
+		backendInformers,
+		unionKubeApplierInformers,
+	)
 
 	nodePoolClusterServiceCreateController := nodepoolcreation.NewNodePoolClusterServiceCreateController(
 		b.options.ResourcesDBClient,
@@ -952,6 +958,7 @@ func (b *Backend) runBackendControllersUnderLeaderElection(ctx context.Context,
 	clusterClusterServiceCreateController := clustercreation.NewClusterClusterServiceCreateController(
 		b.options.ResourcesDBClient,
 		b.options.ClustersServiceClient,
+		managementClusterLister,
 		backendInformers,
 	)
 
@@ -1127,6 +1134,7 @@ 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, 1) // single worker: placement selection reads all MCs/SPCs and must not race itself
 				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

@@ -20,40 +20,47 @@ import (
 	"strings"
 	"time"
 
+	azcorearm "github.com/Azure/azure-sdk-for-go/sdk/azcore/arm"
+
 	arohcpv1alpha1 "github.com/openshift-online/ocm-sdk-go/arohcp/v1alpha1"
 
 	"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/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(
@@ -235,7 +242,12 @@ 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)
 
-	csClusterBuilder, err := ocm.BuildCSCluster(cluster.ID, tenantID, cluster, nil, nil, serviceProviderCluster)
+	requiredProperties, err := c.provisionShardRequiredProperties(ctx, serviceProviderCluster)
+	if err != nil {
+		return nil, utils.TrackError(err)
+	}
+
+	csClusterBuilder, err := ocm.BuildCSCluster(cluster.ID, tenantID, cluster, requiredProperties, nil, serviceProviderCluster)
 	if err != nil {
 		return nil, utils.TrackError(fmt.Errorf("failed to build CS cluster: %w", err))
 	}
@@ -253,3 +265,48 @@ func (c *clusterClusterServiceCreateSyncer) createClusterServiceCluster(ctx cont
 
 	return result, nil
 }
+
+// provisionShardRequiredProperties resolves the Cluster Service provision shard
+// for the management cluster the scheduler pinned on
+// ServiceProviderCluster.Spec.ManagementClusterResourceID, and returns it as the
+// requiredProperties map passed to BuildCSCluster so the new CS cluster is
+// created on the correct provision shard.
+func (c *clusterClusterServiceCreateSyncer) provisionShardRequiredProperties(ctx context.Context, serviceProviderCluster *coreapi.ServiceProviderCluster) (map[string]string, error) {
+	managementClusterResourceID := serviceProviderCluster.Spec.ManagementClusterResourceID
+	if managementClusterResourceID == nil {
+		return nil, fmt.Errorf("ServiceProviderCluster has no Spec.ManagementClusterResourceID; placement is not resolved")
+	}
+
+	managementCluster, err := c.lookupManagementCluster(ctx, managementClusterResourceID)
+	if err != nil {
+		return nil, err
+	}
+
+	if managementCluster.Status.ClusterServiceProvisionShardID == nil {
+		return nil, fmt.Errorf("management cluster %q has no ClusterServiceProvisionShardID", managementClusterResourceID.String())
+	}
+	provisionShardID := managementCluster.Status.ClusterServiceProvisionShardID.ID()
+
+	return map[string]string{ocm.CSPropertyProvisionShardID: provisionShardID}, nil
+}
+
+// lookupManagementCluster returns the ManagementCluster whose resource ID
+// matches the given ID from the fleet lister cache.
+func (c *clusterClusterServiceCreateSyncer) lookupManagementCluster(ctx context.Context, resourceID *azcorearm.ResourceID) (*fleetapi.ManagementCluster, error) {
+	managementClusters, err := c.managementClusterLister.List(ctx)
+	if err != nil {
+		return nil, utils.TrackError(fmt.Errorf("failed to list management clusters: %w", err))
+	}
+
+	want := strings.ToLower(resourceID.String())
+	for _, managementCluster := range managementClusters {
+		if managementCluster == nil || managementCluster.ResourceID == nil {
+			continue
+		}
+		if strings.ToLower(managementCluster.ResourceID.String()) == want {
+			return managementCluster, nil
+		}
+	}
+
+	return nil, fmt.Errorf("management cluster %q not found", resourceID.String())
+}

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,38 @@ const (
 	testClusterUID          = "00000000-0000-0000-0000-000000000000"
 	// testManagedResourceGroup must match what coreapitesting.MinimumValidClusterTestCase() sets.
 	testManagedResourceGroup = "testManagedResourceGroup"
+	// testStampIdentifier and testProvisionShardID drive the management-cluster
+	// placement / provision-shard-pinning test fixtures.
+	testStampIdentifier  = "1"
+	testProvisionShardID = "shard-abc123"
 )
 
+// testManagementClusterResourceID returns the resource ID of the test management cluster.
+func testManagementClusterResourceID() *azcorearm.ResourceID {
+	return metadataapi.Must(fleetapi.ToManagementClusterResourceID(testStampIdentifier))
+}
+
+// testProvisionShardHREF is the CS provision shard HREF whose ID() is testProvisionShardID.
+func testProvisionShardHREF() string {
+	return "/api/aro_hcp/v1alpha1/provision_shards/" + testProvisionShardID
+}
+
+// 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(testProvisionShardHREF()))),
+		},
+	}
+}
+
 // testClusterResourceID builds the ARM resource ID for the test cluster.
 func testClusterResourceID() *azcorearm.ResourceID {
 	return metadataapi.Must(azcorearm.ParseResourceID(
@@ -128,6 +160,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 +175,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 +189,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.Properties()[ocm.CSPropertyProvisionShardID], "PostCluster should pin the provision shard from the placed management cluster")
 						csCluster, err := arohcpv1alpha1.NewCluster().
 							ID(pendingClusterServiceID.ID()).
 							HREF(testClusterServiceIDStr).
@@ -232,7 +268,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 +322,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
 	}