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

fix bug member cluster can not view the notification channels

Browse Source 
Signed-off-by: wanjunlei <wanjunlei@yunify.com>
This commit is contained in:
wanjunlei
2021-04-07 16:23:28 +08:00
parent f382847fb7
commit 8f62294760
3 changed files with 151 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files

145
pkg/controller/notification/notification_controller.go
View File

@@ -18,6 +18,7 @@ package notification
import (
"context"
"encoding/json"
"fmt"
"reflect"
"time"
@@ -40,6 +41,7 @@ import (
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
"kubesphere.io/kubesphere/pkg/apis/cluster/v1alpha1"
"kubesphere.io/kubesphere/pkg/apis/notification/v2beta1"
"kubesphere.io/kubesphere/pkg/apis/types/v1beta1"
"kubesphere.io/kubesphere/pkg/constants"
@@ -49,6 +51,7 @@ const (
// SuccessSynced is used as part of the Event 'reason' when a Foo is synced
successSynced = "Synced"
controllerName = "notification-controller"
messageResourceSynced = "Notification synced successfully"
)
type Controller struct {
@@ -121,6 +124,18 @@ func (c *Controller) setEventHandlers() error {
}
}
// Watch the cluster add and delete operator.
if informer, err := c.ksCache.GetInformer(context.Background(), &v1alpha1.Cluster{}); err != nil {
klog.Errorf("get cluster informer error, %v", err)
return err
} else {
informer.AddEventHandler(toolscache.ResourceEventHandlerFuncs{
AddFunc: c.enqueue,
DeleteFunc: c.enqueue,
})
c.informerSynced = append(c.informerSynced, informer.HasSynced)
}
return nil
}
@@ -224,15 +239,28 @@ func (c *Controller) reconcile(obj interface{}) error {
}
name := accessor.GetName()
kind := runtimeObj.GetObjectKind().GroupVersionKind().String()
// If the cluster changed, add or delete, it should update the annotations of secrets
// which the notification controller managed, then the notification controller
// will receiver a event to reconcile the secret.
if _, ok := obj.(*v1alpha1.Cluster); ok {
err = c.updateSecret()
if err != nil {
klog.Errorf("update secret failed, %s", err)
return err
}
return nil
}
err = c.Get(context.Background(), client.ObjectKey{Name: accessor.GetName(), Namespace: accessor.GetNamespace()}, runtimeObj)
if err != nil {
// The user may no longer exist, in which case we stop
// processing.
if errors.IsNotFound(err) {
utilruntime.HandleError(fmt.Errorf("obj '%s, %s' in work queue no longer exists", kind, name))
c.recorder.Event(runtimeObj, corev1.EventTypeNormal, successSynced, fmt.Sprintf("%s synced successfully", kind))
klog.Infof("Successfully synced %s:%s", kind, name)
utilruntime.HandleError(fmt.Errorf("obj '%s' in work queue no longer exists", name))
c.recorder.Event(runtimeObj, corev1.EventTypeNormal, successSynced, messageResourceSynced)
klog.Infof("Successfully synced %s", name)
return nil
}
klog.Error(err)
@@ -243,8 +271,8 @@ func (c *Controller) reconcile(obj interface{}) error {
return err
}
c.recorder.Event(runtimeObj, corev1.EventTypeNormal, successSynced, fmt.Sprintf("%s synced successfully", kind))
klog.Infof("Successfully synced %s:%s", kind, name)
c.recorder.Event(runtimeObj, corev1.EventTypeNormal, successSynced, messageResourceSynced)
klog.Infof("Successfully synced %s", name)
return nil
}
@@ -299,18 +327,20 @@ func (c *Controller) syncFederatedConfig(obj *v2beta1.Config) error {
},
}
err := controllerutil.SetControllerReference(obj, fedObj, scheme.Scheme)
err = controllerutil.SetControllerReference(obj, fedObj, scheme.Scheme)
if err != nil {
klog.Errorf("FederatedNotificationConfig '%s' SetControllerReference failed, %s", obj.Name, err)
return err
}
if err := c.Create(context.Background(), fedObj); err != nil {
klog.Errorf("create '%s:%s' failed, %s", fedObj.GetObjectKind().GroupVersionKind().String(), obj.Name, err)
if err = c.Create(context.Background(), fedObj); err != nil {
klog.Errorf("create FederatedNotificationConfig '%s' failed, %s", obj.Name, err)
return err
}
return nil
}
klog.Error(err)
klog.Errorf("get FederatedNotificationConfig '%s' failed, %s", obj.Name, err)
return err
}
@@ -320,7 +350,7 @@ func (c *Controller) syncFederatedConfig(obj *v2beta1.Config) error {
fedObj.Spec.Template.Labels = obj.Labels
if err := c.Update(context.Background(), fedObj); err != nil {
klog.Errorf("update '%s:%s' failed, %s", fedObj.GetObjectKind().GroupVersionKind().String(), obj.Name, err)
klog.Errorf("update FederatedNotificationConfig '%s' failed, %s", obj.Name, err)
return err
}
}
@@ -355,18 +385,20 @@ func (c *Controller) syncFederatedReceiver(obj *v2beta1.Receiver) error {
},
}
err := controllerutil.SetControllerReference(obj, fedObj, scheme.Scheme)
err = controllerutil.SetControllerReference(obj, fedObj, scheme.Scheme)
if err != nil {
klog.Errorf("FederatedNotificationReceiver '%s' SetControllerReference failed, %s", obj.Name, err)
return err
}
if err := c.Create(context.Background(), fedObj); err != nil {
klog.Errorf("create '%s:%s' failed, %s", fedObj.GetObjectKind().GroupVersionKind().String(), obj.Name, err)
if err = c.Create(context.Background(), fedObj); err != nil {
klog.Errorf("create FederatedNotificationReceiver '%s' failed, %s", obj.Name, err)
return err
}
return nil
}
klog.Error(err)
klog.Errorf("get FederatedNotificationReceiver '%s' failed, %s", obj.Name, err)
return err
}
@@ -376,7 +408,7 @@ func (c *Controller) syncFederatedReceiver(obj *v2beta1.Receiver) error {
fedObj.Spec.Template.Labels = obj.Labels
if err := c.Update(context.Background(), fedObj); err != nil {
klog.Errorf("update '%s:%s' failed, %s", fedObj.GetObjectKind().GroupVersionKind().String(), obj.Name, err)
klog.Errorf("update FederatedNotificationReceiver '%s' failed, %s", obj.Name, err)
return err
}
}
@@ -411,18 +443,26 @@ func (c *Controller) syncFederatedSecret(obj *corev1.Secret) error {
},
}
err := controllerutil.SetControllerReference(obj, fedObj, scheme.Scheme)
err = c.updateOverrides(obj, fedObj)
if err != nil {
klog.Errorf("update FederatedSecret '%s' overrides failed, %s", obj.Name, err)
return err
}
if err := c.Create(context.Background(), fedObj); err != nil {
klog.Errorf("create '%s:%s' failed, %s", fedObj.GetObjectKind().GroupVersionKind().String(), obj.Name, err)
err = controllerutil.SetControllerReference(obj, fedObj, scheme.Scheme)
if err != nil {
klog.Errorf("FederatedSecret '%s' SetControllerReference failed, %s", obj.Name, err)
return err
}
if err = c.Create(context.Background(), fedObj); err != nil {
klog.Errorf("create FederatedSecret '%s' failed, %s", obj.Name, err)
return err
}
return nil
}
klog.Error(err)
klog.Errorf("get FederatedSecret '%s' failed, %s", obj.Name, err)
return err
}
@@ -433,9 +473,72 @@ func (c *Controller) syncFederatedSecret(obj *corev1.Secret) error {
fedObj.Spec.Template.Data = obj.Data
fedObj.Spec.Template.StringData = obj.StringData
fedObj.Spec.Template.Type = obj.Type
}
err = c.updateOverrides(obj, fedObj)
if err != nil {
klog.Errorf("update FederatedSecret '%s' overrides failed, %s", obj.Name, err)
return err
}
if err := c.Update(context.Background(), fedObj); err != nil {
klog.Errorf("update '%s:%s' failed, %s", fedObj.GetObjectKind().GroupVersionKind().String(), obj.Name, err)
klog.Errorf("update FederatedSecret '%s' failed, %s", obj.Name, err)
return err
}
return nil
}
func (c *Controller) updateOverrides(obj *corev1.Secret, fedSecret *v1beta1.FederatedSecret) error {
clusterList := &v1alpha1.ClusterList{}
err := c.ksCache.List(context.Background(), clusterList)
if err != nil {
return err
}
bs, err := json.Marshal(obj.Labels)
if err != nil {
return err
}
fedSecret.Spec.Overrides = fedSecret.Spec.Overrides[:0]
for _, cluster := range clusterList.Items {
fedSecret.Spec.Overrides = append(fedSecret.Spec.Overrides, v1beta1.GenericOverrideItem{
ClusterName: cluster.Name,
ClusterOverrides: []v1beta1.ClusterOverride{
{
Path: "/metadata/labels",
Value: runtime.RawExtension{
Raw: bs,
},
},
},
})
}
return nil
}
// Update the annotations of secrets which the notification controller managed, rigger a reconcile.
func (c *Controller) updateSecret() error {
secretList := &corev1.SecretList{}
err := c.ksCache.List(context.Background(), secretList,
client.InNamespace(constants.NotificationSecretNamespace),
client.MatchingLabels{
constants.NotificationManagedLabel: "true",
})
if err != nil {
return err
}
for _, secret := range secretList.Items {
if secret.Annotations == nil {
secret.Annotations = make(map[string]string)
}
secret.Annotations["reloadtimestamp"] = time.Now().String()
if err := c.Update(context.Background(), &secret); err != nil {
return err
}
}

37
pkg/controller/notification/notification_controller_suite_test.go
View File

@@ -20,18 +20,15 @@ import (
"path/filepath"
"testing"
"k8s.io/client-go/kubernetes/fake"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"k8s.io/client-go/kubernetes/scheme"
"k8s.io/client-go/rest"
"sigs.k8s.io/controller-runtime/pkg/envtest"
"sigs.k8s.io/controller-runtime/pkg/envtest/printer"
"kubesphere.io/kubesphere/pkg/apis"
"kubesphere.io/kubesphere/pkg/apis/notification/v2beta1"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"k8s.io/client-go/rest"
ctrl "sigs.k8s.io/controller-runtime"
"sigs.k8s.io/controller-runtime/pkg/envtest"
"sigs.k8s.io/controller-runtime/pkg/envtest/printer"
)
func TestSource(t *testing.T) {
@@ -42,7 +39,6 @@ func TestSource(t *testing.T) {
var testenv *envtest.Environment
var cfg *rest.Config
var k8sManager ctrl.Manager
var _ = BeforeSuite(func(done Done) {
@@ -55,27 +51,8 @@ var _ = BeforeSuite(func(done Done) {
Expect(err).NotTo(HaveOccurred())
Expect(cfg).ToNot(BeNil())
err = v2beta1.AddToScheme(scheme.Scheme)
Expect(err).NotTo(HaveOccurred())
err = apis.AddToScheme(scheme.Scheme)
Expect(err).NotTo(HaveOccurred())
k8sManager, err = ctrl.NewManager(cfg, ctrl.Options{
Scheme: scheme.Scheme,
MetricsBindAddress: "0",
})
Expect(err).ToNot(HaveOccurred())
r, err := NewController(fake.NewSimpleClientset(), k8sManager.GetClient(), k8sManager.GetCache())
Expect(err).ToNot(HaveOccurred())
err = k8sManager.Add(r)
Expect(err).ToNot(HaveOccurred())
go func() {
err = k8sManager.Start(ctrl.SetupSignalHandler())
Expect(err).ToNot(HaveOccurred())
}()
Expect(v2beta1.AddToScheme(scheme.Scheme)).NotTo(HaveOccurred())
Expect(apis.AddToScheme(scheme.Scheme)).NotTo(HaveOccurred())
close(done)
}, 60)

26
pkg/controller/notification/notification_controller_test.go
View File

@@ -25,6 +25,7 @@ import (
corev1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
fakek8s "k8s.io/client-go/kubernetes/fake"
"sigs.k8s.io/controller-runtime/pkg/cache"
"sigs.k8s.io/controller-runtime/pkg/client"
@@ -91,16 +92,19 @@ var (
informerCacheCancel()
})
// Add Tests for OpenAPI validation (or additonal CRD features) specified in
// Add Tests for OpenAPI validation (or additional CRD features) specified in
// your API definition.
// Avoid adding tests for vanilla CRUD operations because they would
// test Kubernetes API server, which isn't the goal here.
Context("Notification Controller", func() {
It("Should create successfully", func() {
r, err := NewController(fakek8s.NewSimpleClientset(), cl, ksCache)
Expect(err).ToNot(HaveOccurred())
// Create a secret
Expect(cl.Create(context.Background(), secret)).Should(Succeed())
time.Sleep(time.Second)
Expect(r.reconcile(secret)).Should(Succeed())
fedSecret := &v1beta1.FederatedSecret{}
By("Expecting to create federated secret successfully")
@@ -110,11 +114,11 @@ var (
return !fedSecret.CreationTimestamp.IsZero()
}, timeout, interval).Should(BeTrue())
err := ksCache.Get(context.Background(), client.ObjectKey{Name: secret.Name, Namespace: constants.NotificationSecretNamespace}, secret)
err = ksCache.Get(context.Background(), client.ObjectKey{Name: secret.Name, Namespace: constants.NotificationSecretNamespace}, secret)
Expect(err).Should(Succeed())
secret.StringData = map[string]string{"foo": "bar"}
Expect(cl.Update(context.Background(), secret)).Should(Succeed())
time.Sleep(time.Second)
Expect(r.reconcile(secret)).Should(Succeed())
By("Expecting to update federated secret successfully")
Eventually(func() bool {
@@ -134,7 +138,6 @@ var (
obj := &v2beta1.Config{
ObjectMeta: metav1.ObjectMeta{
Name: "foo",
Namespace: constants.NotificationSecretNamespace,
Labels: map[string]string{
"type": "default",
},
@@ -169,16 +172,19 @@ var (
informerCacheCancel()
})
// Add Tests for OpenAPI validation (or additonal CRD features) specified in
// Add Tests for OpenAPI validation (or additional CRD features) specified in
// your API definition.
// Avoid adding tests for vanilla CRUD operations because they would
// test Kubernetes API server, which isn't the goal here.
Context("Notification Controller", func() {
It("Should create successfully", func() {
// Create a bject
r, err := NewController(fakek8s.NewSimpleClientset(), cl, ksCache)
Expect(err).ToNot(HaveOccurred())
// Create a object
Expect(cl.Create(context.Background(), obj)).Should(Succeed())
time.Sleep(time.Second)
Expect(r.reconcile(obj)).Should(Succeed())
fedObj := &v1beta1.FederatedNotificationConfig{}
By("Expecting to create federated object successfully")
@@ -188,11 +194,11 @@ var (
return !fedObj.CreationTimestamp.IsZero()
}, timeout, interval).Should(BeTrue())
err := ksCache.Get(context.Background(), client.ObjectKey{Name: obj.Name}, obj)
err = ksCache.Get(context.Background(), client.ObjectKey{Name: obj.Name}, obj)
Expect(err).Should(Succeed())
obj.Labels = map[string]string{"foo": "bar"}
Expect(cl.Update(context.Background(), obj)).Should(Succeed())
time.Sleep(time.Second)
Expect(r.reconcile(obj)).Should(Succeed())
By("Expecting to update federated object successfully")
Eventually(func() bool {