Repository: clusternet/clusternet
Pull request #48 (2021-08-05 10:01:22)

notify base handler when manifests get changed
<!--  Thanks for sending a pull request!

-->

#### What type of PR is this?
kind/feature

#### What this PR does / why we need it:
- add labels to `Manifest` and `Base`
- when `Manifest` gets changed, `Base` handler will be called to re-populate/update `Description`.

#### Which issue(s) this PR fixes:
<!--
*Automatically closes linked issue when PR is merged.
Usage: `Fixes #<issue number>`, or `Fixes (paste link of issue)`.
-->
Fixes #

#### Special notes for your reviewer:



Patch:
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 1fb3f78..d560f0d 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -14,7 +14,7 @@ jobs:
       run:
         working-directory: ${{ env.GOPATH }}/src/github.com/${{ github.repository }}
     strategy:
-      max-parallel: 2
+      max-parallel: 3
       ## this will contain a matrix of all of the combinations
       ## we wish to test again:
       matrix:
diff --git a/pkg/controllers/apps/base/base.go b/pkg/controllers/apps/base/base.go
index 6984df2..972ba0c 100644
--- a/pkg/controllers/apps/base/base.go
+++ b/pkg/controllers/apps/base/base.go
@@ -18,6 +18,7 @@ package base
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"reflect"
 	"time"
@@ -32,11 +33,12 @@ import (
 	"k8s.io/client-go/tools/record"
 	"k8s.io/client-go/util/workqueue"
 	"k8s.io/klog/v2"
+	utilpointer "k8s.io/utils/pointer"
 
 	appsapi "github.com/clusternet/clusternet/pkg/apis/apps/v1alpha1"
-	clusternetClientSet "github.com/clusternet/clusternet/pkg/generated/clientset/versioned"
-	appInformers "github.com/clusternet/clusternet/pkg/generated/informers/externalversions/apps/v1alpha1"
-	appListers "github.com/clusternet/clusternet/pkg/generated/listers/apps/v1alpha1"
+	clusternetclientset "github.com/clusternet/clusternet/pkg/generated/clientset/versioned"
+	appinformers "github.com/clusternet/clusternet/pkg/generated/informers/externalversions/apps/v1alpha1"
+	applisters "github.com/clusternet/clusternet/pkg/generated/listers/apps/v1alpha1"
 	"github.com/clusternet/clusternet/pkg/known"
 	"github.com/clusternet/clusternet/pkg/utils"
 )
