[mesh]: Add leader judgement for syncing

pkg/cluster/cluster.go

@@ -163,6 +163,15 @@ func New(opt *option.Options) (Cluster, error) {
 	return c, nil
 }
 
+func (c *cluster) IsLeader() bool {
+	server, err := c.getServer()
+	if err != nil {
+		return false
+	}
+
+	return server.Server.Leader() == server.Server.ID()
+}
+
 // requestContext returns context with request timeout,
 // please use it immediately in case of incorrect timeout.
 func (c *cluster) requestContext() context.Context {

pkg/cluster/cluster_interface.go

@@ -28,6 +28,8 @@ import (
 type (
 	// Cluster is the open cluster interface.
 	Cluster interface {
+		IsLeader() bool
+
 		Layout() *Layout
 
 		Get(key string) (*string, error)

pkg/object/meshcontroller/master/registrysyncer.go

@@ -82,6 +82,10 @@ func (rs *registrySyncer) run() {
 		case <-rs.done:
 			return
 		case event := <-rs.registryWatcher.Watch():
+			if !rs.needSync() {
+				continue
+			}
+
 			rs.handleEvent(event)
 		}
 	}
@@ -133,6 +137,10 @@ func (rs *registrySyncer) handleEvent(event *serviceregistry.RegistryEvent) {
 }
 
 func (rs *registrySyncer) serviceInstanceSpecsFunc(meshInstances map[string]*spec.ServiceInstanceSpec) bool {
+	if !rs.needSync() {
+		return true
+	}
+
 	oldInstances, err := rs.serviceRegistry.ListAllServiceInstances(rs.spec.ExternalServiceRegistry)
 	if err != nil {
 		logger.Errorf("list all service instances of %s: %v", rs.spec.ExternalServiceRegistry, err)
@@ -163,6 +171,11 @@ func (rs *registrySyncer) serviceInstanceSpecsFunc(meshInstances map[string]*spe
 	return true
 }
 
+func (rs *registrySyncer) needSync() bool {
+	// NOTE: Only need one member in the cluster to do sync.
+	return rs.superSpec.Super().Cluster().IsLeader()
+}
+
 func (rs *registrySyncer) meshRegistryName() string {
 	return rs.superSpec.Name()
 }