feat: add sync poolservice and webhook

Signed-off-by: wangxye <[email protected]>

charts/yurt-manager/templates/yurt-manager-auto-generated.yaml

@@ -672,6 +672,27 @@ webhooks:
     resources:
     - pods
   sideEffects: None
+- admissionReviewVersions:
+  - v1
+  - v1beta1
+  clientConfig:
+    service:
+      name: yurt-manager-webhook-service
+      namespace: {{ .Release.Namespace }}
+      path: /mutate-network-openyurt-io-poolservice
+  failurePolicy: Fail
+  name: mutate.network.v1alpha1.poolservice.openyurt.io
+  rules:
+  - apiGroups:
+    - network.openyurt.io
+    apiVersions:
+    - v1alpha1
+    operations:
+    - CREATE
+    - UPDATE
+    resources:
+    - poolservices
+  sideEffects: None
 - admissionReviewVersions:
   - v1
   - v1beta1

cmd/yurt-manager/app/options/options.go

@@ -40,6 +40,7 @@ type YurtManagerOptions struct {
 	NodeLifeCycleController    *NodeLifecycleControllerOptions
 	NodeBucketController       *NodeBucketControllerOptions
 	LoadBalancerSetController  *LoadBalancerSetControllerOptions
+	VipLoadBalancerController  *VipLoadBalancerControllerOptions
 }
 
 // NewYurtManagerOptions creates a new YurtManagerOptions with a default config.
@@ -61,6 +62,7 @@ func NewYurtManagerOptions() (*YurtManagerOptions, error) {
 		NodeLifeCycleController:    NewNodeLifecycleControllerOptions(),
 		NodeBucketController:       NewNodeBucketControllerOptions(),
 		LoadBalancerSetController:  NewLoadBalancerSetControllerOptions(),
+		VipLoadBalancerController:  NewVipLoadBalancerControllerOptions(),
 	}
 
 	return &s, nil
@@ -82,6 +84,7 @@ func (y *YurtManagerOptions) Flags(allControllers, disabledByDefaultControllers
 	y.NodeLifeCycleController.AddFlags(fss.FlagSet("nodelifecycle controller"))
 	y.NodeBucketController.AddFlags(fss.FlagSet("nodebucket controller"))
 	y.LoadBalancerSetController.AddFlags(fss.FlagSet("loadbalancerset controller"))
+	y.VipLoadBalancerController.AddFlags(fss.FlagSet("viploadbalancer controller"))
 	return fss
 }
 
@@ -102,6 +105,7 @@ func (y *YurtManagerOptions) Validate(allControllers []string, controllerAliases
 	errs = append(errs, y.NodeLifeCycleController.Validate()...)
 	errs = append(errs, y.NodeBucketController.Validate()...)
 	errs = append(errs, y.LoadBalancerSetController.Validate()...)
+	errs = append(errs, y.VipLoadBalancerController.Validate()...)
 	return utilerrors.NewAggregate(errs)
 }
 