@@ -50,7 +52,7 @@ type SyncHandlerFunc func(orig *appsapi.Base) error
 type Controller struct {
 	ctx context.Context
 
-	clusternetClient clusternetClientSet.Interface
+	clusternetClient clusternetclientset.Interface
 
 	// workqueue is a rate limited work queue. This is used to queue work to be
 	// processed instead of performing it as soon as a change happens. This
@@ -59,7 +61,7 @@ type Controller struct {
 	// simultaneously in two different workers.
 	workqueue workqueue.RateLimitingInterface
 
-	baseLister appListers.BaseLister
+	baseLister applisters.BaseLister
 	baseSynced cache.InformerSynced
 
 	recorder record.EventRecorder
@@ -67,8 +69,8 @@ type Controller struct {
 	SyncHandler SyncHandlerFunc
 }
 
-func NewController(ctx context.Context, clusternetClient clusternetClientSet.Interface,
-	baseInformer appInformers.BaseInformer, descInformer appInformers.DescriptionInformer,
+func NewController(ctx context.Context, clusternetClient clusternetclientset.Interface,
+	baseInformer appinformers.BaseInformer, descInformer appinformers.DescriptionInformer,
 	recorder record.EventRecorder, syncHandler SyncHandlerFunc) (*Controller, error) {
 	if syncHandler == nil {
 		return nil, fmt.Errorf("syncHandler must be set")
@@ -148,6 +150,17 @@ func (c *Controller) addBase(obj interface{}) {
 		}
 	}
 
+	// label Base self uid
+	if val, ok := base.Labels[string(base.UID)]; !ok || val != controllerKind.Kind {
+		err := c.patchBaseLabels(base, map[string]*string{
+			string(base.UID): utilpointer.StringPtr(controllerKind.Kind),
+		})
+		if err != nil {
+			klog.ErrorDepth(5, fmt.Sprintf("failed to patch Base labels: %v", err))
+			c.addBase(obj)
+		}
+	}
+
 	c.enqueue(base)
 }
 
@@ -167,6 +180,18 @@ func (c *Controller) updateBase(old, cur interface{}) {
 	}
 
 	klog.V(4).Infof("updating Base %q", klog.KObj(oldBase))
+
+	// label Base self uid
+	if val, ok := newBase.Labels[string(newBase.UID)]; !ok || val != controllerKind.Kind {
+		err := c.patchBaseLabels(newBase, map[string]*string{
+			string(newBase.UID): utilpointer.StringPtr(controllerKind.Kind),
+		})
+		if err != nil {
+			klog.ErrorDepth(5, fmt.Sprintf("failed to patch Base labels: %v", err))
+			c.updateBase(old, cur)
+		}
+	}
+
 	c.enqueue(newBase)
 }
 
@@ -345,3 +370,27 @@ func (c *Controller) enqueue(base *appsapi.Base) {
 	}
 	c.workqueue.Add(key)
 }
+
+type LabelOption struct {
+	Meta Meta `json:"metadata"`
+}
+
+type Meta struct {
+	Labels map[string]*string `json:"labels"`
+}
+
+func (c *Controller) patchBaseLabels(base *appsapi.Base, labels map[string]*string) error {
+	klog.V(5).Infof("patching Base labels")
+	option := LabelOption{Meta: Meta{Labels: labels}}
+	patchData, err := json.Marshal(option)
+	if err != nil {
+		return err
+	}
+
+	_, err = c.clusternetClient.AppsV1alpha1().Bases(base.Namespace).Patch(context.TODO(),
+		base.Name,
+		types.MergePatchType,
+		patchData,
+		metav1.PatchOptions{})
+	return err
+}
diff --git a/pkg/controllers/apps/manifest/manifest.go b/pkg/controllers/apps/manifest/manifest.go
index 8a6ce29..f6ba4ba 100644
--- a/pkg/controllers/apps/manifest/manifest.go
+++ b/pkg/controllers/apps/manifest/manifest.go
@@ -22,10 +22,13 @@ import (
 	"reflect"
 	"time"
 
+	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/tools/record"
 	"k8s.io/client-go/util/workqueue"
 	"k8s.io/klog/v2"
 
@@ -33,6 +36,8 @@ import (
 	clusternetClientSet "github.com/clusternet/clusternet/pkg/generated/clientset/versioned"
 	appInformers "github.com/clusternet/clusternet/pkg/generated/informers/externalversions/apps/v1alpha1"
 	appListers "github.com/clusternet/clusternet/pkg/generated/listers/apps/v1alpha1"
+	"github.com/clusternet/clusternet/pkg/known"
+	"github.com/clusternet/clusternet/pkg/utils"
 )
 
 // controllerKind contains the schema.GroupVersionKind for this controller type.
@@ -56,12 +61,14 @@ type Controller struct {
 	manifestLister appListers.ManifestLister
 	manifestSynced cache.InformerSynced
 
+	recorder record.EventRecorder
+
 	SyncHandler SyncHandlerFunc
 }
 
 func NewController(ctx context.Context, clusternetClient clusternetClientSet.Interface,
 	manifestInformer appInformers.ManifestInformer,
-	syncHandler SyncHandlerFunc) (*Controller, error) {
+	recorder record.EventRecorder, syncHandler SyncHandlerFunc) (*Controller, error) {
 	if syncHandler == nil {
 		return nil, fmt.Errorf("syncHandler must be set")
 	}
@@ -72,6 +79,7 @@ func NewController(ctx context.Context, clusternetClient clusternetClientSet.Int
 		workqueue:        workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "manifest"),
 		manifestLister:   manifestInformer.Lister(),
 		manifestSynced:   manifestInformer.Informer().HasSynced,
+		recorder:         recorder,
 		SyncHandler:      syncHandler,
 	}
 
@@ -114,6 +122,27 @@ func (c *Controller) Run(workers int, stopCh <-chan struct{}) {
 func (c *Controller) addManifest(obj interface{}) {
 	manifest := obj.(*appsapi.Manifest)
 	klog.V(4).Infof("adding Manifest %q", klog.KObj(manifest))
+
+	// add finalizer
+	if manifest.DeletionTimestamp == nil {
+		if !utils.ContainsString(manifest.Finalizers, known.AppFinalizer) {
+			manifest.Finalizers = append(manifest.Finalizers, known.AppFinalizer)
+		}
+		_, err := c.clusternetClient.AppsV1alpha1().Manifests(manifest.Namespace).Update(context.TODO(),
+			manifest, metav1.UpdateOptions{})
+		if err == nil {
+			msg := fmt.Sprintf("successfully inject finalizer %s to Manifest %s", known.AppFinalizer, klog.KObj(manifest))
+			klog.V(4).Info(msg)
+			c.recorder.Event(manifest, corev1.EventTypeNormal, "FinalizerInjected", msg)
+		} else {
+			msg := fmt.Sprintf("failed to inject finalizer %s to Manifest %s: %v", known.AppFinalizer, klog.KObj(manifest), err)
+			klog.WarningDepth(4, msg)
+			c.recorder.Event(manifest, corev1.EventTypeWarning, "FailedInjectingFinalizer", msg)
+			c.addManifest(obj)
+			return
+		}
+	}
+
 	c.enqueue(manifest)
 }
 
diff --git a/pkg/hub/deployer/deployer.go b/pkg/hub/deployer/deployer.go
index 22f9d25..1dc3685 100644
--- a/pkg/hub/deployer/deployer.go
+++ b/pkg/hub/deployer/deployer.go
@@ -18,14 +18,17 @@ package deployer
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"reflect"
+	"sync"
 
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
 	utilerrors "k8s.io/apimachinery/pkg/util/errors"
 	"k8s.io/apimachinery/pkg/util/sets"
 	kubeinformers "k8s.io/client-go/informers"
@@ -146,6 +149,7 @@ func NewDeployer(ctx context.Context, kubeclient *kubernetes.Clientset, clustern
 	mfstController, err := manifest.NewController(ctx,
 		clusternetclient,
 		clusternetInformerFactory.Apps().V1alpha1().Manifests(),
+		deployer.recorder,
 		deployer.handleManifest)
 	if err != nil {
 		return nil, err
@@ -391,6 +395,11 @@ func (deployer *Deployer) syncBase(sub *appsapi.Subscription, base *appsapi.Base
 }
 
 func (deployer *Deployer) handleBase(base *appsapi.Base) error {
+	// patch manifest labels
+	if err := deployer.patchLabelsToReferredManifests(base); err != nil {
+		return err
+	}
+
 	if base.DeletionTimestamp != nil {
 		descs, err := deployer.descLister.List(labels.SelectorFromSet(labels.Set{
 			known.ConfigKindLabel:      baseKind.Kind,
@@ -657,7 +666,130 @@ func (deployer *Deployer) handleDescription(desc *appsapi.Description) error {
 }
 
 func (deployer *Deployer) handleManifest(manifest *appsapi.Manifest) error {
-	// TODO: @dixudx
+	if manifest.DeletionTimestamp != nil {
+		// TODO: other strategies
 
-	return nil
+		// remove finalizer
+		manifest.Finalizers = utils.RemoveString(manifest.Finalizers, known.AppFinalizer)
+		_, err := deployer.clusternetclient.AppsV1alpha1().Manifests(manifest.Namespace).Update(context.TODO(), manifest, metav1.UpdateOptions{})
+		if err != nil {
+			klog.WarningDepth(4,
+				fmt.Sprintf("failed to remove finalizer %s from Manifest %s: %v", known.AppFinalizer, klog.KObj(manifest), err))
+		}
+		return err
+	}
+
+	// find all referred Base UIDs
+	var baseUIDs []string
+	for key, val := range manifest.Labels {
+		if val == baseKind.Kind {
+			baseUIDs = append(baseUIDs, key)
+		}
+	}
+
+	wg := sync.WaitGroup{}
+	wg.Add(len(baseUIDs))
+	errCh := make(chan error, len(baseUIDs))
+	for _, baseuid := range baseUIDs {
+		bases, err := deployer.baseLister.List(labels.SelectorFromSet(labels.Set{
+			baseuid: baseKind.Kind,
+		}))
+
+		if err != nil {
+			return err
+		}
+
+		go func() {
+			defer wg.Done()
+			if len(bases) == 0 {
+				return
+			}
+			// here the length should always be 1
+			if err := deployer.populateDescriptions(bases[0]); err != nil {
+				errCh <- err
+			}
+		}()
+	}
+
+	wg.Wait()
+
+	// collect errors
+	close(errCh)
+	var allErrs []error
+	for err := range errCh {
+		allErrs = append(allErrs, err)
+	}
+	return utilerrors.NewAggregate(allErrs)
+}
+
+func (deployer *Deployer) patchLabelsToReferredManifests(b *appsapi.Base) error {
+	var allManifests []*appsapi.Manifest
+	var allErrs []error
+	for _, feed := range b.Spec.Feeds {
+		if feed.Kind != helmChartKind.Kind {
+			manifests, err := deployer.getManifestsBySelector(b, feed)
+			if err == nil {
+				allManifests = append(allManifests, manifests...)
+			}
+
+			if errors.IsNotFound(err) {
+				continue
+			}
+			if err != nil {
+				allErrs = append(allErrs, err)
+			}
+
+		}
+	}
+
+	wg := sync.WaitGroup{}
+	wg.Add(len(allManifests))
+	errCh := make(chan error, len(allManifests))
+	for _, manifest := range allManifests {
+		go func(manifest *appsapi.Manifest) {
+			defer wg.Done()
+
+			curVal, ok := manifest.Labels[string(b.UID)]
+			var labelsToPatch map[string]*string
+			if b.DeletionTimestamp != nil {
+				if !ok {
+					return
+				}
+				labelsToPatch[string(b.UID)] = nil
+			} else {
+				if curVal == baseKind.Kind {
+					return
+				}
+				labelsToPatch[string(b.UID)] = utilpointer.StringPtr(baseKind.Kind)
+			}
+
+			if err := deployer.patchManifestLabels(manifest, labelsToPatch); err != nil {
+				errCh <- err
+			}
+		}(manifest)
+	}
+
+	wg.Wait()
+
+	// collect errors
+	close(errCh)
+	for err := range errCh {
+		allErrs = append(allErrs, err)
+	}
+	return utilerrors.NewAggregate(allErrs)
+}
+
+func (deployer *Deployer) patchManifestLabels(manifest *appsapi.Manifest, labels map[string]*string) error {
+	labelOption := base.LabelOption{Meta: base.Meta{Labels: labels}}
+	patchData, err := json.Marshal(labelOption)
+	if err != nil {
+		return err
+	}
+
+	_, err = deployer.clusternetclient.AppsV1alpha1().Manifests(manifest.Namespace).Patch(context.TODO(),
+		manifest.Name,
+		types.MergePatchType,
+		patchData,
+		metav1.PatchOptions{})
+	return err
 }
diff --git a/pkg/hub/hub.go b/pkg/hub/hub.go
index e683936..b542975 100644
--- a/pkg/hub/hub.go
+++ b/pkg/hub/hub.go
@@ -90,23 +90,27 @@ func NewHub(ctx context.Context, opts *options.HubServerOptions) (*Hub, error) {
 		return nil, err
 	}
 
-	deployer, err := deployer.NewDeployer(ctx, kubeclient, clusternetclient, clusternetInformerFactory, kubeInformerFactory)
-	if err != nil {
-		return nil, err
-	}
-
-	// add informers
+	// add informers for minimum requirements
 	kubeInformerFactory.Core().V1().Namespaces().Informer()
 	kubeInformerFactory.Core().V1().ServiceAccounts().Informer()
 	kubeInformerFactory.Core().V1().Secrets().Informer()
 	clusternetInformerFactory.Clusters().V1beta1().ClusterRegistrationRequests().Informer()
 	clusternetInformerFactory.Clusters().V1beta1().ManagedClusters().Informer()
 	crdInformerFactory.Apiextensions().V1().CustomResourceDefinitions().Informer()
+
+	var d *deployer.Deployer
 	if deployerEnabled {
+		clusternetInformerFactory.Apps().V1alpha1().Manifests().Informer()
+		clusternetInformerFactory.Apps().V1alpha1().Bases().Informer()
 		clusternetInformerFactory.Apps().V1alpha1().Subscriptions().Informer()
 		clusternetInformerFactory.Apps().V1alpha1().HelmCharts().Informer()
 		clusternetInformerFactory.Apps().V1alpha1().Descriptions().Informer()
 		clusternetInformerFactory.Apps().V1alpha1().HelmReleases().Informer()
+
+		d, err = deployer.NewDeployer(ctx, kubeclient, clusternetclient, clusternetInformerFactory, kubeInformerFactory)
+		if err != nil {
+			return nil, err
+		}
 	}
 
 	if utilfeature.DefaultFeatureGate.Enabled(features.ShadowAPI) {
@@ -124,7 +128,7 @@ func NewHub(ctx context.Context, opts *options.HubServerOptions) (*Hub, error) {
 		kubeInformerFactory:       kubeInformerFactory,
 		crdInformerFactory:        crdInformerFactory,
 		socketConnection:          socketConnection,
-		deployer:                  deployer,
+		deployer:                  d,
 		deployerEnabled:           deployerEnabled,
 	}
 
diff --git a/pkg/registry/shadow/template/rest.go b/pkg/registry/shadow/template/rest.go
index da19f12..7bf6dfd 100644
--- a/pkg/registry/shadow/template/rest.go
+++ b/pkg/registry/shadow/template/rest.go
@@ -94,7 +94,7 @@ func (r *REST) Create(ctx context.Context, obj runtime.Object, createValidation
 	// next we create manifest to store the result
 	manifest := &appsapi.Manifest{
 		ObjectMeta: metav1.ObjectMeta{
-			Name:      r.generateNameForManifest(result.GetName()),
+			Name:      r.generateNameForManifest(result.GetNamespace(), result.GetName()),
 			Namespace: appsapi.ReservedNamespace,
 			Labels: map[string]string{
 				known.ConfigGroupLabel:     r.group,
@@ -123,9 +123,11 @@ func (r *REST) Get(ctx context.Context, name string, options *metav1.GetOptions)
 	var manifest *appsapi.Manifest
 	var err error
 	if len(options.ResourceVersion) == 0 {
-		manifest, err = r.clusternetInformerFactory.Apps().V1alpha1().Manifests().Lister().Manifests(appsapi.ReservedNamespace).Get(r.generateNameForManifest(name))
+		manifest, err = r.clusternetInformerFactory.Apps().V1alpha1().Manifests().Lister().Manifests(appsapi.ReservedNamespace).
+			Get(r.generateNameForManifest(request.NamespaceValue(ctx), name))
 	} else {
-		manifest, err = r.clusternetClient.AppsV1alpha1().Manifests(appsapi.ReservedNamespace).Get(ctx, r.generateNameForManifest(name), *options)
+		manifest, err = r.clusternetClient.AppsV1alpha1().Manifests(appsapi.ReservedNamespace).
+			Get(ctx, r.generateNameForManifest(request.NamespaceValue(ctx), name), *options)
 	}
 	if err != nil {
 		if errors.IsNotFound(err) {
@@ -152,7 +154,8 @@ func (r *REST) Update(ctx context.Context, name string, objInfo rest.UpdatedObje
 		return nil, false, err
 	}
 
-	manifest, err := r.clusternetInformerFactory.Apps().V1alpha1().Manifests().Lister().Manifests(appsapi.ReservedNamespace).Get(r.generateNameForManifest(name))
+	manifest, err := r.clusternetInformerFactory.Apps().V1alpha1().Manifests().Lister().Manifests(appsapi.ReservedNamespace).
+		Get(r.generateNameForManifest(request.NamespaceValue(ctx), name))
 	if err != nil {
 		if errors.IsNotFound(err) {
 			return nil, false, errors.NewNotFound(schema.GroupResource{Group: r.group, Resource: r.name}, name)
@@ -191,7 +194,8 @@ func (r *REST) Update(ctx context.Context, name string, objInfo rest.UpdatedObje
 // Delete removes the item from storage.
 // options can be mutated by rest.BeforeDelete due to a graceful deletion strategy.
 func (r *REST) Delete(ctx context.Context, name string, deleteValidation rest.ValidateObjectFunc, options *metav1.DeleteOptions) (runtime.Object, bool, error) {
-	err := r.clusternetClient.AppsV1alpha1().Manifests(appsapi.ReservedNamespace).Delete(ctx, r.generateNameForManifest(name), *options)
+	err := r.clusternetClient.AppsV1alpha1().Manifests(appsapi.ReservedNamespace).
+		Delete(ctx, r.generateNameForManifest(request.NamespaceValue(ctx), name), *options)
 	if err != nil {
 		if errors.IsNotFound(err) {
 			err = errors.NewNotFound(schema.GroupResource{Group: r.group, Resource: r.name}, name)
@@ -447,8 +451,11 @@ func (r *REST) normalizeRequest(req *clientgorest.Request, namespace string) *cl
 	return req
 }
 
-func (r *REST) generateNameForManifest(name string) string {
+func (r *REST) generateNameForManifest(namespace, name string) string {
 	resource, _ := r.getResourceName()
+	if len(namespace) > 0 {
+		return fmt.Sprintf("%s-%s-%s", resource, namespace, name)
+	}
 	return fmt.Sprintf("%s-%s", resource, name)
 }
 


Test patch:
diff --git a/pkg/controllers/apps/base/base_test.go b/pkg/controllers/apps/base/base_test.go
new file mode 100644
index 0000000..6f9cf76
--- /dev/null
+++ b/pkg/controllers/apps/base/base_test.go
@@ -0,0 +1,59 @@
+/*
+Copyright 2021 The Clusternet 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 base
+
+import (
+	"encoding/json"
+	"testing"
+
+	utilpointer "k8s.io/utils/pointer"
+)
+
+func TestMarshalLabelOption(t *testing.T) {
+	tests := []struct {
+		name        string
+		labelOption LabelOption
+		want        string
+	}{
+		{
+			name: "add/update label",
+			labelOption: LabelOption{Meta: Meta{Labels: map[string]*string{
+				"key": utilpointer.StringPtr("val"),
+			}}},
+			want: `{"metadata":{"labels":{"key":"val"}}}`,
+		},
+		{
+			name: "remove label",
+			labelOption: LabelOption{Meta: Meta{Labels: map[string]*string{
+				"key": nil,
+			}}},
+			want: `{"metadata":{"labels":{"key":null}}}`,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := json.Marshal(tt.labelOption)
+			if err != nil {
+				t.Errorf("failed to marshal LabelOption: error = %v", err)
+				return
+			}
+			if string(got) != tt.want {
+				t.Errorf("got: %s, want: %s", string(got), tt.want)
+			}
+		})
+	}
+}