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

Merge pull request #3696 from wanjunlei/notification

Browse Source 
fix bug member cluster can not view the notification channels
This commit is contained in:
KubeSphere CI Bot
2021-04-09 10:02:59 +08:00
committed by GitHub
parent 6a58572167 cbdc81a10a
commit 0688ca4472
3 changed files with 241 additions and 123 deletions
Download Patch File Download Diff File Expand all files Collapse all files

150
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"
@@ -47,8 +49,9 @@ import (
const (
// SuccessSynced is used as part of the Event 'reason' when a Foo is synced
successSynced = "Synced"
controllerName = "notification-controller"
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 operations.
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,27 @@ func (c *Controller) reconcile(obj interface{}) error {
}
name := accessor.GetName()
kind := runtimeObj.GetObjectKind().GroupVersionKind().String()
// The notification controller should update the annotations of secrets managed by itself
// whenever a cluster is added or deleted. This way, the controller will have a chance to override 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 +270,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 +326,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 +349,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 +384,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 +407,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 +442,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 +472,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
}
if err := c.Update(context.Background(), fedObj); err != nil {
klog.Errorf("update '%s:%s' failed, %s", fedObj.GetObjectKind().GroupVersionKind().String(), obj.Name, err)
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 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 managed by the notification controller to trigger 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)

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

@@ -25,9 +25,11 @@ 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"
"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"
@@ -55,6 +57,30 @@ var (
},
}
config := &v2beta1.Config{
ObjectMeta: metav1.ObjectMeta{
Name: "foo",
Labels: map[string]string{
"type": "global",
},
},
}
receiver := &v2beta1.Receiver{
ObjectMeta: metav1.ObjectMeta{
Name: "foo",
Labels: map[string]string{
"type": "default",
},
},
}
host := &v1alpha1.Cluster{
ObjectMeta: metav1.ObjectMeta{
Name: "host",
},
}
var (
cl client.Client
ksCache cache.Cache
@@ -91,16 +117,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 +139,12 @@ var (
return !fedSecret.CreationTimestamp.IsZero()
}, timeout, interval).Should(BeTrue())
err := ksCache.Get(context.Background(), client.ObjectKey{Name: secret.Name, Namespace: constants.NotificationSecretNamespace}, secret)
// Update a 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 {
@@ -122,83 +152,92 @@ var (
Expect(err).Should(Succeed())
return string(fedSecret.Spec.Template.Data["foo"]) == "bar"
}, timeout, interval).Should(BeTrue())
})
})
})
_ = Describe("Notification", func() {
// Create a Config
Expect(cl.Create(context.Background(), config)).Should(Succeed())
Expect(r.reconcile(config)).Should(Succeed())
const timeout = time.Second * 30
const interval = time.Second * 1
obj := &v2beta1.Config{
ObjectMeta: metav1.ObjectMeta{
Name: "foo",
Namespace: constants.NotificationSecretNamespace,
Labels: map[string]string{
"type": "default",
},
},
}
var (
cl client.Client
ksCache cache.Cache
informerCacheCtx context.Context
informerCacheCancel context.CancelFunc
)
BeforeEach(func() {
var err error
cl, err = client.New(cfg, client.Options{})
Expect(err).NotTo(HaveOccurred())
ksCache, err = cache.New(cfg, cache.Options{})
Expect(err).NotTo(HaveOccurred())
informerCacheCtx, informerCacheCancel = context.WithCancel(context.Background())
go func(ctx context.Context) {
defer GinkgoRecover()
Expect(ksCache.Start(ctx.Done())).To(Succeed())
}(informerCacheCtx)
Expect(ksCache.WaitForCacheSync(informerCacheCtx.Done())).To(BeTrue())
})
AfterEach(func() {
By("cleaning up")
informerCacheCancel()
})
// Add Tests for OpenAPI validation (or additonal 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
Expect(cl.Create(context.Background(), obj)).Should(Succeed())
time.Sleep(time.Second)
fedObj := &v1beta1.FederatedNotificationConfig{}
fedConfig := &v1beta1.FederatedNotificationConfig{}
By("Expecting to create federated object successfully")
Eventually(func() bool {
err := ksCache.Get(context.Background(), client.ObjectKey{Name: obj.Name}, fedObj)
err := ksCache.Get(context.Background(), client.ObjectKey{Name: config.Name}, fedConfig)
Expect(err).Should(Succeed())
return !fedObj.CreationTimestamp.IsZero()
return !fedConfig.CreationTimestamp.IsZero()
}, timeout, interval).Should(BeTrue())
err := ksCache.Get(context.Background(), client.ObjectKey{Name: obj.Name}, obj)
// Update a config
err = ksCache.Get(context.Background(), client.ObjectKey{Name: config.Name}, config)
Expect(err).Should(Succeed())
obj.Labels = map[string]string{"foo": "bar"}
Expect(cl.Update(context.Background(), obj)).Should(Succeed())
time.Sleep(time.Second)
config.Labels = map[string]string{"foo": "bar"}
Expect(cl.Update(context.Background(), config)).Should(Succeed())
Expect(r.reconcile(config)).Should(Succeed())
By("Expecting to update federated object successfully")
Eventually(func() bool {
err := ksCache.Get(context.Background(), client.ObjectKey{Name: obj.Name}, fedObj)
err := ksCache.Get(context.Background(), client.ObjectKey{Name: config.Name}, fedConfig)
Expect(err).Should(Succeed())
return fedObj.Spec.Template.Labels["foo"] == "bar"
return fedConfig.Spec.Template.Labels["foo"] == "bar"
}, timeout, interval).Should(BeTrue())
// Create a receiver
Expect(cl.Create(context.Background(), receiver)).Should(Succeed())
Expect(r.reconcile(receiver)).Should(Succeed())
fedReceiver := &v1beta1.FederatedNotificationReceiver{}
By("Expecting to create federated object successfully")
Eventually(func() bool {
err := ksCache.Get(context.Background(), client.ObjectKey{Name: receiver.Name}, fedReceiver)
Expect(err).Should(Succeed())
return !fedReceiver.CreationTimestamp.IsZero()
}, timeout, interval).Should(BeTrue())
// Update a receiver
err = ksCache.Get(context.Background(), client.ObjectKey{Name: receiver.Name}, receiver)
Expect(err).Should(Succeed())
receiver.Labels = map[string]string{"foo": "bar"}
Expect(cl.Update(context.Background(), receiver)).Should(Succeed())
Expect(r.reconcile(receiver)).Should(Succeed())
By("Expecting to update federated object successfully")
Eventually(func() bool {
err := ksCache.Get(context.Background(), client.ObjectKey{Name: receiver.Name}, fedReceiver)
Expect(err).Should(Succeed())
return fedReceiver.Spec.Template.Labels["foo"] == "bar"
}, timeout, interval).Should(BeTrue())
// Add a cluster
Expect(cl.Create(informerCacheCtx, host)).Should(Succeed())
Expect(r.reconcile(secret)).Should(Succeed())
By("Expecting to update federated secret successfully")
Eventually(func() bool {
err := ksCache.Get(context.Background(), client.ObjectKey{Name: secret.Name, Namespace: constants.NotificationSecretNamespace}, fedSecret)
Expect(err).Should(Succeed())
if fedSecret.Spec.Overrides == nil ||
len(fedSecret.Spec.Overrides) != 1 ||
fedSecret.Spec.Overrides[0].ClusterName != "host" {
return false
}
return true
}, timeout, interval).Should(BeTrue())
// Delete a cluster
Expect(cl.Delete(informerCacheCtx, host)).Should(Succeed())
Expect(r.reconcile(secret)).Should(Succeed())
By("Expecting to update federated secret successfully")
Eventually(func() bool {
err := ksCache.Get(context.Background(), client.ObjectKey{Name: secret.Name, Namespace: constants.NotificationSecretNamespace}, fedSecret)
Expect(err).Should(Succeed())
if fedSecret.Spec.Overrides != nil ||
len(fedSecret.Spec.Overrides) != 0 {
return false
}
return true
}, timeout, interval).Should(BeTrue())
})
})