Merge pull request #123 from fleetboard-io/feat/fit-with-mcs-serviceimport

lmxia · web-flow · commit a9011f9bc108 · 2025-03-15T12:36:14.000+08:00
Feat: Remove local syncer-operator namespace
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
@@ -30,7 +30,7 @@ jobs:
           fetch-depth: 1
           path: ${{ env.GOPATH }}/src/github.com/${{ github.repository }}
       - name: Cache go modules and build cache
-        uses: actions/cache@v4.0.1
+        uses: actions/cache@v4
         with:
           # In order:
           # * Module download cache
diff --git a/cmd/proxy/app/proxy.go b/cmd/proxy/app/proxy.go
@@ -632,8 +632,13 @@ func (s *ProxyServer) Run() error {
 		return err
 	}
 
+	hasFleetServiceName, err := labels.NewRequirement(known.LabelServiceName, selection.Exists, nil)
+	if err != nil {
+		return err
+	}
+
 	labelSelector := labels.NewSelector()
-	labelSelector = labelSelector.Add(*noProxyName, *noHeadlessEndpoints)
+	labelSelector = labelSelector.Add(*noProxyName, *noHeadlessEndpoints, *hasFleetServiceName)
 
 	// Make informers that filter out objects that want a non-default service proxy.
 	informerFactory := informers.NewSharedInformerFactoryWithOptions(s.Client, s.Config.ConfigSyncPeriod.Duration,
diff --git a/examples/client-import.yaml b/examples/client-import.yaml
@@ -36,10 +36,6 @@ apiVersion: multicluster.x-k8s.io/v1alpha1
 kind: ServiceImport
 metadata:
   name: nginx-svc
-  namespace: syncer-operator
-  labels:
-    services.fleetboard.io/multi-cluster-service-LocalNamespace: default
-    services.fleetboard.io/multi-cluster-service-name: nginx-svc
 spec:
   type: "ClusterSetIP"
   ports:
diff --git a/pkg/cnf/CNFManager.go b/pkg/cnf/CNFManager.go
@@ -158,7 +158,6 @@ func NewCNFManager(opts *tunnel.Options) (*Manager, error) {
 	serviceSyncer, errSyncer := syncer.New(&agentSpec, known.SyncerConfig{
 		LocalRestConfig: localConfig,
 		LocalClient:     dynamicLocalClient,
-		LocalNamespace:  agentSpec.LocalNamespace,
 		LocalClusterID:  agentSpec.ClusterID,
 	}, hubConfig)
 	if errSyncer != nil {
diff --git a/pkg/controller/mcs/serviceimport_controller.go b/pkg/controller/mcs/serviceimport_controller.go
@@ -71,8 +71,7 @@ func NewServiceImportController(kubeclient kubernetes.Interface,
 			default:
 				return sic.Handle(key)
 			}
-		}).
-		WithEnqueueFilterFunc(preFilter)
+		})
 	_, err := sic.localSIInformer.Informer().AddEventHandler(yachtcontroller.DefaultResourceEventHandlerFuncs())
 	if err != nil {
 		klog.Errorf("failed to add event handler for serviceimport: %v", err)
@@ -131,40 +130,19 @@ func (s *ServiceImportController) AddInitialInfoToServiceImport(si *v1alpha1.Ser
 func (s *ServiceImportController) getServiceImportFromEndpointSlice(obj interface{}) (*v1alpha1.ServiceImport, error) {
 	slice := obj.(*discoveryv1.EndpointSlice)
 	rawServiceName, serviceExist := slice.Labels[known.LabelServiceName]
-	rawServiceNamespace, serviceNamespaceExsit := slice.Labels[known.LabelServiceNameSpace]
+	_, serviceNamespaceExsit := slice.Labels[known.LabelServiceNameSpace]
 	if serviceExist && serviceNamespaceExsit {
-		if siList, err := s.localSILister.ServiceImports(s.operatorNamespace).List(
-			labels.SelectorFromSet(labels.Set{
-				known.LabelServiceName:      rawServiceName,
-				known.LabelServiceNameSpace: rawServiceNamespace,
-			})); err == nil && len(siList) > 0 {
-			return siList[0], nil
+		if si, err := s.localSILister.ServiceImports(slice.Namespace).Get(rawServiceName); err == nil {
+			return si, nil
 		} else {
-			klog.Errorf("failed to list ServiceImport for ServiceImport: %v", err)
+			klog.Errorf("failed to list ServiceImport for EndpointSlice %s/%s: %v",
+				slice.Namespace, slice.Name, err)
 		}
 	}
 	klog.Infof("can't resolve service import from this slice %s/%s", slice.Namespace, slice.Name)
 	return nil, fmt.Errorf("can't resolve service import from this slice %s/%s", slice.Namespace, slice.Name)
 }
 
-// preFilter filter ServiceImport if has no label known.LabelServiceName and known.LabelServiceNameSpace
-func preFilter(oldObj, newObj interface{}) (bool, error) {
-	var si *v1alpha1.ServiceImport
-	if newObj == nil {
-		// Delete
-		si = oldObj.(*v1alpha1.ServiceImport)
-	} else {
-		// Add or Update
-		si = newObj.(*v1alpha1.ServiceImport)
-	}
-	_, serviceExist := si.Labels[known.LabelServiceName]
-	_, serviceNamespaceExist := si.Labels[known.LabelServiceNameSpace]
-	if !serviceExist || !serviceNamespaceExist {
-		return false, nil
-	}
-	return true, nil
-}
-
 func (s *ServiceImportController) Handle(obj interface{}) (requeueAfter *time.Duration, err error) {
 	ctx := context.Background()
 	key := obj.(string)
@@ -202,19 +180,20 @@ func (s *ServiceImportController) Handle(obj interface{}) (requeueAfter *time.Du
 		si, err = s.mcsClientset.MulticlusterV1alpha1().ServiceImports(namespace).Update(context.TODO(),
 			si, metav1.UpdateOptions{})
 		if err != nil {
+			klog.Errorf("failed to update serviceimport %s/%s, for %v", si.Namespace, si.Name, err)
 			d := time.Second
 			return &d, err
 		}
 	}
 
 	// apply endpoint slices.
 	srcLabelMap := labels.Set{
-		known.LabelServiceName:      si.Labels[known.LabelServiceName],
-		known.LabelServiceNameSpace: si.Labels[known.LabelServiceNameSpace],
+		known.LabelServiceName:      si.Name,
+		known.LabelServiceNameSpace: si.Namespace,
 	}
 	dstLabelMap := labels.Set{
-		known.LabelServiceName:      si.Labels[known.LabelServiceName],
-		known.LabelServiceNameSpace: si.Labels[known.LabelServiceNameSpace],
+		known.LabelServiceName:      si.Name,
+		known.LabelServiceNameSpace: si.Namespace,
 	}
 
 	endpointSliceList, err := utils.RemoveNonexistentEndpointslice(s.sourceEndpointSlicesLister, "",
@@ -266,8 +245,8 @@ func (s *ServiceImportController) Run(ctx context.Context, delicatedNamespace st
 
 // recycleServiceImport recycle derived service and derived endpoint slices.
 func (s *ServiceImportController) recycleServiceImport(ctx context.Context, si *v1alpha1.ServiceImport) error {
-	rawServiceName := si.Labels[known.LabelServiceName]
-	rawServiceNamespace := si.Labels[known.LabelServiceNameSpace]
+	rawServiceName := si.Name
+	rawServiceNamespace := si.Namespace
 	// 1. recycle endpoint slices.
 	if err := s.localk8sClient.DiscoveryV1().EndpointSlices(si.Namespace).
 		DeleteCollection(ctx, metav1.DeleteOptions{}, metav1.ListOptions{
diff --git a/pkg/controller/syncer/agent.go b/pkg/controller/syncer/agent.go
@@ -31,8 +31,7 @@ type AgentConfig struct {
 }
 
 type Syncer struct {
-	ClusterID      string
-	LocalNamespace string
+	ClusterID string
 
 	HubKubeConfig           *rest.Config
 	SyncerConf              known.SyncerConfig
@@ -78,7 +77,6 @@ func New(spec *tunnel.Specification, syncerConf known.SyncerConfig, hubKubeConfi
 		return nil, err
 	}
 
-	syncerConf.LocalNamespace = spec.LocalNamespace
 	syncerConf.LocalClusterID = spec.ClusterID
 	syncerConf.RemoteNamespace = spec.ShareNamespace
 
@@ -88,7 +86,6 @@ func New(spec *tunnel.Specification, syncerConf known.SyncerConfig, hubKubeConfi
 		HubKubeConfig:           hubKubeConfig,
 		ServiceExportController: serviceExportController,
 		ServiceImportController: serviceImportController,
-		LocalNamespace:          syncerConf.LocalNamespace,
 		KubeInformerFactory:     kubeInformerFactory,
 		KubeClientSet:           localKubeClientSet,
 		McsInformerFactory:      mcsInformerFactory,
@@ -108,8 +105,7 @@ func (s *Syncer) Start(ctx context.Context) (err error) {
 
 	klog.Info("Starting Syncer and init virtual service CIDR...")
 	var cidr string
-	if cidr, err = s.ServiceImportController.IPAM.InitNewCIDR(s.LocalMcsClientSet,
-		s.LocalNamespace, s.KubeClientSet); err != nil {
+	if cidr, err = s.ServiceImportController.IPAM.InitNewCIDR(s.LocalMcsClientSet, s.KubeClientSet); err != nil {
 		klog.Errorf("we allocate for virtual service failed for %v", err)
 		return err
 	} else {
diff --git a/pkg/known/constants.go b/pkg/known/constants.go
@@ -7,7 +7,6 @@ import (
 const (
 	Fleetboard                = "fleetboard"
 	FleetboardSystemNamespace = "fleetboard-system"
-	SyncNamespace             = "syncer-operator"
 	HubClusterName            = "hub"
 	HubSecretName             = Fleetboard
 )
diff --git a/pkg/known/types.go b/pkg/known/types.go
@@ -12,7 +12,6 @@ type SyncerConfig struct {
 	// LocalClient the client used to access local resources to sync. This is optional and is provided for unit testing
 	// in lieu of the LocalRestConfig. If not specified, one is created from the LocalRestConfig.
 	LocalClient     dynamic.Interface
-	LocalNamespace  string
 	LocalClusterID  string
 	RemoteNamespace string
 }
diff --git a/pkg/plugin/crossdns.go b/pkg/plugin/crossdns.go
@@ -75,11 +75,7 @@ func (c *CrossDNS) getDNSRecord(ctx context.Context, _ string, state *request.Re
 		klog.Fatal("unable to sync caches for endpointslices or service import")
 	}
 
-	si, errGetSI := c.SILister.ServiceImports("syncer-operator").List(labels.SelectorFromSet(
-		labels.Set{
-			known.LabelServiceNameSpace: pReq.namespace,
-			known.LabelServiceName:      pReq.service,
-		}))
+	si, errGetSI := c.SILister.ServiceImports(pReq.namespace).Get(pReq.service)
 	if errGetSI != nil {
 		klog.Errorf("Failed to get service import %v", errGetSI)
 		return dns.RcodeServerFailure, errors.New("failed to write response")
@@ -88,24 +84,24 @@ func (c *CrossDNS) getDNSRecord(ctx context.Context, _ string, state *request.Re
 	var dnsRecords []DNSRecord
 	var err error
 	var srcEndpointSliceList []*v1.EndpointSlice
-	if si[0].Spec.Type == v1alpha1.ClusterSetIP {
-		if len(si[0].Spec.IPs) != 0 {
+	if si.Spec.Type == v1alpha1.ClusterSetIP {
+		if len(si.Spec.IPs) != 0 {
 			record := DNSRecord{
-				IP: si[0].Spec.IPs[0],
+				IP: si.Spec.IPs[0],
 			}
 			dnsRecords = append(dnsRecords, record)
 		}
 	} else {
 		if pReq.cluster != "" {
-			srcEndpointSliceList, err = c.endpointSlicesLister.EndpointSlices("syncer-operator").List(
+			srcEndpointSliceList, err = c.endpointSlicesLister.EndpointSlices(pReq.namespace).List(
 				labels.SelectorFromSet(
 					labels.Set{
 						known.LabelServiceNameSpace: pReq.namespace,
 						known.LabelServiceName:      pReq.service,
 						known.LabelClusterID:        pReq.cluster,
 					}))
 		} else {
-			srcEndpointSliceList, err = c.endpointSlicesLister.EndpointSlices("syncer-operator").List(
+			srcEndpointSliceList, err = c.endpointSlicesLister.EndpointSlices(pReq.namespace).List(
 				labels.SelectorFromSet(
 					labels.Set{
 						known.LabelServiceNameSpace: pReq.namespace,
diff --git a/pkg/proxy/config/config.go b/pkg/proxy/config/config.go
@@ -27,8 +27,6 @@ import (
 	"k8s.io/klog/v2"
 	"sigs.k8s.io/mcs-api/pkg/apis/v1alpha1"
 	mcsinformer "sigs.k8s.io/mcs-api/pkg/client/informers/externalversions/apis/v1alpha1"
-
-	"github.com/fleetboard-io/fleetboard/pkg/known"
 )
 
 // ServiceImportHandler is an abstract interface of objects which receive
@@ -119,11 +117,6 @@ func (c *EndpointSliceConfig) handleAddEndpointSlice(obj interface{}) {
 		return
 	}
 	for _, h := range c.eventHandlers {
-		if endpointSlice.Namespace != known.SyncNamespace {
-			klog.Infof("EndpointSliceConfig.handleAddEndpointSlice will not deal: %v/%v",
-				endpointSlice.Namespace, endpointSlice.Name)
-			continue
-		}
 		klog.V(4).InfoS("Calling handler.OnEndpointSliceAdd", "endpointslice",
 			klog.KObj(endpointSlice))
 		h.OnEndpointSliceAdd(endpointSlice)
@@ -142,11 +135,6 @@ func (c *EndpointSliceConfig) handleUpdateEndpointSlice(oldObj, newObj interface
 		return
 	}
 	for _, h := range c.eventHandlers {
-		if newEndpointSlice.Namespace != known.SyncNamespace {
-			klog.Infof("EndpointSliceConfig.handleUpdateEndpointSlice will not deal: %v/%v",
-				newEndpointSlice.Namespace, newEndpointSlice.Name)
-			continue
-		}
 		klog.V(4).InfoS("Calling handler.OnEndpointSliceUpdate", "endpointslice",
 			klog.KObj(newEndpointSlice))
 		h.OnEndpointSliceUpdate(oldEndpointSlice, newEndpointSlice)
@@ -167,12 +155,6 @@ func (c *EndpointSliceConfig) handleDeleteEndpointSlice(obj interface{}) {
 		}
 	}
 	for _, h := range c.eventHandlers {
-		if endpointSlice.Namespace != known.SyncNamespace {
-			klog.Infof("EndpointSliceConfig.handleDeleteEndpointSlice will not deal: %v/%v",
-				endpointSlice.Namespace, endpointSlice.Name)
-			continue
-		}
-
 		klog.V(4).InfoS("Calling handler.OnEndpointsDelete", "endpointslice", klog.KObj(endpointSlice))
 		h.OnEndpointSliceDelete(endpointSlice)
 	}
diff --git a/pkg/proxy/endpointslicecache.go b/pkg/proxy/endpointslicecache.go
@@ -123,12 +123,12 @@ func newEndpointSliceTracker() *endpointSliceTracker {
 func newEndpointSliceInfo(endpointSlice *discovery.EndpointSlice, remove bool) *endpointSliceInfo {
 	nameLabel, ok := endpointSlice.Labels[known.LabelServiceName]
 	if !ok {
-		klog.ErrorS(nil, "ServiceImport has no name label", "serviceImport", klog.KObj(endpointSlice))
+		klog.ErrorS(nil, "EndpointSlice has no name label", "EndpointSlice", klog.KObj(endpointSlice))
 		return nil
 	}
 	namespaceLabel, ok := endpointSlice.Labels[known.LabelServiceNameSpace]
 	if !ok {
-		klog.ErrorS(nil, "ServiceImport has no namespace label", "serviceImport",
+		klog.ErrorS(nil, "EndpointSlice has no namespace label", "EndpointSlice",
 			klog.KObj(endpointSlice))
 		return nil
 	}
diff --git a/pkg/proxy/ipvs/proxier.go b/pkg/proxy/ipvs/proxier.go
@@ -50,7 +50,6 @@ import (
 	netutils "k8s.io/utils/net"
 	"sigs.k8s.io/mcs-api/pkg/apis/v1alpha1"
 
-	"github.com/fleetboard-io/fleetboard/pkg/known"
 	"github.com/fleetboard-io/fleetboard/pkg/proxy"
 )
 
@@ -471,20 +470,9 @@ type servicePortInfo struct {
 func newServiceInfo(port *v1.ServicePort, serviceImport *v1alpha1.ServiceImport,
 	bsvcPortInfo *proxy.BaseServicePortInfo) proxy.ServicePort {
 	svcPort := &servicePortInfo{BaseServicePortInfo: bsvcPortInfo}
-	// get service name and namespace
-	nameLabel, ok := serviceImport.Labels[known.LabelServiceName]
-	if !ok {
-		klog.ErrorS(nil, "ServiceImport has no name label", "serviceImport", klog.KObj(serviceImport))
-		return nil
-	}
-	namespaceLabel, ok := serviceImport.Labels[known.LabelServiceNameSpace]
-	if !ok {
-		klog.ErrorS(nil, "ServiceImport has no namespace label", "serviceImport", klog.KObj(serviceImport))
-		return nil
-	}
 
 	// Store the following for performance reasons.
-	svcName := types.NamespacedName{Namespace: namespaceLabel, Name: nameLabel}
+	svcName := types.NamespacedName{Namespace: serviceImport.Namespace, Name: serviceImport.Name}
 	svcPortName := proxy.ServicePortName{NamespacedName: svcName, Port: port.Name}
 	svcPort.nameString = svcPortName.String()
 
diff --git a/pkg/proxy/service_import.go b/pkg/proxy/service_import.go
@@ -31,8 +31,6 @@ import (
 	"k8s.io/kubernetes/pkg/proxy/metrics"
 	netutils "k8s.io/utils/net"
 	"sigs.k8s.io/mcs-api/pkg/apis/v1alpha1"
-
-	"github.com/fleetboard-io/fleetboard/pkg/known"
 )
 
 // BaseServicePortInfo contains base information that defines a service.
@@ -315,20 +313,8 @@ func (sct *ServiceImportChangeTracker) serviceImportToServiceMap(serviceImport *
 		return nil
 	}
 
-	// get service name and namespace
-	nameLabel, ok := serviceImport.Labels[known.LabelServiceName]
-	if !ok {
-		klog.ErrorS(nil, "ServiceImport has no name label", "serviceImport", klog.KObj(serviceImport))
-		return nil
-	}
-	namespaceLabel, ok := serviceImport.Labels[known.LabelServiceNameSpace]
-	if !ok {
-		klog.ErrorS(nil, "ServiceImport has no namespace label", "serviceImport", klog.KObj(serviceImport))
-		return nil
-	}
-
 	svcPortMap := make(ServicePortMap)
-	svcName := types.NamespacedName{Namespace: namespaceLabel, Name: nameLabel}
+	svcName := types.NamespacedName{Namespace: serviceImport.Namespace, Name: serviceImport.Name}
 	for i := range serviceImport.Spec.Ports {
 		servicePort := serviceImport.Spec.Ports[i]
 
diff --git a/pkg/tunnel/options.go b/pkg/tunnel/options.go
@@ -23,8 +23,6 @@ type Options struct {
 	HubSecretName string
 	// used to share endpoint slices in hub
 	ShareNamespace string
-	// used to store endpoint slices in local clusters
-	LocalNamespace string
 	// true means run as hub
 	AsHub bool
 	// true means run as cluster
@@ -64,10 +62,6 @@ func (o *Options) Validate() []error {
 		allErrors = append(allErrors, fmt.Errorf("--shared-namespace must be specified"))
 	}
 
-	if o.AsCluster && len(o.LocalNamespace) == 0 {
-		allErrors = append(allErrors, fmt.Errorf("--local-namespace must be specified when run as cluster"))
-	}
-
 	if o.AsCluster && len(o.HubSecretNamespace) == 0 {
 		allErrors = append(allErrors, fmt.Errorf("--hub-secret-namespace must be specified when run as cluser"))
 	}
@@ -114,8 +108,5 @@ func (o *Options) Flags() (fss cliflag.NamedFlagSets) {
 	fs.StringVar(&o.ShareNamespace, "shared-namespace", o.ShareNamespace,
 		"shared namespace in hub used to share endpoint slices across clusters")
 
-	fs.StringVar(&o.LocalNamespace, "local-namespace", o.LocalNamespace,
-		"local namespace in cluster used to share endpoint slices across clusters")
-
 	return fss
 }
diff --git a/utils/endpointslice.go b/utils/endpointslice.go

Original file line number	Diff line number	Diff line change
`@@ -7,7 +7,6 @@ import (`
`7`	`7`	`const (`
`8`	`8`	`Fleetboard = "fleetboard"`
`9`	`9`	`FleetboardSystemNamespace = "fleetboard-system"`
`10`		`- SyncNamespace = "syncer-operator"`
`11`	`10`	`HubClusterName = "hub"`
`12`	`11`	`HubSecretName = Fleetboard`
`13`	`12`	`)`
Original file line number	Diff line number	Diff line change
`@@ -12,7 +12,6 @@ type SyncerConfig struct {`
`12`	`12`	`// LocalClient the client used to access local resources to sync. This is optional and is provided for unit testing`
`13`	`13`	`// in lieu of the LocalRestConfig. If not specified, one is created from the LocalRestConfig.`
`14`	`14`	`LocalClient dynamic.Interface`
`15`		`- LocalNamespace string`
`16`	`15`	`LocalClusterID string`
`17`	`16`	`RemoteNamespace string`
`18`	`17`	`}`