feat: update hub-agent to use custom cluster affinity plugin in scheduler when property checker is enabled (#1221)

Author: britaniar

cmd/hubagent/options/azureoptions.go

@@ -0,0 +1,31 @@
+//
+//Copyright (c) Microsoft Corporation.
+//Licensed under the MIT license.
+
+package options
+
+import (
+	"flag"
+)
+
+// AzurePropertyCheckerOptions holds the options for the Azure property checker.
+type AzurePropertyCheckerOptions struct {
+	// IsEnabled indicates whether the Azure property checker is enabled.
+	IsEnabled bool
+	// ComputeServiceAddressWithBasePath is the address of the Azure compute service with base path.
+	ComputeServiceAddressWithBasePath string
+}
+
+// NewAzurePropertyCheckerOptions creates a new AzurePropertyCheckerOptions with default values.
+func NewAzurePropertyCheckerOptions() AzurePropertyCheckerOptions {
+	return AzurePropertyCheckerOptions{
+		IsEnabled:                         false,
+		ComputeServiceAddressWithBasePath: "http://localhost:8421/compute",
+	}
+}
+
+// AddFlags adds flags for Azure Property Checker options to the given FlagSet.
+func (o AzurePropertyCheckerOptions) AddFlags(flags *flag.FlagSet) {
+	flags.BoolVar(&o.IsEnabled, "azure-property-checker-enabled", o.IsEnabled, "Enable Azure property checker for validating Azure-specific cluster properties.")
+	flags.StringVar(&o.ComputeServiceAddressWithBasePath, "azure-compute-server-address", o.ComputeServiceAddressWithBasePath, "The address of the Azure compute service with base path.")
+}

cmd/hubagent/options/options.go

@@ -112,6 +112,8 @@ type Options struct {
 	ResourceSnapshotCreationMinimumInterval time.Duration
 	// ResourceChangesCollectionDuration is the duration for collecting resource changes into one snapshot.
 	ResourceChangesCollectionDuration time.Duration
+	// AzurePropertyCheckerOpts contains options for Azure property checker
+	AzurePropertyCheckerOpts AzurePropertyCheckerOptions
 }
 
 // NewOptions builds an empty options.
@@ -134,6 +136,7 @@ func NewOptions() *Options {
 		PprofPort:                               6065,
 		ResourceSnapshotCreationMinimumInterval: 30 * time.Second,
 		ResourceChangesCollectionDuration:       15 * time.Second,
+		AzurePropertyCheckerOpts:                NewAzurePropertyCheckerOptions(),
 	}
 }
 
@@ -185,4 +188,5 @@ func (o *Options) AddFlags(flags *flag.FlagSet) {
 	flags.DurationVar(&o.ResourceChangesCollectionDuration, "resource-changes-collection-duration", 15*time.Second,
 		"The duration for collecting resource changes into one snapshot. The default is 15 seconds, which means that the controller will collect resource changes for 15 seconds before creating a resource snapshot.")
 	o.RateLimiterOpts.AddFlags(flags)
+	o.AzurePropertyCheckerOpts.AddFlags(flags)
 }

cmd/hubagent/workload/setup.go

@@ -33,6 +33,8 @@ import (
 	clusterv1beta1 "go.goms.io/fleet/apis/cluster/v1beta1"
 	placementv1beta1 "go.goms.io/fleet/apis/placement/v1beta1"
 	"go.goms.io/fleet/cmd/hubagent/options"
+	"go.goms.io/fleet/pkg/clients/azure/compute"
+	"go.goms.io/fleet/pkg/clients/httputil"
 	"go.goms.io/fleet/pkg/controllers/bindingwatcher"
 	"go.goms.io/fleet/pkg/controllers/clusterinventory/clusterprofile"
 	"go.goms.io/fleet/pkg/controllers/clusterresourceplacementeviction"
@@ -45,10 +47,12 @@ import (
 	"go.goms.io/fleet/pkg/controllers/schedulingpolicysnapshot"
 	"go.goms.io/fleet/pkg/controllers/updaterun"
 	"go.goms.io/fleet/pkg/controllers/workgenerator"
+	"go.goms.io/fleet/pkg/propertychecker/azure"
 	"go.goms.io/fleet/pkg/resourcewatcher"
 	"go.goms.io/fleet/pkg/scheduler"
 	"go.goms.io/fleet/pkg/scheduler/clustereligibilitychecker"
 	"go.goms.io/fleet/pkg/scheduler/framework"
+	"go.goms.io/fleet/pkg/scheduler/framework/plugins/clusteraffinity"
 	"go.goms.io/fleet/pkg/scheduler/profile"
 	"go.goms.io/fleet/pkg/scheduler/queue"
 	schedulerbindingwatcher "go.goms.io/fleet/pkg/scheduler/watchers/binding"
@@ -354,8 +358,22 @@ func SetupControllers(ctx context.Context, wg *sync.WaitGroup, mgr ctrl.Manager,
 
 		// Set up the scheduler
 		klog.Info("Setting up scheduler")
-		defaultProfile := profile.NewDefaultProfile()
-		defaultFramework := framework.NewFramework(defaultProfile, mgr)
+		schedulerProfile := profile.NewDefaultProfile()
+		if opts.AzurePropertyCheckerOpts.IsEnabled {
+			klog.Info("Azure property checker is enabled for cluster property validation")
+			client, err := compute.NewAttributeBasedVMSizeRecommenderClient(opts.AzurePropertyCheckerOpts.ComputeServiceAddressWithBasePath, httputil.DefaultClientForAzure)
+			if err != nil {
+				klog.ErrorS(err, "Unable to create Azure vm size recommender client")
+				return err
+			}
+			klog.Info("Setting up cluster affinity plugin with Azure property checker")
+			clusterAffinityPlugin := clusteraffinity.New(clusteraffinity.WithPropertyChecker(azure.NewPropertyChecker(*client)))
+			profileOpts := profile.Options{
+				ClusterAffinityPlugin: &clusterAffinityPlugin,
+			}
+			schedulerProfile = profile.NewProfile(profileOpts)
+		}
+		defaultFramework := framework.NewFramework(schedulerProfile, mgr)
 		defaultSchedulingQueue := queue.NewSimplePlacementSchedulingQueue(
 			queue.WithName(schedulerQueueName),
 		)