feat: multiple clusters filter

multicluster/clustersmanager.go

@@ -0,0 +1,106 @@
+/*
+Copyright 2023 The Katanomi 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 multicluster
+
+import (
+	"context"
+	"sync"
+
+	"github.com/katanomi/pkg/parallel"
+	"github.com/katanomi/pkg/scheme"
+	corev1 "k8s.io/api/core/v1"
+	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
+	"knative.dev/pkg/logging"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+// ClusterFilter for filter cluster reousrces
+type ClusterFilter func(ctx context.Context, clusterRef corev1.ObjectReference) bool
+
+// ClusterManager manages clusters collection by filters
+type ClusterManager struct {
+	Concurrent int
+	Filters    []ClusterFilter
+}
+
+// FilterClusters returns a filtered list of clusters
+func (m *ClusterManager) FilterClusters(ctx context.Context, clusterRefs []corev1.ObjectReference) []corev1.ObjectReference {
+	if len(m.Filters) == 0 {
+		return clusterRefs
+	}
+
+	var filteredClusters []corev1.ObjectReference
+	mutex := sync.Mutex{}
+	addCluster := func(clusterRef corev1.ObjectReference) {
+		mutex.Lock()
+		defer mutex.Unlock()
+		filteredClusters = append(filteredClusters, clusterRef)
+	}
+
+	p := parallel.P(logging.FromContext(ctx), "filterClusters")
+
+	taskFunc := func(objRef corev1.ObjectReference) func() (interface{}, error) {
+		return func() (interface{}, error) {
+			for _, opt := range m.Filters {
+				if !opt(ctx, objRef) {
+					return nil, nil
+				}
+			}
+			addCluster(objRef)
+			return nil, nil
+		}
+	}
+
+	for _, objRef := range clusterRefs {
+		p.Add(taskFunc(objRef))
+	}
+
+	concurrent := m.Concurrent
+	if concurrent == 0 {
+		concurrent = parallel.DefaultConcurrentNum
+	}
+
+	p.Context(ctx).SetConcurrent(concurrent).Do().Wait()
+
+	return filteredClusters
+}
+
+// CustomResourceDefinitionExists returns true if the CRD exists in the cluster
+func CustomResourceDefinitionExists(cliGetter ClientGetter, CRDName string) ClusterFilter {
+	return func(ctx context.Context, clusterRef corev1.ObjectReference) bool {
+		logger := logging.FromContext(ctx)
+
+		if cliGetter == nil {
+			logger.Errorw("clientGetter is nil")
+			return false
+		}
+
+		clusterCli, err := cliGetter.GetClient(ctx, &clusterRef, scheme.Scheme(ctx))
+		if err != nil {
+			logger.Errorw("failed to get cluster client", "cluster", clusterRef, "error", err)
+			return false
+		}
+		crd := &apiextensionsv1.CustomResourceDefinition{}
+		err = clusterCli.Get(ctx, client.ObjectKey{Name: CRDName}, crd)
+		if err != nil {
+			logger.Errorw("failed to get crd", "crd", CRDName, "error", err)
+			return false
+		}
+		logger.Debugw("crd exists in cluster", "crd", CRDName, "cluster", clusterRef.Name)
+		return true
+	}
+}

multicluster/clustersmanager_test.go

@@ -0,0 +1,155 @@
+/*
+Copyright 2023 The Katanomi 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 multicluster
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/golang/mock/gomock"
+	pkgscheme "github.com/katanomi/pkg/scheme"
+	pkgtesting "github.com/katanomi/pkg/testing"
+	multiclustertesting "github.com/katanomi/pkg/testing/mock/github.com/katanomi/pkg/multicluster"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	corev1 "k8s.io/api/core/v1"
+	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+)
+
+var _ = Describe("FilterClusters", func() {
+	var (
+		clusterRefs         []corev1.ObjectReference
+		filteredClusterRefs []corev1.ObjectReference
+		ctx                 context.Context
+
+		mgr *ClusterManager
+	)
+
+	BeforeEach(func() {
+		ctx = context.TODO()
+		pkgtesting.MustLoadYaml("testdata/clusterreferences.init.yaml", &clusterRefs)
+
+		mgr = &ClusterManager{
+			Concurrent: 1,
+			Filters:    []ClusterFilter{},
+		}
+
+	})
+
+	JustBeforeEach(func() {
+		filteredClusterRefs = mgr.FilterClusters(ctx, clusterRefs)
+	})
+
+	Context("empty filters", func() {
+		It("should return all clusters", func() {
+			Expect(filteredClusterRefs).To(HaveLen(4))
+		})
+	})
+
+	Context("filter by namespace", func() {
+		BeforeEach(func() {
+			mgr.Filters = append(mgr.Filters, func(ctx context.Context, objRef corev1.ObjectReference) bool {
+				return objRef.Namespace == "default"
+			})
+		})
+
+		It("should return only cluster in default namespace", func() {
+			Expect(filteredClusterRefs).To(HaveLen(2))
+		})
+	})
+
+})
+
+var _ = Describe("ClusterFilterManager.FilterClusters.CustomResourceDefinitionExists", func() {
+
+	var (
+		clusterRefs         []corev1.ObjectReference
+		filteredClusterRefs []corev1.ObjectReference
+		ctx                 context.Context
+
+		mockCtl       *gomock.Controller
+		mockCliGetter *multiclustertesting.MockInterface
+		scheme        *runtime.Scheme
+		crd           *apiextensionsv1.CustomResourceDefinition
+
+		mgr *ClusterManager
+	)
+
+	BeforeEach(func() {
+		ctx = context.TODO()
+		pkgtesting.MustLoadYaml("testdata/clusterreferences.init.yaml", &clusterRefs)
+
+		mockCtl = gomock.NewController(GinkgoT())
+		mockCliGetter = multiclustertesting.NewMockInterface(mockCtl)
+
+		crd = &apiextensionsv1.CustomResourceDefinition{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "customresourcedefinitions.apps.k8s.io",
+			},
+		}
+
+		scheme = runtime.NewScheme()
+		apiextensionsv1.AddToScheme(scheme)
+
+		ctx = pkgscheme.WithScheme(ctx, scheme)
+
+		mgr = &ClusterManager{
+			Concurrent: 1,
+			Filters:    []ClusterFilter{},
+		}
+	})
+
+	JustBeforeEach(func() {
+		mgr.Filters = append(mgr.Filters, CustomResourceDefinitionExists(mockCliGetter, "customresourcedefinitions.apps.k8s.io"))
+		filteredClusterRefs = mgr.FilterClusters(ctx, clusterRefs)
+	})
+
+	Context("error with client getter", func() {
+
+		BeforeEach(func() {
+			mockCliGetter.EXPECT().GetClient(ctx, gomock.Any(), gomock.Any()).Return(nil, fmt.Errorf("get client err")).AnyTimes()
+		})
+
+		It("should return no clusters", func() {
+			Expect(filteredClusterRefs).To(HaveLen(0))
+		})
+	})
+
+	Context("crd exists in cluster", func() {
+
+		BeforeEach(func() {
+			mockClientWithCRD := fake.NewClientBuilder().WithScheme(scheme).WithObjects(crd).Build()
+			mockClientWithoutCRD := fake.NewClientBuilder().WithScheme(scheme).Build()
+
+			gomock.InOrder(
+				mockCliGetter.EXPECT().GetClient(ctx, gomock.Any(), gomock.Any()).Return(mockClientWithCRD, nil),
+				mockCliGetter.EXPECT().GetClient(ctx, gomock.Any(), gomock.Any()).Return(mockClientWithoutCRD, nil),
+				mockCliGetter.EXPECT().GetClient(ctx, gomock.Any(), gomock.Any()).Return(mockClientWithoutCRD, nil),
+				mockCliGetter.EXPECT().GetClient(ctx, gomock.Any(), gomock.Any()).Return(mockClientWithCRD, nil),
+			)
+
+		})
+
+		It("should return clusters with crd", func() {
+			Expect(filteredClusterRefs).To(HaveLen(2))
+		})
+	})
+})

multicluster/filter_test.go

@@ -29,6 +29,8 @@ import (
 	"github.com/katanomi/pkg/testing/mock/github.com/katanomi/pkg/multicluster"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
+	"k8s.io/apimachinery/pkg/api/meta"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apiserver/pkg/authentication/user"
 	apiserverrequest "k8s.io/apiserver/pkg/endpoints/request"
 	"k8s.io/client-go/rest"
@@ -48,6 +50,7 @@ var _ = Describe("TestCrossClusterSubjectReview_GetClient", func() {
 		crossReview = &CrossClusterSubjectReview{
 			ClusterParameter: "cluster",
 			ClusterNamespace: "test",
+			restMapper:       meta.NewDefaultRESTMapper([]schema.GroupVersion{}),
 		}
 		req = restful.NewRequest(&http.Request{})
 		clt = nil

multicluster/interface.go

@@ -33,7 +33,6 @@ import (
 // Interface interface for a multi-cluster functionality
 type Interface interface {
 	GetConfig(ctx context.Context, clusterRef *corev1.ObjectReference) (config *rest.Config, err error)
-	GetClient(ctx context.Context, clusterRef *corev1.ObjectReference, scheme *runtime.Scheme) (clt client.Client, err error)
 	GetDynamic(ctx context.Context, clusterRef *corev1.ObjectReference) (dyn dynamic.Interface, err error)
 	GetConfigFromCluster(ctx context.Context, cluster *unstructured.Unstructured) (config *rest.Config, err error)
 
@@ -42,6 +41,20 @@ type Interface interface {
 	ListClustersNamespaces(ctx context.Context, namespace string) (clusterNamespaces map[*corev1.ObjectReference][]corev1.Namespace, err error)
 	// StartWarmUpClientCache used to start warming the client cache, only needs to be called once.
 	StartWarmUpClientCache(ctx context.Context)
-	// GetNamespaceClusters returns a list of clusters related by special namespace
-	GetNamespaceClusters(ctx context.Context, namespace string) (clusters []corev1.ObjectReference, err error)
+
+	// ClientGetter for getting client for a clusterRef and given scheme
+	ClientGetter
+
+	// NamespaceClustersGetter for getting list of clusters related by special namespace
+	NamespaceClustersGetter
+}
+
+// NamespaceClustersGetter interface get list of clusters related by special namespace
+type NamespaceClustersGetter interface {
+	GetNamespaceClusters(ctx context.Context, namespace string) ([]corev1.ObjectReference, error)
+}
+
+// ClientGetter interface get client for a clusterRef and given scheme
+type ClientGetter interface {
+	GetClient(ctx context.Context, clusterRef *corev1.ObjectReference, scheme *runtime.Scheme) (clt client.Client, err error)
 }

multicluster/multicluster_suite_test.go

@@ -0,0 +1,29 @@
+/*
+Copyright 2023 The Katanomi 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 multicluster
+
+import (
+	"testing"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+func TestMulticluster(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Multicluster Suite")
+}

multicluster/testdata/clusterreferences.init.yaml

@@ -0,0 +1,16 @@
+- apiVersion: clusterregistry.k8s.io/v1alpha1
+  kind: Cluster
+  namespace: default
+  name: my-cluster
+- apiVersion: clusterregistry.k8s.io/v1alpha1
+  kind: Cluster
+  namespace: default
+  name: other-cluster
+- apiVersion: clusterregistry.k8s.io/v1alpha1
+  kind: Cluster
+  namespace: foo
+  name: cluster1
+- apiVersion: clusterregistry.k8s.io/v1alpha1
+  kind: Cluster
+  namespace: bar
+  name: cluster2

parallel/task.go

@@ -29,6 +29,11 @@ import (
 	"k8s.io/apimachinery/pkg/util/errors"
 )
 
+const (
+	// DefaultConcurrentNum for default concurrent number of parallel
+	DefaultConcurrentNum = 10
+)
+
 type Task func() (interface{}, error)
 
 // ParallelTasks will construct a parallel tasks struct

secret/mutate_test.go

@@ -19,10 +19,11 @@ package secret
 import (
 	"context"
 	"errors"
+	"testing"
+
 	. "github.com/onsi/gomega"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"testing"
 )
 
 // MockMutator is a mock implementation of the Mutator interface for testing.