feat: Add webhook for ClusterResourcePlacement (#249)

* feat: Add webhook for ClusterResourcePlacement

apis/v1alpha1/clusterresourceplacement_webhook.go

@@ -0,0 +1,96 @@
+/*
+Copyright (c) Microsoft Corporation.
+Licensed under the MIT license.
+*/
+
+package v1alpha1
+
+import (
+	"fmt"
+
+	"github.com/pkg/errors"
+	"k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	apiErrors "k8s.io/apimachinery/pkg/util/errors"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/webhook"
+
+	"go.goms.io/fleet/pkg/utils/informer"
+)
+
+var ResourceInformer informer.Manager
+var restMapper meta.RESTMapper
+
+func (c *ClusterResourcePlacement) SetupWebhookWithManager(mgr ctrl.Manager) error {
+	restMapper = mgr.GetRESTMapper()
+	return ctrl.NewWebhookManagedBy(mgr).
+		For(c).
+		Complete()
+}
+
+var _ webhook.Validator = &ClusterResourcePlacement{}
+
+func (c *ClusterResourcePlacement) ValidateCreate() error {
+	return ValidateClusterResourcePlacement(c)
+}
+
+func (c *ClusterResourcePlacement) ValidateUpdate(old runtime.Object) error {
+	// TODO: validate changes against old if needed
+	return ValidateClusterResourcePlacement(c)
+}
+
+func (c *ClusterResourcePlacement) ValidateDelete() error {
+	// do nothing for delete request
+	return nil
+}
+
+// ValidateClusterResourcePlacement validate a ClusterResourcePlacement object
+func ValidateClusterResourcePlacement(clusterResourcePlacement *ClusterResourcePlacement) error {
+	allErr := make([]error, 0)
+
+	for _, selector := range clusterResourcePlacement.Spec.ResourceSelectors {
+		if selector.LabelSelector != nil {
+			if len(selector.Name) != 0 {
+				allErr = append(allErr, fmt.Errorf("the labelSelector and name fields are mutually exclusive in selector %+v", selector))
+			}
+			if _, err := metav1.LabelSelectorAsSelector(selector.LabelSelector); err != nil {
+				allErr = append(allErr, errors.Wrap(err, fmt.Sprintf("the labelSelector in resource selector %+v is invalid", selector)))
+			}
+		}
+	}
+
+	if clusterResourcePlacement.Spec.Policy != nil && clusterResourcePlacement.Spec.Policy.Affinity != nil &&
+		clusterResourcePlacement.Spec.Policy.Affinity.ClusterAffinity != nil {
+		for _, selector := range clusterResourcePlacement.Spec.Policy.Affinity.ClusterAffinity.ClusterSelectorTerms {
+			if _, err := metav1.LabelSelectorAsSelector(&selector.LabelSelector); err != nil {
+				allErr = append(allErr, errors.Wrap(err, fmt.Sprintf("the labelSelector in cluster selector %+v is invalid", selector)))
+			}
+		}
+	}
+
+	// we leverage the informermanager in the changedetector controller to do the resource scope validation
+	if ResourceInformer == nil {
+		allErr = append(allErr, fmt.Errorf("cannot perform resource scope check for now, please retry"))
+	} else {
+		for _, selector := range clusterResourcePlacement.Spec.ResourceSelectors {
+			gk := schema.GroupKind{
+				Group: selector.Group,
+				Kind:  selector.Kind,
+			}
+
+			restMapping, err := restMapper.RESTMapping(gk, selector.Version)
+			if err != nil {
+				allErr = append(allErr, errors.Wrap(err, fmt.Sprintf("failed to get GVR of GVK (%s/%s/%s), please retry if the GVK is valid", selector.Group, selector.Version, selector.Kind)))
+				continue
+			}
+
+			if !ResourceInformer.IsClusterScopedResources(restMapping.Resource) {
+				allErr = append(allErr, fmt.Errorf("the resource is not found in schema (please retry) or it is not a cluster scoped resource: %v", restMapping.Resource))
+			}
+		}
+	}
+
+	return apiErrors.NewAggregate(allErr)
+}

cmd/hubagent/workload/setup.go

@@ -24,6 +24,7 @@ import (
 	"go.goms.io/fleet/pkg/resourcewatcher"
 	"go.goms.io/fleet/pkg/utils"
 	"go.goms.io/fleet/pkg/utils/controller"
+	"go.goms.io/fleet/pkg/utils/informer"
 )
 
 const (
@@ -79,7 +80,8 @@ func SetupControllers(ctx context.Context, mgr ctrl.Manager, config *rest.Config
 	}
 
 	// the manager for all the dynamically created informers
-	dynamicInformerManager := utils.NewInformerManager(dynamicClient, opts.ResyncPeriod.Duration, ctx.Done())
+	dynamicInformerManager := informer.NewInformerManager(dynamicClient, opts.ResyncPeriod.Duration, ctx.Done())
+	fleetv1alpha1.ResourceInformer = dynamicInformerManager // webhook needs this to check resource scope
 
 	// Set up  a custom controller to reconcile cluster resource placement
 	klog.Info("Setting up clusterResourcePlacement controller")

pkg/controllers/clusterresourceplacement/placement_controller.go

@@ -24,6 +24,7 @@ import (
 	fleetv1alpha1 "go.goms.io/fleet/apis/v1alpha1"
 	"go.goms.io/fleet/pkg/utils"
 	"go.goms.io/fleet/pkg/utils/controller"
+	"go.goms.io/fleet/pkg/utils/informer"
 	"go.goms.io/fleet/pkg/utils/validator"
 )
 
@@ -41,7 +42,7 @@ var (
 // Reconciler reconciles a cluster resource placement object
 type Reconciler struct {
 	// the informer contains the cache for all the resources we need
-	InformerManager utils.InformerManager
+	InformerManager informer.Manager
 
 	RestMapper meta.RESTMapper
 

pkg/controllers/memberclusterplacement/membercluster_controller.go

@@ -20,12 +20,13 @@ import (
 	fleetv1alpha1 "go.goms.io/fleet/apis/v1alpha1"
 	"go.goms.io/fleet/pkg/utils"
 	"go.goms.io/fleet/pkg/utils/controller"
+	"go.goms.io/fleet/pkg/utils/informer"
 )
 
 // Reconciler reconciles a MemberCluster object
 type Reconciler struct {
 	// the informer contains the cache for all the resources we need
-	InformerManager utils.InformerManager
+	InformerManager informer.Manager
 
 	// PlacementController maintains a rate limited queue which used to store
 	// the name of the clusterResourcePlacement and a reconcile function to consume the items in queue.

pkg/controllers/resourcechange/resourcechange_controller.go

@@ -26,6 +26,7 @@ import (
 	fleetv1alpha1 "go.goms.io/fleet/apis/v1alpha1"
 	"go.goms.io/fleet/pkg/utils"
 	"go.goms.io/fleet/pkg/utils/controller"
+	"go.goms.io/fleet/pkg/utils/informer"
 	"go.goms.io/fleet/pkg/utils/keys"
 	"go.goms.io/fleet/pkg/utils/validator"
 )
@@ -39,7 +40,7 @@ type Reconciler struct {
 	RestMapper meta.RESTMapper
 
 	// InformerManager holds all the informers that we can use to read from
-	InformerManager utils.InformerManager
+	InformerManager informer.Manager
 
 	// PlacementController exposes the placement queue for the reconciler to push to
 	PlacementController controller.Controller

pkg/resourcewatcher/change_dector.go

@@ -22,6 +22,7 @@ import (
 
 	"go.goms.io/fleet/pkg/utils"
 	"go.goms.io/fleet/pkg/utils/controller"
+	"go.goms.io/fleet/pkg/utils/informer"
 	"go.goms.io/fleet/pkg/utils/keys"
 )
 
@@ -52,7 +53,7 @@ type ChangeDetector struct {
 	MemberClusterPlacementController controller.Controller
 
 	// InformerManager manages all the dynamic informers created by the discovery client
-	InformerManager utils.InformerManager
+	InformerManager informer.Manager
 
 	// DisabledResourceConfig contains all the api resources that we won't select
 	DisabledResourceConfig *utils.DisabledResourceConfig
@@ -80,7 +81,7 @@ func (d *ChangeDetector) Start(ctx context.Context) error {
 	clusterPlacementEventHandler := newHandlerOnEvents(d.onClusterResourcePlacementAdded,
 		d.onClusterResourcePlacementUpdated, d.onClusterResourcePlacementDeleted)
 	d.InformerManager.AddStaticResource(
-		utils.APIResourceMeta{
+		informer.APIResourceMeta{
 			GroupVersionResource: utils.ClusterResourcePlacementGVR,
 			IsClusterScoped:      true,
 		}, clusterPlacementEventHandler)
@@ -89,7 +90,7 @@ func (d *ChangeDetector) Start(ctx context.Context) error {
 	// the placement queue. We don't need to handle the add event as they are placed by the placement controller.
 	workEventHandler := newHandlerOnEvents(nil, d.onWorkUpdated, d.onWorkDeleted)
 	d.InformerManager.AddStaticResource(
-		utils.APIResourceMeta{
+		informer.APIResourceMeta{
 			GroupVersionResource: utils.WorkGVR,
 			IsClusterScoped:      false,
 		}, workEventHandler)
@@ -98,7 +99,7 @@ func (d *ChangeDetector) Start(ctx context.Context) error {
 	// delete event as the work resources in this cluster will all get deleted which will trigger placement reconcile.
 	memberClusterEventHandler := newHandlerOnEvents(nil, d.onMemberClusterUpdated, nil)
 	d.InformerManager.AddStaticResource(
-		utils.APIResourceMeta{
+		informer.APIResourceMeta{
 			GroupVersionResource: utils.MemberClusterGVR,
 			IsClusterScoped:      true,
 		}, memberClusterEventHandler)
@@ -144,7 +145,7 @@ func (d *ChangeDetector) discoverAPIResourcesLoop(ctx context.Context, period ti
 // discoverResources goes through all the api resources in the cluster and create informers on selected types
 func (d *ChangeDetector) discoverResources(dynamicResourceEventHandler cache.ResourceEventHandler) {
 	newResources, err := d.getWatchableResources()
-	var dynamicResources []utils.APIResourceMeta
+	var dynamicResources []informer.APIResourceMeta
 	if err != nil {
 		klog.ErrorS(err, "Failed to get all the api resources from the cluster")
 	}

pkg/resourcewatcher/resource_collector.go

@@ -11,14 +11,14 @@ import (
 	"k8s.io/client-go/discovery"
 	"k8s.io/klog/v2"
 
-	"go.goms.io/fleet/pkg/utils"
+	"go.goms.io/fleet/pkg/utils/informer"
 )
 
 // getWatchableResources returns all api resources from discoveryClient that we can watch.
 // More specifically, all api resources which support the 'list', and 'watch' verbs.
 // All discovery errors are considered temporary. Upon encountering any error,
 // getWatchableResources will log and return any discovered resources it was able to process (which may be none).
-func (d *ChangeDetector) getWatchableResources() ([]utils.APIResourceMeta, error) {
+func (d *ChangeDetector) getWatchableResources() ([]informer.APIResourceMeta, error) {
 	// Get all the resources this cluster has. We only need to care about the preferred version as the informers watch
 	// the preferred version will get watch event for resources on the other versions since there is only one version in etcd.
 	allResources, discoverError := d.DiscoveryClient.ServerPreferredResources()
@@ -35,7 +35,7 @@ func (d *ChangeDetector) getWatchableResources() ([]utils.APIResourceMeta, error
 		return nil, discoverError
 	}
 
-	watchableGroupVersionResources := make([]utils.APIResourceMeta, 0)
+	watchableGroupVersionResources := make([]informer.APIResourceMeta, 0)
 
 	// This is extracted from discovery.GroupVersionResources to only watch watchable resources
 	watchableResources := discovery.FilteredBy(discovery.SupportsAllVerbs{Verbs: []string{"list", "watch"}}, allResources)
@@ -48,7 +48,7 @@ func (d *ChangeDetector) getWatchableResources() ([]utils.APIResourceMeta, error
 		}
 		for i := range rl.APIResources {
 			gvr := schema.GroupVersionResource{Group: gv.Group, Version: gv.Version, Resource: rl.APIResources[i].Name}
-			watchableGroupVersionResources = append(watchableGroupVersionResources, utils.APIResourceMeta{
+			watchableGroupVersionResources = append(watchableGroupVersionResources, informer.APIResourceMeta{
 				GroupVersionResource: gvr,
 				IsClusterScoped:      !rl.APIResources[i].Namespaced,
 			})

pkg/utils/common.go

@@ -6,7 +6,6 @@ Licensed under the MIT license.
 package utils
 
 import (
-	"context"
 	"crypto/rand"
 	"fmt"
 	"log"
@@ -27,6 +26,7 @@ import (
 	workv1alpha1 "sigs.k8s.io/work-api/pkg/apis/v1alpha1"
 
 	fleetv1alpha1 "go.goms.io/fleet/apis/v1alpha1"
+	"go.goms.io/fleet/pkg/utils/informer"
 )
 
 const (
@@ -155,25 +155,6 @@ func RandStr() string {
 	return string(ret)
 }
 
-// ContextForChannel derives a child context from a parent channel.
-//
-// The derived context's Done channel is closed when the returned cancel function
-// is called or when the parent channel is closed, whichever happens first.
-//
-// Note the caller must *always* call the CancelFunc, otherwise resources may be leaked.
-func ContextForChannel(parentCh <-chan struct{}) (context.Context, context.CancelFunc) {
-	ctx, cancel := context.WithCancel(context.Background())
-
-	go func() {
-		select {
-		case <-parentCh:
-			cancel()
-		case <-ctx.Done():
-		}
-	}()
-	return ctx, cancel
-}
-
 // CheckCRDInstalled checks if the custom resource definition is installed
 func CheckCRDInstalled(discoveryClient discovery.DiscoveryInterface, gvk schema.GroupVersionKind) error {
 	startTime := time.Now()
@@ -197,7 +178,7 @@ func CheckCRDInstalled(discoveryClient discovery.DiscoveryInterface, gvk schema.
 }
 
 // ShouldPropagateObj decides if one should propagate the object
-func ShouldPropagateObj(informerManager InformerManager, uObj *unstructured.Unstructured) (bool, error) {
+func ShouldPropagateObj(informerManager informer.Manager, uObj *unstructured.Unstructured) (bool, error) {
 	// TODO:  add more special handling for different resource kind
 	switch uObj.GroupVersionKind() {
 	case corev1.SchemeGroupVersion.WithKind("ConfigMap"):

pkg/utils/informermanager.go → pkg/utils/informer/informermanager.go

@@ -3,7 +3,7 @@ Copyright (c) Microsoft Corporation.
 Licensed under the MIT license.
 */
 
-package utils
+package informer
 
 import (
 	"context"
@@ -20,7 +20,7 @@ import (
 
 // InformerManager manages dynamic shared informer for all resources, include Kubernetes resource and
 // custom resources defined by CustomResourceDefinition.
-type InformerManager interface {
+type Manager interface {
 	// AddDynamicResources builds a dynamicInformer for each resource in the resources list with the event handler.
 	// A resource is dynamic if its definition can be created/deleted/updated during runtime.
 	// Normally, it is a custom resource that is installed by users. The handler should not be nil.
@@ -61,7 +61,7 @@ type InformerManager interface {
 
 // NewInformerManager constructs a new instance of informerManagerImpl.
 // defaultResync with value '0' means no re-sync.
-func NewInformerManager(client dynamic.Interface, defaultResync time.Duration, parentCh <-chan struct{}) InformerManager {
+func NewInformerManager(client dynamic.Interface, defaultResync time.Duration, parentCh <-chan struct{}) Manager {
 	// TODO: replace this with plain context
 	ctx, cancel := ContextForChannel(parentCh)
 	return &informerManagerImpl{
@@ -211,3 +211,22 @@ func (s *informerManagerImpl) IsClusterScopedResources(resource schema.GroupVers
 func (s *informerManagerImpl) Stop() {
 	s.cancel()
 }
+
+// ContextForChannel derives a child context from a parent channel.
+//
+// The derived context's Done channel is closed when the returned cancel function
+// is called or when the parent channel is closed, whichever happens first.
+//
+// Note the caller must *always* call the CancelFunc, otherwise resources may be leaked.
+func ContextForChannel(parentCh <-chan struct{}) (context.Context, context.CancelFunc) {
+	ctx, cancel := context.WithCancel(context.Background())
+
+	go func() {
+		select {
+		case <-parentCh:
+			cancel()
+		case <-ctx.Done():
+		}
+	}()
+	return ctx, cancel
+}

pkg/webhook/add_clusterresourceplacement.go

@@ -0,0 +1,15 @@
+/*
+Copyright (c) Microsoft Corporation.
+Licensed under the MIT license.
+*/
+
+package webhook
+
+import (
+	"go.goms.io/fleet/pkg/webhook/clusterresourceplacement"
+)
+
+func init() {
+	// AddToManagerFuncs is a list of functions to create webhook and add them to a manager.
+	AddToManagerFuncs = append(AddToManagerFuncs, clusterresourceplacement.Add)
+}

pkg/webhook/clusterresourceplacement/clusterresourceplacement_webhook.go

@@ -0,0 +1,16 @@
+/*
+Copyright (c) Microsoft Corporation.
+Licensed under the MIT license.
+*/
+
+package clusterresourceplacement
+
+import (
+	"sigs.k8s.io/controller-runtime/pkg/manager"
+
+	fleetv1alpha1 "go.goms.io/fleet/apis/v1alpha1"
+)
+
+func Add(mgr manager.Manager) error {
+	return (&fleetv1alpha1.ClusterResourcePlacement{}).SetupWebhookWithManager(mgr)
+}