@@ -149,6 +153,9 @@ func (y *YurtManagerOptions) ApplyTo(c *config.Config, controllerAliases map[str
 	if err := y.LoadBalancerSetController.ApplyTo(&c.ComponentConfig.LoadBalancerSetController); err != nil {
 		return err
 	}
+	if err := y.VipLoadBalancerController.ApplyTo(&c.ComponentConfig.VipLoadBalancerController); err != nil {
+		return err
+	}
 	return nil
 }
 

cmd/yurt-manager/app/options/vipbalancercontroller.go

@@ -0,0 +1,64 @@
+/*
+Copyright 2024 The OpenYurt Authors.
+
+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 options
+
+import (
+	"github.com/spf13/pflag"
+
+	"github.com/openyurtio/openyurt/pkg/yurtmanager/controller/loadbalancerset/viploadbalancer/config"
+)
+
+type VipLoadBalancerControllerOptions struct {
+	*config.VipLoadBalancerControllerConfiguration
+}
+
+func NewVipLoadBalancerControllerOptions() *VipLoadBalancerControllerOptions {
+	return &VipLoadBalancerControllerOptions{
+		&config.VipLoadBalancerControllerConfiguration{
+			ConcurrentLoadBalancerSetWorkers: 3,
+		},
+	}
+}
+
+// AddFlags adds flags related to poolservice for yurt-manager to the specified FlagSet.
+func (n *VipLoadBalancerControllerOptions) AddFlags(fs *pflag.FlagSet) {
+	if n == nil {
+		return
+	}
+
+	fs.Int32Var(&n.ConcurrentLoadBalancerSetWorkers, "concurrent-load-balancer-set-workers", n.ConcurrentLoadBalancerSetWorkers, "The number of load-balancer-set service objects that are allowed to reconcile concurrently. Larger number = more responsive load-balancer-set services, but more CPU (and network) load")
+}
+
+// ApplyTo fills up poolservice config with options.
+func (o *VipLoadBalancerControllerOptions) ApplyTo(cfg *config.VipLoadBalancerControllerConfiguration) error {
+	if o == nil {
+		return nil
+	}
+
+	cfg.ConcurrentLoadBalancerSetWorkers = o.ConcurrentLoadBalancerSetWorkers
+
+	return nil
+}
+
+// Validate checks validation of VipLoadBalancerControllerOptions.
+func (o *VipLoadBalancerControllerOptions) Validate() []error {
+	if o == nil {
+		return nil
+	}
+	errs := []error{}
+	return errs
+}

pkg/yurtmanager/controller/controller.go

@@ -33,6 +33,7 @@ import (
 	"github.com/openyurtio/openyurt/pkg/yurtmanager/controller/csrapprover"
 	"github.com/openyurtio/openyurt/pkg/yurtmanager/controller/daemonpodupdater"
 	"github.com/openyurtio/openyurt/pkg/yurtmanager/controller/loadbalancerset/loadbalancerset"
+	"github.com/openyurtio/openyurt/pkg/yurtmanager/controller/loadbalancerset/viploadbalancer"
 	"github.com/openyurtio/openyurt/pkg/yurtmanager/controller/nodebucket"
 	"github.com/openyurtio/openyurt/pkg/yurtmanager/controller/nodelifecycle"
 	"github.com/openyurtio/openyurt/pkg/yurtmanager/controller/nodepool"
@@ -99,6 +100,7 @@ func NewControllerInitializers() map[string]InitFunc {
 	register(names.NodeLifeCycleController, nodelifecycle.Add)
 	register(names.NodeBucketController, nodebucket.Add)
 	register(names.LoadBalancerSetController, loadbalancerset.Add)
+	register(names.VipLoadBalancerController, viploadbalancer.Add)
 
 	return controllers
 }

pkg/yurtmanager/controller/loadbalancerset/viploadbalancer/config/types.go

@@ -18,4 +18,5 @@ package config
 
 // VipLoadBalancerControllerConfiguration contains elements describing EdgeLoadBalanceController.
 type VipLoadBalancerControllerConfiguration struct {
+	ConcurrentLoadBalancerSetWorkers int32
 }

pkg/yurtmanager/controller/loadbalancerset/viploadbalancer/predicate.go

@@ -63,10 +63,6 @@ func NewPoolServicePredicated() predicate.Predicate {
 }
 
 func predicatedPoolService(ps *v1alpha1.PoolService) bool {
-	if ps.Labels == nil {
-		return false
-	}
-
 	return matchLoadBalancerClass(ps)
 }
 
@@ -75,9 +71,9 @@ func matchLoadBalancerClass(ps *v1alpha1.PoolService) bool {
 		return false
 	}
 
-	if *ps.Spec.LoadBalancerClass == VipLoadBalancerClass {
-		return true
+	if *ps.Spec.LoadBalancerClass != VipLoadBalancerClass {
+		return false
 	}
 
-	return false
+	return true
 }

pkg/yurtmanager/controller/loadbalancerset/viploadbalancer/predicate_test.go

@@ -0,0 +1,72 @@
+/*
+Copyright 2024 The OpenYurt Authors.
+
+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 viploadbalancer_test
+
+import (
+	"testing"
+
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"sigs.k8s.io/controller-runtime/pkg/event"
+
+	vip "github.com/openyurtio/openyurt/pkg/yurtmanager/controller/loadbalancerset/viploadbalancer"
+)
+
+var (
+	elbClass = "elb"
+)
+
+func TestPoolServicePredicated(t *testing.T) {
+	f := vip.NewPoolServicePredicated()
+
+	t.Run("create/delete/update/generic pool service has not LoadBalancerClass", func(t *testing.T) {
+		ps := newPoolService(v1.NamespaceDefault, "np123", nil, nil)
+		ps.Spec.LoadBalancerClass = nil
+
+		assertBool(t, false, f.Create(event.CreateEvent{Object: ps}))
+		assertBool(t, false, f.Update(event.UpdateEvent{ObjectOld: ps, ObjectNew: ps}))
+		assertBool(t, false, f.Delete(event.DeleteEvent{Object: ps}))
+		assertBool(t, false, f.Generic(event.GenericEvent{Object: ps}))
+	})
+
+	t.Run("create/delete/update/generic pool service LoadBalancerClass is not valid", func(t *testing.T) {
+		ps := newPoolService(v1.NamespaceDefault, "np123", nil, nil)
+		ps.Spec.LoadBalancerClass = &elbClass
+
+		assertBool(t, false, f.Create(event.CreateEvent{Object: ps}))
+		assertBool(t, false, f.Update(event.UpdateEvent{ObjectOld: ps, ObjectNew: ps}))
+		assertBool(t, false, f.Delete(event.DeleteEvent{Object: ps}))
+		assertBool(t, false, f.Generic(event.GenericEvent{Object: ps}))
+	})
+
+	t.Run("create/delete/update/generic pool service", func(t *testing.T) {
+		ps := newPoolService(v1.NamespaceDefault, "np123", nil, nil)
+
+		assertBool(t, true, f.Create(event.CreateEvent{Object: ps}))
+		assertBool(t, true, f.Update(event.UpdateEvent{ObjectOld: ps, ObjectNew: ps}))
+		assertBool(t, true, f.Delete(event.DeleteEvent{Object: ps}))
+		assertBool(t, true, f.Generic(event.GenericEvent{Object: ps}))
+
+	})
+}
+
+func assertBool(t testing.TB, expected, got bool) {
+	t.Helper()
+
+	if expected != got {
+		t.Errorf("expected %v, but got %v", expected, got)
+	}
+}

pkg/yurtmanager/controller/loadbalancerset/viploadbalancer/util.go

@@ -18,6 +18,7 @@ package viploadbalancer
 import "sync"
 
 const (
+	MINVRIDLIMIT = 0
 	MAXVRIDLIMIT = 255
 	EVICTED      = -1
 )
@@ -43,7 +44,7 @@ func (vm *VRIDManager) GetVRID(key string) int {
 		vm.vridMap[key] = make(map[int]bool)
 	}
 
-	for vrid := 1; vrid <= vm.maxVRID; vrid++ {
+	for vrid := 0; vrid < vm.maxVRID; vrid++ {
 		if !vm.vridMap[key][vrid] {
 			vm.vridMap[key][vrid] = true
 			return vrid
@@ -53,12 +54,6 @@ func (vm *VRIDManager) GetVRID(key string) int {
 	return EVICTED
 }
 
-func (vm *VRIDManager) isEmpty(key string) bool {
-	vm.mutex.Lock()
-	defer vm.mutex.Unlock()
-	return len(vm.vridMap[key]) == 0
-}
-
 func (vm *VRIDManager) ReleaseVRID(key string, vrid int) {
 	vm.mutex.Lock()
 	defer vm.mutex.Unlock()
@@ -70,7 +65,7 @@ func (vm *VRIDManager) ReleaseVRID(key string, vrid int) {
 	delete(vm.vridMap[key], vrid)
 }
 
-func (vm *VRIDManager) isValid(key string, vrid int) bool {
+func (vm *VRIDManager) IsValid(key string, vrid int) bool {
 	vm.mutex.Lock()
 	defer vm.mutex.Unlock()
 
@@ -84,3 +79,16 @@ func (vm *VRIDManager) isValid(key string, vrid int) bool {
 
 	return true
 }
+
+func (vm *VRIDManager) SyncVRID(key string, vird []int) {
+	vm.mutex.Lock()
+	defer vm.mutex.Unlock()
+
+	if _, ok := vm.vridMap[key]; !ok {
+		vm.vridMap[key] = make(map[int]bool)
+	}
+
+	for _, v := range vird {
+		vm.vridMap[key][v] = true
+	}
+}

pkg/yurtmanager/controller/loadbalancerset/viploadbalancer/util_test.go

@@ -0,0 +1,75 @@
+/*
+Copyright 2024 The OpenYurt Authors.
+
+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 viploadbalancer_test
+
+import (
+	"testing"
+
+	vip "github.com/openyurtio/openyurt/pkg/yurtmanager/controller/loadbalancerset/viploadbalancer"
+)
+
+var (
+	nodepool = "np123"
+)
+
+func TestVRIDManage(t *testing.T) {
+	vm := vip.NewVRIDManager()
+
+	t.Run("get vrid", func(t *testing.T) {
+		// Test getting VRID
+		vrid := vm.GetVRID(nodepool)
+		if vrid != 0 {
+			t.Errorf("Expected VRID 0, but got %d", vrid)
+		}
+	})
+
+	t.Run("release vrid", func(t *testing.T) {
+		// Test releasing VRID
+		target := vm.GetVRID(nodepool)
+		vm.ReleaseVRID(nodepool, target)
+		vrid := vm.GetVRID(nodepool)
+		if vrid != target {
+			t.Errorf("Expected VRID %d after release, but got %d", target, vrid)
+		}
+	})
+
+	t.Run("vrid is valid", func(t *testing.T) {
+		// Test isValid function
+		isValid := vm.IsValid(nodepool, 0)
+		if !isValid {
+			t.Errorf("Expected VRID 0 to be valid, but it's not")
+		}
+	})
+
+	t.Run("vrid is not valid", func(t *testing.T) {
+		// Test isValid function
+		isValid := vm.IsValid(nodepool, -1)
+		if isValid {
+			t.Errorf("Expected VRID -1 to be invalid, but it's not")
+		}
+	})
+
+	t.Run("sync vrid", func(t *testing.T) {
+		// Test SyncVRID function
+		vm.SyncVRID(nodepool, []int{1, 2, 3})
+		target := 4
+		vrid := vm.GetVRID(nodepool)
+		if vrid != target {
+			t.Errorf("Expected VRID %d after sync, but got %d", target, vrid)
+		}
+	})
+}