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

Merge pull request #3161 from RolandMa1986/feat-group-fedsync

Browse Source 
enable multicluster sync for groups
This commit is contained in:
KubeSphere CI Bot
2020-12-09 15:17:43 +08:00
committed by GitHub
parent e4e6405c8c 5248e622f2
commit dab0842a32
25 changed files with 2158 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

123
pkg/controller/group/group_controller.go
View File

@@ -18,6 +18,7 @@ package group
import (
"fmt"
"reflect"
corev1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/api/errors"
@@ -33,11 +34,16 @@ import (
"k8s.io/client-go/util/workqueue"
"k8s.io/klog"
iam1alpha2 "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"
iamv1alpha1listers "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/controller/controllerutil"
)
const (
@@ -49,16 +55,21 @@ const (
type Controller struct {
controller.BaseController
scheme *runtime.Scheme
k8sClient kubernetes.Interface
ksClient kubesphere.Interface
groupInformer iamv1alpha2informers.GroupInformer
groupLister iamv1alpha1listers.GroupLister
recorder record.EventRecorder
scheme *runtime.Scheme
k8sClient kubernetes.Interface
ksClient kubesphere.Interface
groupInformer iamv1alpha2informers.GroupInformer
groupLister iamv1alpha1listers.GroupLister
recorder record.EventRecorder
federatedGroupInformer fedv1beta1informers.FederatedGroupInformer
federatedGroupLister fedv1beta1lister.FederatedGroupLister
multiClusterEnabled bool
}
// NewController creates Group Controller instance
func NewController(k8sClient kubernetes.Interface, ksClient kubesphere.Interface, groupInformer iamv1alpha2informers.GroupInformer) *Controller {
func NewController(k8sClient kubernetes.Interface, ksClient kubesphere.Interface, groupInformer iamv1alpha2informers.GroupInformer,
federatedGroupInformer fedv1beta1informers.FederatedGroupInformer,
multiClusterEnabled bool) *Controller {
klog.V(4).Info("Creating event broadcaster")
eventBroadcaster := record.NewBroadcaster()
@@ -70,12 +81,20 @@ func NewController(k8sClient kubernetes.Interface, ksClient kubesphere.Interface
Synced: []cache.InformerSynced{groupInformer.Informer().HasSynced},
Name: controllerName,
},
recorder: eventBroadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: controllerName}),
k8sClient: k8sClient,
ksClient: ksClient,
groupInformer: groupInformer,
groupLister: groupInformer.Lister(),
recorder: eventBroadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: controllerName}),
k8sClient: k8sClient,
ksClient: ksClient,
groupInformer: groupInformer,
groupLister: groupInformer.Lister(),
federatedGroupInformer: federatedGroupInformer,
federatedGroupLister: federatedGroupInformer.Lister(),
multiClusterEnabled: multiClusterEnabled,
}
if ctl.multiClusterEnabled {
ctl.Synced = append(ctl.Synced, ctl.federatedGroupInformer.Informer().HasSynced)
}
ctl.Handler = ctl.reconcile
klog.Info("Setting up event handlers")
@@ -106,10 +125,27 @@ func (c *Controller) reconcile(key string) error {
return err
}
if group.ObjectMeta.DeletionTimestamp.IsZero() {
var g *iam1alpha2.Group
if !sliceutil.HasString(group.Finalizers, finalizer) {
group.ObjectMeta.Finalizers = append(group.ObjectMeta.Finalizers, finalizer)
g = group.DeepCopy()
g.ObjectMeta.Finalizers = append(g.ObjectMeta.Finalizers, finalizer)
}
if group, err = c.ksClient.IamV1alpha2().Groups().Update(group); err != nil {
if c.multiClusterEnabled {
// Ensure not controlled by Kubefed
if group.Labels == nil || group.Labels[constants.KubefedManagedLabel] != "false" {
if g == nil {
g = group.DeepCopy()
}
if g.Labels == nil {
g.Labels = make(map[string]string, 0)
}
g.Labels[constants.KubefedManagedLabel] = "false"
}
}
if g != nil {
if _, err = c.ksClient.IamV1alpha2().Groups().Update(g); err != nil {
return err
}
// Skip reconcile when group is updated.
@@ -139,6 +175,14 @@ func (c *Controller) reconcile(key string) error {
return nil
}
// synchronization through kubefed-controller when multi cluster is enabled
if c.multiClusterEnabled {
if err = c.multiClusterSync(group); err != nil {
klog.Error(err)
return err
}
}
c.recorder.Event(group, corev1.EventTypeNormal, successSynced, messageResourceSynced)
return nil
}
@@ -192,3 +236,54 @@ func (c *Controller) deleteRoleBindings(group *iam1alpha2.Group) error {
return nil
}
func (c *Controller) multiClusterSync(group *iam1alpha2.Group) error {
obj, err := c.federatedGroupLister.Get(group.Name)
if err != nil {
if errors.IsNotFound(err) {
return c.createFederatedGroup(group)
}
klog.Error(err)
return err
}
if !reflect.DeepEqual(obj.Spec.Template.Labels, group.Labels) {
obj.Spec.Template.Labels = group.Labels
if _, err = c.ksClient.TypesV1beta1().FederatedGroups().Update(obj); err != nil {
return err
}
}
return nil
}
func (c *Controller) createFederatedGroup(group *iam1alpha2.Group) error {
federatedGroup := &fedv1beta1types.FederatedGroup{
ObjectMeta: metav1.ObjectMeta{
Name: group.Name,
},
Spec: fedv1beta1types.FederatedGroupSpec{
Template: fedv1beta1types.GroupTemplate{
ObjectMeta: metav1.ObjectMeta{
Labels: group.Labels,
},
Spec: group.Spec,
},
Placement: fedv1beta1types.GenericPlacementFields{
ClusterSelector: &metav1.LabelSelector{},
},
},
}
// must bind group lifecycle
err := controllerutil.SetControllerReference(group, federatedGroup, scheme.Scheme)
if err != nil {
return err
}
if _, err = c.ksClient.TypesV1beta1().FederatedGroups().Create(federatedGroup); err != nil {
return err
}
return nil
}

153
pkg/controller/group/group_controller_test.go
View File

@@ -21,17 +21,22 @@ import (
"testing"
"time"
"k8s.io/apimachinery/pkg/labels"
"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"
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"
"kubesphere.io/kubesphere/pkg/constants"
"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)
@@ -41,13 +46,18 @@ 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.
groupLister []*v1alpha2.Group
groupLister []*v1alpha2.Group
fedgroupLister []*fedv1beta1types.FederatedGroup
// Actions expected to happen on the client.
kubeactions []core.Action
actions []core.Action
@@ -66,7 +76,7 @@ func newFixture(t *testing.T) *fixture {
func newGroup(name string) *v1alpha2.Group {
return &v1alpha2.Group{
TypeMeta: metav1.TypeMeta{APIVersion: v1alpha2.SchemeGroupVersion.String()},
TypeMeta: metav1.TypeMeta{APIVersion: v1alpha2.SchemeGroupVersion.String(), Kind: "Group"},
ObjectMeta: metav1.ObjectMeta{
Name: name,
},
@@ -74,6 +84,38 @@ func newGroup(name string) *v1alpha2.Group {
}
}
func newUnmanagedGroup(name string) *v1alpha2.Group {
return &v1alpha2.Group{
TypeMeta: metav1.TypeMeta{APIVersion: v1alpha2.SchemeGroupVersion.String(), Kind: "Group"},
ObjectMeta: metav1.ObjectMeta{
Name: name,
Labels: map[string]string{constants.KubefedManagedLabel: "false"},
Finalizers: []string{"finalizers.kubesphere.io/groups"},
},
Spec: v1alpha2.GroupSpec{},
}
}
func newFederatedGroup(group *v1alpha2.Group) *fedv1beta1types.FederatedGroup {
return &fedv1beta1types.FederatedGroup{
ObjectMeta: metav1.ObjectMeta{
Name: group.Name,
},
Spec: fedv1beta1types.FederatedGroupSpec{
Template: fedv1beta1types.GroupTemplate{
ObjectMeta: metav1.ObjectMeta{
Labels: group.Labels,
},
Spec: group.Spec,
},
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...)
@@ -88,8 +130,16 @@ func (f *fixture) newController() (*Controller, ksinformers.SharedInformerFactor
}
}
for _, group := range f.fedgroupLister {
err := ksinformers.Types().V1beta1().FederatedGroups().Informer().GetIndexer().Add(group)
if err != nil {
f.t.Errorf("add federated group:%s", err)
}
}
c := NewController(f.k8sclient, f.ksclient,
ksinformers.Iam().V1alpha2().Groups())
ksinformers.Iam().V1alpha2().Groups(),
ksinformers.Types().V1beta1().FederatedGroups(), true)
c.recorder = &record.FakeRecorder{}
return c, ksinformers, k8sinformers
@@ -177,10 +227,8 @@ func checkAction(expected, actual core.Action, t *testing.T) {
e, _ := expected.(core.UpdateActionImpl)
expObject := e.GetObject()
object := a.GetObject()
expUser := expObject.(*v1alpha2.Group)
group := object.(*v1alpha2.Group)
if !reflect.DeepEqual(expUser, group) {
if !reflect.DeepEqual(expObject, object) {
t.Errorf("Action %s %s has wrong object\nDiff:\n %s",
a.GetVerb(), a.GetResource().Resource, diff.ObjectGoPrintSideBySide(expObject, object))
}
@@ -193,6 +241,15 @@ func checkAction(expected, actual core.Action, t *testing.T) {
t.Errorf("Action %s %s has wrong patch\nDiff:\n %s",
a.GetVerb(), a.GetResource().Resource, diff.ObjectGoPrintSideBySide(expPatch, patch))
}
case core.DeleteCollectionActionImpl:
e, _ := expected.(core.DeleteCollectionActionImpl)
exp := e.GetListRestrictions()
target := a.GetListRestrictions()
if !reflect.DeepEqual(exp, target) {
t.Errorf("Action %s %s has wrong Query\nDiff:\n %s",
a.GetVerb(), a.GetResource().Resource, diff.ObjectGoPrintSideBySide(exp, target))
}
default:
t.Errorf("Uncaptured Action %s %s, you should explicitly add a case to capture it",
actual.GetVerb(), actual.GetResource().Resource)
@@ -205,7 +262,13 @@ func checkAction(expected, actual core.Action, t *testing.T) {
func filterInformerActions(actions []core.Action) []core.Action {
var ret []core.Action
for _, action := range actions {
if !action.Matches("update", "groups") {
if len(action.GetNamespace()) == 0 &&
(action.Matches("list", "groups") ||
action.Matches("watch", "groups") ||
action.Matches("list", "groups") ||
action.Matches("list", "namespaces") ||
action.Matches("list", "federatedgroups") ||
action.Matches("watch", "federatedgroups")) {
continue
}
ret = append(ret, action)
@@ -216,14 +279,47 @@ func filterInformerActions(actions []core.Action) []core.Action {
func (f *fixture) expectUpdateGroupsFinalizerAction(group *v1alpha2.Group) {
expect := group.DeepCopy()
if expect.Labels == nil {
expect.Labels = make(map[string]string, 0)
}
expect.Finalizers = []string{"finalizers.kubesphere.io/groups"}
expect.Labels[constants.KubefedManagedLabel] = "false"
action := core.NewUpdateAction(schema.GroupVersionResource{Resource: "groups"}, "", expect)
f.actions = append(f.actions, action)
}
func (f *fixture) expectCreateFederatedGroupsAction(group *v1alpha2.Group) {
federatedGroup := newFederatedGroup(group)
controllerutil.SetControllerReference(group, federatedGroup, scheme.Scheme)
actionCreate := core.NewCreateAction(schema.GroupVersionResource{Resource: "federatedgroups"}, "", federatedGroup)
f.actions = append(f.actions, actionCreate)
}
func (f *fixture) expectUpdateFederatedGroupsAction(group *v1alpha2.Group) {
g := newFederatedGroup(group)
controllerutil.SetControllerReference(group, g, scheme.Scheme)
actionCreate := core.NewUpdateAction(schema.GroupVersionResource{Group: "types.kubefed.io", Version: "v1beta1", Resource: "federatedgroups"}, "", g)
f.actions = append(f.actions, actionCreate)
}
func (f *fixture) expectUpdateGroupsDeleteAction(group *v1alpha2.Group) {
expect := group.DeepCopy()
expect.Finalizers = []string{}
listOptions := metav1.ListOptions{
LabelSelector: labels.SelectorFromSet(labels.Set{v1alpha2.GroupReferenceLabel: group.Name}).String(),
}
actionDelete := core.NewDeleteCollectionAction(schema.GroupVersionResource{Resource: "groupbindings"}, "", listOptions)
f.actions = append(f.actions, actionDelete)
actionDelete = core.NewDeleteCollectionAction(schema.GroupVersionResource{Group: "rbac.authorization.k8s.io", Version: "v1", Resource: "clusterrolebindings"}, "", listOptions)
f.kubeactions = append(f.kubeactions, actionDelete)
actionDelete = core.NewDeleteCollectionAction(schema.GroupVersionResource{Resource: "workspacerolebindings"}, "", listOptions)
f.actions = append(f.actions, actionDelete)
action := core.NewUpdateAction(schema.GroupVersionResource{Resource: "groups"}, "", expect)
f.actions = append(f.actions, action)
}
@@ -239,18 +335,8 @@ func getKey(group *v1alpha2.Group, t *testing.T) string {
func TestDeletesGroup(t *testing.T) {
f := newFixture(t)
group := newGroup("test")
deletedGroup := newUnmanagedGroup("test")
f.groupLister = append(f.groupLister, group)
f.objects = append(f.objects, group)
f.expectUpdateGroupsFinalizerAction(group)
f.run(getKey(group, t))
f = newFixture(t)
deletedGroup := group.DeepCopy()
deletedGroup.Finalizers = []string{"finalizers.kubesphere.io/groups"}
now := metav1.Now()
deletedGroup.ObjectMeta.DeletionTimestamp = &now
@@ -270,3 +356,34 @@ func TestDoNothing(t *testing.T) {
f.expectUpdateGroupsFinalizerAction(group)
f.run(getKey(group, t))
}
func TestFederetedGroupCreate(t *testing.T) {
f := newFixture(t)
group := newUnmanagedGroup("test")
f.groupLister = append(f.groupLister, group)
f.objects = append(f.objects, group)
f.expectCreateFederatedGroupsAction(group)
f.run(getKey(group, t))
}
func TestFederetedGroupUpdate(t *testing.T) {
f := newFixture(t)
group := newUnmanagedGroup("test")
federatedGroup := newFederatedGroup(group.DeepCopy())
controllerutil.SetControllerReference(group, federatedGroup, scheme.Scheme)
f.fedgroupLister = append(f.fedgroupLister, federatedGroup)
f.objects = append(f.objects, federatedGroup)
group.Labels["foo"] = "bar"
f.groupLister = append(f.groupLister, group)
f.objects = append(f.objects, group)
f.expectUpdateFederatedGroupsAction(group.DeepCopy())
f.run(getKey(group, t))
}