make servicetopology filter in yurthub work properly when service or nodepool change

charts/openyurt/templates/yurt-controller-manager.yaml

@@ -95,6 +95,40 @@ rules:
       - kubernetes.io/kubelet-serving
     verbs:
       - approve
+  - apiGroups:
+      - discovery.k8s.io
+    resources:
+      - endpointslices
+    verbs:
+      - get
+      - list
+      - watch
+      - patch
+  - apiGroups:
+      - ""
+    resources:
+      - endpoints
+    verbs:
+      - get
+      - list
+      - watch
+      - patch
+  - apiGroups:
+      - "apps.openyurt.io"
+    resources:
+      - nodepools
+    verbs:
+      - get
+      - list
+      - watch
+  - apiGroups:
+      - ""
+    resources:
+      - services
+    verbs:
+      - get
+      - list
+      - watch
 ---
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRoleBinding

cmd/yurt-controller-manager/app/controllermanager.go

@@ -54,6 +54,8 @@ import (
 	"github.com/openyurtio/openyurt/cmd/yurt-controller-manager/app/options"
 	yurtctrlmgrconfig "github.com/openyurtio/openyurt/pkg/controller/apis/config"
 	"github.com/openyurtio/openyurt/pkg/projectinfo"
+	yurtclientset "github.com/openyurtio/yurt-app-manager-api/pkg/yurtappmanager/client/clientset/versioned"
+	yurtinformers "github.com/openyurtio/yurt-app-manager-api/pkg/yurtappmanager/client/informers/externalversions"
 )
 
 const (
@@ -204,6 +206,7 @@ func Run(c *config.CompletedConfig, stopCh <-chan struct{}) error {
 		}
 
 		controllerContext.InformerFactory.Start(controllerContext.Stop)
+		controllerContext.YurtInformerFactory.Start(controllerContext.Stop)
 		close(controllerContext.InformersStarted)
 
 		select {}
@@ -259,6 +262,9 @@ type ControllerContext struct {
 	// InformerFactory gives access to informers for the controller.
 	InformerFactory informers.SharedInformerFactory
 
+	// YurtInformerFactory gives access to yurt informers for the controller.
+	YurtInformerFactory yurtinformers.SharedInformerFactory
+
 	// ComponentConfig provides access to init options for a given controller
 	ComponentConfig yurtctrlmgrconfig.YurtControllerManagerConfiguration
 
@@ -306,6 +312,7 @@ func NewControllerInitializers() map[string]InitFunc {
 	controllers["nodelifecycle"] = startNodeLifecycleController
 	controllers["yurtcsrapprover"] = startYurtCSRApproverController
 	controllers["daemonpodupdater"] = startDaemonPodUpdaterController
+	controllers["servicetopologycontroller"] = startServiceTopologyController
 	return controllers
 }
 
@@ -314,21 +321,24 @@ func NewControllerInitializers() map[string]InitFunc {
 // the shared-informers client and token controller.
 func CreateControllerContext(s *config.CompletedConfig, rootClientBuilder, clientBuilder clientbuilder.ControllerClientBuilder, stop <-chan struct{}) (ControllerContext, error) {
 	versionedClient := rootClientBuilder.ClientOrDie("shared-informers")
+	kubeConfig := rootClientBuilder.ConfigOrDie("yurt-informers")
+	yurtClient := yurtclientset.NewForConfigOrDie(kubeConfig)
 	sharedInformers := informers.NewSharedInformerFactory(versionedClient, ResyncPeriod(s)())
-
+	yurtInformers := yurtinformers.NewSharedInformerFactory(yurtClient, ResyncPeriod(s)())
 	// If apiserver is not running we should wait for some time and fail only then. This is particularly
 	// important when we start apiserver and controller manager at the same time.
 	if err := genericcontrollermanager.WaitForAPIServer(versionedClient, 10*time.Second); err != nil {
 		return ControllerContext{}, fmt.Errorf("failed to wait for apiserver being healthy: %w", err)
 	}
 
 	ctx := ControllerContext{
-		ClientBuilder:    clientBuilder,
-		InformerFactory:  sharedInformers,
-		ComponentConfig:  s.ComponentConfig,
-		Stop:             stop,
-		InformersStarted: make(chan struct{}),
-		ResyncPeriod:     ResyncPeriod(s),
+		ClientBuilder:       clientBuilder,
+		InformerFactory:     sharedInformers,
+		YurtInformerFactory: yurtInformers,
+		ComponentConfig:     s.ComponentConfig,
+		Stop:                stop,
+		InformersStarted:    make(chan struct{}),
+		ResyncPeriod:        ResyncPeriod(s),
 	}
 	return ctx, nil
 }

cmd/yurt-controller-manager/app/core.go

@@ -28,6 +28,7 @@ import (
 	"github.com/openyurtio/openyurt/pkg/controller/certificates"
 	daemonpodupdater "github.com/openyurtio/openyurt/pkg/controller/daemonpodupdater"
 	lifecyclecontroller "github.com/openyurtio/openyurt/pkg/controller/nodelifecycle"
+	"github.com/openyurtio/openyurt/pkg/controller/servicetopology"
 )
 
 func startNodeLifecycleController(ctx ControllerContext) (http.Handler, bool, error) {
@@ -76,6 +77,20 @@ func startDaemonPodUpdaterController(ctx ControllerContext) (http.Handler, bool,
 	)
 
 	go daemonPodUpdaterCtrl.Run(2, ctx.Stop)
+	return nil, true, nil
+}
 
+func startServiceTopologyController(ctx ControllerContext) (http.Handler, bool, error) {
+	clientSet := ctx.ClientBuilder.ClientOrDie("yurt-servicetopology-controller")
+
+	svcTopologyController, err := servicetopology.NewServiceTopologyController(
+		clientSet,
+		ctx.InformerFactory,
+		ctx.YurtInformerFactory,
+	)
+	if err != nil {
+		return nil, false, err
+	}
+	go svcTopologyController.Run(ctx.Stop)
 	return nil, true, nil
 }

pkg/controller/servicetopology/adapter/adapter.go

@@ -0,0 +1,90 @@
+/*
+Copyright 2022 The OpenYurt Authors.
+Copyright 2017 The Kubernetes 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 adapter
+
+import (
+	"fmt"
+	"time"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/client-go/tools/cache"
+
+	"github.com/openyurtio/openyurt/pkg/yurthub/filter/servicetopology"
+)
+
+type Adapter interface {
+	GetEnqueueKeysBySvc(svc *corev1.Service) []string
+	GetEnqueueKeysByNodePool(svcTopologyTypes map[string]string, allNpNodes sets.String) []string
+	UpdateTriggerAnnotations(namespace, name string) error
+}
+
+func getSvcSelector(key, value string) labels.Selector {
+	return labels.SelectorFromSet(
+		map[string]string{
+			key: value,
+		},
+	)
+}
+
+func appendKeys(keys []string, obj interface{}) []string {
+	key, err := cache.MetaNamespaceKeyFunc(obj)
+	if err != nil {
+		runtime.HandleError(err)
+		return nil
+	}
+	keys = append(keys, key)
+	return keys
+}
+
+func isNodePoolTypeSvc(namespace, name string, svcTopologyTypes map[string]string) bool {
+	svc := &corev1.Service{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: namespace,
+			Name:      name,
+		},
+	}
+	key, err := cache.MetaNamespaceKeyFunc(svc)
+	if err != nil {
+		runtime.HandleError(err)
+		return false
+	}
+
+	return isNodePoolTypeTopology(svcTopologyTypes[key])
+}
+
+// TODO: if service topology need to support multi types,
+// like openyurt.io/topologyKeys: "kubernetes.io/hostname, openyurt.io/nodepool, *".
+// For simplicity，as long as the value of service topology annotation contains nodepool type,
+// then this topology is recognized as nodepool type
+func isNodePoolTypeTopology(topologyType string) bool {
+	if topologyType == servicetopology.AnnotationServiceTopologyValueNodePool {
+		return true
+	}
+	if topologyType == servicetopology.AnnotationServiceTopologyValueZone {
+		return true
+	}
+	return false
+}
+
+func getUpdateTriggerPatch() []byte {
+	patch := fmt.Sprintf(`{"metadata":{"annotations": {"openyurt.io/update-trigger": "%d"}}}`, time.Now().Unix())
+	return []byte(patch)
+}

pkg/controller/servicetopology/adapter/endpoints_adapter.go

@@ -0,0 +1,86 @@
+/*
+Copyright 2022 The OpenYurt Authors.
+Copyright 2017 The Kubernetes 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 adapter
+
+import (
+	"context"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/client-go/kubernetes"
+	corelisters "k8s.io/client-go/listers/core/v1"
+	"k8s.io/klog/v2"
+)
+
+func NewEndpointsAdapter(client kubernetes.Interface, epLister corelisters.EndpointsLister) Adapter {
+	return &endpoints{
+		client:   client,
+		epLister: epLister,
+	}
+}
+
+type endpoints struct {
+	client   kubernetes.Interface
+	epLister corelisters.EndpointsLister
+}
+
+func (s *endpoints) GetEnqueueKeysBySvc(svc *corev1.Service) []string {
+	var keys []string
+	return appendKeys(keys, svc)
+}
+
+func (s *endpoints) GetEnqueueKeysByNodePool(svcTopologyTypes map[string]string, allNpNodes sets.String) []string {
+	var keys []string
+	endpointsList, err := s.epLister.List(labels.Everything())
+	if err != nil {
+		klog.V(4).Infof("Error listing endpoints sets: %v", err)
+		return keys
+	}
+
+	for _, ep := range endpointsList {
+		if !isNodePoolTypeSvc(ep.Namespace, ep.Name, svcTopologyTypes) {
+			continue
+		}
+
+		if s.getNodesInEp(ep).Intersection(allNpNodes).Len() == 0 {
+			continue
+		}
+		keys = appendKeys(keys, ep)
+	}
+	return keys
+}
+
+func (s *endpoints) getNodesInEp(ep *corev1.Endpoints) sets.String {
+	nodes := sets.NewString()
+	for _, subset := range ep.Subsets {
+		for _, addr := range subset.Addresses {
+			if addr.NodeName != nil {
+				nodes.Insert(*addr.NodeName)
+			}
+		}
+	}
+	return nodes
+}
+
+func (s *endpoints) UpdateTriggerAnnotations(namespace, name string) error {
+	patch := getUpdateTriggerPatch()
+	_, err := s.client.CoreV1().Endpoints(namespace).Patch(context.Background(), name, types.StrategicMergePatchType, patch, metav1.PatchOptions{})
+	return err
+}