admin/kubesphere
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

enable multicluster sync for groups

Browse Source 
Signed-off-by: Roland.Ma <rolandma@yunify.com>
This commit is contained in:
Roland.Ma
2020-12-03 07:19:31 +00:00
parent c5de21af4a
commit 5248e622f2
25 changed files with 2158 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

136
pkg/controller/groupbinding/groupbinding_controller.go
View File

@@ -19,6 +19,8 @@ package groupbinding
import (
"fmt"
"reflect"
corev1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -32,12 +34,17 @@ import (
"k8s.io/client-go/util/workqueue"
"k8s.io/klog"
iamv1alpha2 "kubesphere.io/kubesphere/pkg/apis/iam/v1alpha2"
fedv1beta1types "kubesphere.io/kubesphere/pkg/apis/types/v1beta1"
kubesphere "kubesphere.io/kubesphere/pkg/client/clientset/versioned"
iamv1alpha2informers "kubesphere.io/kubesphere/pkg/client/informers/externalversions/iam/v1alpha2"
fedv1beta1informers "kubesphere.io/kubesphere/pkg/client/informers/externalversions/types/v1beta1"
iamv1alpha2listers "kubesphere.io/kubesphere/pkg/client/listers/iam/v1alpha2"
fedv1beta1lister "kubesphere.io/kubesphere/pkg/client/listers/types/v1beta1"
"kubesphere.io/kubesphere/pkg/constants"
"kubesphere.io/kubesphere/pkg/controller/utils/controller"
"kubesphere.io/kubesphere/pkg/utils/sliceutil"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
)
const (
@@ -49,16 +56,21 @@ const (
type Controller struct {
controller.BaseController
scheme *runtime.Scheme
k8sClient kubernetes.Interface
ksClient kubesphere.Interface
groupBindingInformer iamv1alpha2informers.GroupBindingInformer
groupBindingLister iamv1alpha2listers.GroupBindingLister
recorder record.EventRecorder
scheme *runtime.Scheme
k8sClient kubernetes.Interface
ksClient kubesphere.Interface
groupBindingInformer iamv1alpha2informers.GroupBindingInformer
groupBindingLister iamv1alpha2listers.GroupBindingLister
recorder record.EventRecorder
federatedGroupBindingInformer fedv1beta1informers.FederatedGroupBindingInformer
federatedGroupBindingLister fedv1beta1lister.FederatedGroupBindingLister
multiClusterEnabled bool
}
// NewController creates GroupBinding Controller instance
func NewController(k8sClient kubernetes.Interface, ksClient kubesphere.Interface, groupBindingInformer iamv1alpha2informers.GroupBindingInformer) *Controller {
func NewController(k8sClient kubernetes.Interface, ksClient kubesphere.Interface,
groupBindingInformer iamv1alpha2informers.GroupBindingInformer,
federatedGroupBindingInformer fedv1beta1informers.FederatedGroupBindingInformer, multiClusterEnabled bool) *Controller {
klog.V(4).Info("Creating event broadcaster")
eventBroadcaster := record.NewBroadcaster()
eventBroadcaster.StartLogging(klog.Infof)
@@ -70,13 +82,19 @@ func NewController(k8sClient kubernetes.Interface, ksClient kubesphere.Interface
Synced: []cache.InformerSynced{groupBindingInformer.Informer().HasSynced},
Name: controllerName,
},
k8sClient: k8sClient,
ksClient: ksClient,
groupBindingInformer: groupBindingInformer,
groupBindingLister: groupBindingInformer.Lister(),
recorder: recorder,
k8sClient: k8sClient,
ksClient: ksClient,
groupBindingInformer: groupBindingInformer,
groupBindingLister: groupBindingInformer.Lister(),
federatedGroupBindingInformer: federatedGroupBindingInformer,
federatedGroupBindingLister: federatedGroupBindingInformer.Lister(),
multiClusterEnabled: multiClusterEnabled,
recorder: recorder,
}
ctl.Handler = ctl.reconcile
if ctl.multiClusterEnabled {
ctl.Synced = append(ctl.Synced, ctl.federatedGroupBindingInformer.Informer().HasSynced)
}
klog.Info("Setting up event handlers")
groupBindingInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
AddFunc: ctl.Enqueue,
@@ -101,14 +119,32 @@ func (c *Controller) reconcile(key string) error {
return err
}
if groupBinding.ObjectMeta.DeletionTimestamp.IsZero() {
var g *iamv1alpha2.GroupBinding
if !sliceutil.HasString(groupBinding.Finalizers, finalizer) {
groupBinding.ObjectMeta.Finalizers = append(groupBinding.ObjectMeta.Finalizers, finalizer)
if groupBinding, err = c.ksClient.IamV1alpha2().GroupBindings().Update(groupBinding); err != nil {
g = groupBinding.DeepCopy()
g.ObjectMeta.Finalizers = append(g.ObjectMeta.Finalizers, finalizer)
}
if c.multiClusterEnabled {
// Ensure not controlled by Kubefed
if groupBinding.Labels == nil || groupBinding.Labels[constants.KubefedManagedLabel] != "false" {
if g == nil {
g = groupBinding.DeepCopy()
}
if g.Labels == nil {
g.Labels = make(map[string]string, 0)
}
g.Labels[constants.KubefedManagedLabel] = "false"
}
}
if g != nil {
if groupBinding, err = c.ksClient.IamV1alpha2().GroupBindings().Update(g); err != nil {
return err
}
// Skip reconcile when groupbinding is updated.
// Skip reconcile when group is updated.
return nil
}
} else {
// The object is being deleted
if sliceutil.HasString(groupBinding.ObjectMeta.Finalizers, finalizer) {
@@ -133,6 +169,14 @@ func (c *Controller) reconcile(key string) error {
return err
}
// synchronization through kubefed-controller when multi cluster is enabled
if c.multiClusterEnabled {
if err = c.multiClusterSync(groupBinding); err != nil {
klog.Error(err)
return err
}
}
c.recorder.Event(groupBinding, corev1.EventTypeNormal, successSynced, messageResourceSynced)
return nil
}
@@ -204,3 +248,65 @@ func (c *Controller) patchUser(user *iamv1alpha2.User, groups []string) error {
}
return nil
}
func (c *Controller) multiClusterSync(groupBinding *iamv1alpha2.GroupBinding) error {
fedGroupBinding, err := c.federatedGroupBindingLister.Get(groupBinding.Name)
if err != nil {
if errors.IsNotFound(err) {
return c.createFederatedGroupBinding(groupBinding)
}
klog.Error(err)
return err
}
if !reflect.DeepEqual(fedGroupBinding.Spec.Template.GroupRef, groupBinding.GroupRef) ||
!reflect.DeepEqual(fedGroupBinding.Spec.Template.Users, groupBinding.Users) ||
!reflect.DeepEqual(fedGroupBinding.Spec.Template.Labels, groupBinding.Labels) {
fedGroupBinding.Spec.Template.GroupRef = groupBinding.GroupRef
fedGroupBinding.Spec.Template.Users = groupBinding.Users
fedGroupBinding.Spec.Template.Labels = groupBinding.Labels
if _, err = c.ksClient.TypesV1beta1().FederatedGroupBindings().Update(fedGroupBinding); err != nil {
return err
}
}
return nil
}
func (c *Controller) createFederatedGroupBinding(groupBinding *iamv1alpha2.GroupBinding) error {
federatedGroup := &fedv1beta1types.FederatedGroupBinding{
TypeMeta: metav1.TypeMeta{
Kind: fedv1beta1types.FederatedGroupBindingKind,
APIVersion: fedv1beta1types.SchemeGroupVersion.String(),
},
ObjectMeta: metav1.ObjectMeta{
Name: groupBinding.Name,
},
Spec: fedv1beta1types.FederatedGroupBindingSpec{
Template: fedv1beta1types.GroupBindingTemplate{
ObjectMeta: metav1.ObjectMeta{
Labels: groupBinding.Labels,
},
GroupRef: groupBinding.GroupRef,
Users: groupBinding.Users,
},
Placement: fedv1beta1types.GenericPlacementFields{
ClusterSelector: &metav1.LabelSelector{},
},
},
}
// must bind groupBinding lifecycle
err := controllerutil.SetControllerReference(groupBinding, federatedGroup, scheme.Scheme)
if err != nil {
return err
}
if _, err = c.ksClient.TypesV1beta1().FederatedGroupBindings().Create(federatedGroup); err != nil {
return err
}
return nil
}

59
pkg/controller/groupbinding/groupbinding_controller_test.go
View File

@@ -22,20 +22,24 @@ import (
"testing"
"time"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/schema"
"k8s.io/apimachinery/pkg/util/diff"
kubeinformers "k8s.io/client-go/informers"
k8sfake "k8s.io/client-go/kubernetes/fake"
"k8s.io/client-go/kubernetes/scheme"
core "k8s.io/client-go/testing"
"k8s.io/client-go/tools/cache"
"k8s.io/client-go/tools/record"
iamv1alpha2 "kubesphere.io/kubesphere/pkg/apis/iam/v1alpha2"
v1alpha2 "kubesphere.io/kubesphere/pkg/apis/iam/v1alpha2"
fedv1beta1types "kubesphere.io/kubesphere/pkg/apis/types/v1beta1"
"kubesphere.io/kubesphere/pkg/client/clientset/versioned/fake"
ksinformers "kubesphere.io/kubesphere/pkg/client/informers/externalversions"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"kubesphere.io/kubesphere/pkg/constants"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
)
var (
@@ -43,14 +47,19 @@ var (
noResyncPeriodFunc = func() time.Duration { return 0 }
)
func init() {
v1alpha2.AddToScheme(scheme.Scheme)
}
type fixture struct {
t *testing.T
ksclient *fake.Clientset
k8sclient *k8sfake.Clientset
// Objects to put in the store.
groupBindingLister []*v1alpha2.GroupBinding
userLister []*v1alpha2.User
groupBindingLister []*v1alpha2.GroupBinding
fedgroupBindingLister []*fedv1beta1types.FederatedGroupBinding
userLister []*v1alpha2.User
// Actions expected to happen on the client.
kubeactions []core.Action
actions []core.Action
@@ -94,6 +103,30 @@ func newUser(name string) *v1alpha2.User {
}
}
func newFederatedGroupBinding(groupBinding *iamv1alpha2.GroupBinding) *fedv1beta1types.FederatedGroupBinding {
return &fedv1beta1types.FederatedGroupBinding{
TypeMeta: metav1.TypeMeta{
Kind: fedv1beta1types.FederatedGroupBindingKind,
APIVersion: fedv1beta1types.SchemeGroupVersion.String(),
},
ObjectMeta: metav1.ObjectMeta{
Name: groupBinding.Name,
},
Spec: fedv1beta1types.FederatedGroupBindingSpec{
Template: fedv1beta1types.GroupBindingTemplate{
ObjectMeta: metav1.ObjectMeta{
Labels: groupBinding.Labels,
},
GroupRef: groupBinding.GroupRef,
Users: groupBinding.Users,
},
Placement: fedv1beta1types.GenericPlacementFields{
ClusterSelector: &metav1.LabelSelector{},
},
},
}
}
func (f *fixture) newController() (*Controller, ksinformers.SharedInformerFactory, kubeinformers.SharedInformerFactory) {
f.ksclient = fake.NewSimpleClientset(f.objects...)
f.k8sclient = k8sfake.NewSimpleClientset(f.kubeobjects...)
@@ -116,7 +149,8 @@ func (f *fixture) newController() (*Controller, ksinformers.SharedInformerFactor
}
c := NewController(f.k8sclient, f.ksclient,
ksinformers.Iam().V1alpha2().GroupBindings())
ksinformers.Iam().V1alpha2().GroupBindings(),
ksinformers.Types().V1beta1().FederatedGroupBindings(), true)
c.Synced = []cache.InformerSynced{alwaysReady}
c.recorder = &record.FakeRecorder{}
@@ -236,6 +270,7 @@ func filterInformerActions(actions []core.Action) []core.Action {
if len(action.GetNamespace()) == 0 &&
(action.Matches("list", "groupbindings") ||
action.Matches("watch", "groupbindings") ||
action.Matches("list", "federatedgroupbindings") ||
action.Matches("list", "users") ||
action.Matches("watch", "users") ||
action.Matches("get", "users")) {
@@ -250,6 +285,7 @@ func filterInformerActions(actions []core.Action) []core.Action {
func (f *fixture) expectUpdateGroupsFinalizerAction(groupBinding *v1alpha2.GroupBinding) {
expect := groupBinding.DeepCopy()
expect.Finalizers = []string{"finalizers.kubesphere.io/groupsbindings"}
expect.Labels = map[string]string{constants.KubefedManagedLabel: "false"}
action := core.NewUpdateAction(schema.GroupVersionResource{Group: "iam.kubesphere.io", Version: "v1alpha2", Resource: "groupbindings"}, "", expect)
f.actions = append(f.actions, action)
}
@@ -270,6 +306,15 @@ func (f *fixture) expectPatchUserAction(user *v1alpha2.User, groups []string) {
f.actions = append(f.actions, core.NewPatchAction(schema.GroupVersionResource{Group: "iam.kubesphere.io", Resource: "users", Version: "v1alpha2"}, user.Namespace, user.Name, patch.Type(), patchData))
}
func (f *fixture) expectCreateFederatedGroupBindingsAction(groupBinding *v1alpha2.GroupBinding) {
b := newFederatedGroupBinding(groupBinding)
controllerutil.SetControllerReference(groupBinding, b, scheme.Scheme)
actionCreate := core.NewCreateAction(schema.GroupVersionResource{Group: "types.kubefed.io", Version: "v1beta1", Resource: "federatedgroupbindings"}, "", b)
f.actions = append(f.actions, actionCreate)
}
func getKey(groupBinding *v1alpha2.GroupBinding, t *testing.T) string {
key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(groupBinding)
if err != nil {
@@ -285,6 +330,7 @@ func TestCreatesGroupBinding(t *testing.T) {
users := []string{"user1"}
groupbinding := newGroupBinding("test", users)
groupbinding.ObjectMeta.Finalizers = append(groupbinding.ObjectMeta.Finalizers, finalizer)
groupbinding.Labels = map[string]string{constants.KubefedManagedLabel: "false"}
f.groupBindingLister = append(f.groupBindingLister, groupbinding)
f.objects = append(f.objects, groupbinding)
@@ -294,7 +340,9 @@ func TestCreatesGroupBinding(t *testing.T) {
f.objects = append(f.objects, user)
excepctGroups := []string{"test"}
f.expectPatchUserAction(user, excepctGroups)
f.expectCreateFederatedGroupBindingsAction(groupbinding)
f.run(getKey(groupbinding, t))
}
@@ -326,6 +374,7 @@ func TestDeletesGroupBinding(t *testing.T) {
func TestDoNothing(t *testing.T) {
f := newFixture(t)
users := []string{"user1"}
groupBinding := newGroupBinding("test", users)