Merge pull request #1002 from andrewshan/release-v1.15.0

fix: 解决eureka replicate因串行出现的超时性能问题（cherrypick 1.15）

Author: andrewshan

apiserver/eurekaserver/access_test.go

@@ -158,7 +158,7 @@ func TestCreateInstance(t *testing.T) {
 	startPort := 8900
 	host := "127.0.1.1"
 	total := 10
-	instances := batchBuildInstances(appId, "127.0.1.1", 8900, &LeaseInfo{
+	instances := batchBuildInstances(appId, host, startPort, &LeaseInfo{
 		RenewalIntervalInSecs: 30,
 		DurationInSecs:        120,
 	}, total)

apiserver/eurekaserver/replicate.go

@@ -21,6 +21,8 @@ import (
 	"context"
 	"net/http"
 	"strconv"
+	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/emicklei/go-restful/v3"
@@ -86,17 +88,31 @@ func (h *EurekaServer) BatchReplication(req *restful.Request, rsp *restful.Respo
 
 func (h *EurekaServer) doBatchReplicate(
 	replicateRequest *ReplicationList, token string) (*ReplicationListResponse, uint32) {
-	batchResponse := &ReplicationListResponse{ResponseList: []*ReplicationInstanceResponse{}}
+	batchResponse := &ReplicationListResponse{}
 	var resultCode = api.ExecuteSuccess
-	for _, instanceInfo := range replicateRequest.ReplicationList {
-		resp, code := h.dispatch(instanceInfo, token)
-		if code != api.ExecuteSuccess {
-			resultCode = code
-			log.Warnf("[EUREKA-SERVER] fail to process replicate instance request, code is %d, action %s, instance %s, app %s",
-				code, instanceInfo.Action, instanceInfo.Id, instanceInfo.AppName)
-		}
-		batchResponse.ResponseList = append(batchResponse.ResponseList, resp)
+	itemCount := len(replicateRequest.ReplicationList)
+	if itemCount == 0 {
+		return batchResponse, resultCode
+	}
+	batchResponse.ResponseList = make([]*ReplicationInstanceResponse, itemCount)
+	wg := &sync.WaitGroup{}
+	wg.Add(itemCount)
+	mutex := &sync.Mutex{}
+	for i, inst := range replicateRequest.ReplicationList {
+		go func(idx int, instanceInfo *ReplicationInstance) {
+			defer wg.Done()
+			resp, code := h.dispatch(instanceInfo, token)
+			if code != api.ExecuteSuccess {
+				atomic.CompareAndSwapUint32(&resultCode, api.ExecuteSuccess, code)
+				log.Warnf("[EUREKA-SERVER] fail to process replicate instance request, code is %d, action %s, instance %s, app %s",
+					code, instanceInfo.Action, instanceInfo.Id, instanceInfo.AppName)
+			}
+			mutex.Lock()
+			batchResponse.ResponseList[idx] = resp
+			mutex.Unlock()
+		}(i, inst)
 	}
+	wg.Wait()
 	return batchResponse, resultCode
 }
 
@@ -213,7 +229,7 @@ func (h *EurekaServer) handleInstanceEvent(ctx context.Context, i interface{}) e
 	appName := formatReadName(e.Service)
 	curTimeMilli := time.Now().UnixMilli()
 	switch e.EType {
-	case model.EventInstanceOnline:
+	case model.EventInstanceOnline, model.EventInstanceUpdate:
 		instanceInfo := eventToInstance(&e, appName, curTimeMilli)
 		h.replicateWorker.AddReplicateTask(&ReplicationInstance{
 			AppName:            appName,
@@ -242,22 +258,6 @@ func (h *EurekaServer) handleInstanceEvent(ctx context.Context, i interface{}) e
 			rInstance.OverriddenStatus = StatusOutOfService
 		}
 		h.replicateWorker.AddReplicateTask(rInstance)
-	case model.EventInstanceTurnHealth:
-		h.replicateWorker.AddReplicateTask(&ReplicationInstance{
-			AppName:            appName,
-			Id:                 e.Id,
-			LastDirtyTimestamp: curTimeMilli,
-			Status:             StatusUp,
-			Action:             actionStatusUpdate,
-		})
-	case model.EventInstanceTurnUnHealth:
-		h.replicateWorker.AddReplicateTask(&ReplicationInstance{
-			AppName:            appName,
-			Id:                 e.Id,
-			LastDirtyTimestamp: curTimeMilli,
-			Status:             StatusDown,
-			Action:             actionStatusUpdate,
-		})
 	case model.EventInstanceOpenIsolate:
 		h.replicateWorker.AddReplicateTask(&ReplicationInstance{
 			AppName:            appName,

apiserver/eurekaserver/replicate_test.go

@@ -0,0 +1,80 @@
+/**
+ * Tencent is pleased to support the open source community by making Polaris available.
+ *
+ * Copyright (C) 2019 THL A29 Limited, a Tencent company. All rights reserved.
+ *
+ * Licensed under the BSD 3-Clause License (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://opensource.org/licenses/BSD-3-Clause
+ *
+ * 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 eurekaserver
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+
+	api "github.com/polarismesh/polaris/common/api/v1"
+	testsuit "github.com/polarismesh/polaris/test/suit"
+)
+
+func TestDispatchHeartbeat(t *testing.T) {
+	discoverSuit := &testsuit.DiscoverTestSuit{}
+	if err := discoverSuit.Initialize(); err != nil {
+		t.Fatal(err)
+	}
+	defer discoverSuit.Destroy()
+
+	options := map[string]interface{}{optionRefreshInterval: 5, optionDeltaExpireInterval: 120}
+	eurekaSrv, err := createEurekaServerForTest(discoverSuit, options)
+	assert.Nil(t, err)
+	eurekaSrv.worker = NewApplicationsWorker(eurekaSrv.refreshInterval, eurekaSrv.deltaExpireInterval,
+		eurekaSrv.enableSelfPreservation, eurekaSrv.namingServer, eurekaSrv.healthCheckServer, eurekaSrv.namespace)
+
+	appId := "TESTAPP"
+	startPort := 8900
+	host := "127.0.1.1"
+	total := 30
+	instances := batchBuildInstances(appId, host, startPort, &LeaseInfo{
+		RenewalIntervalInSecs: 30,
+		DurationInSecs:        120,
+	}, total)
+
+	var replicateInstances = &ReplicationList{}
+
+	for i, instance := range instances {
+		log.Infof("replicate test: register %d", i)
+		replicateInstances.ReplicationList = append(replicateInstances.ReplicationList, &ReplicationInstance{
+			AppName:      appId,
+			Id:           instance.InstanceId,
+			InstanceInfo: instance,
+			Action:       actionRegister,
+		})
+	}
+	_, code := eurekaSrv.doBatchReplicate(replicateInstances, "")
+	assert.Equal(t, api.ExecuteSuccess, code)
+
+	time.Sleep(10 * time.Second)
+	for i := 0; i < 5; i++ {
+		log.Infof("replicate test: heartbeat %d", i)
+		replicateInstances = &ReplicationList{}
+		for _, instance := range instances {
+			replicateInstances.ReplicationList = append(replicateInstances.ReplicationList, &ReplicationInstance{
+				AppName: appId,
+				Id:      instance.InstanceId,
+				Action:  actionHeartbeat,
+			})
+		}
+		_, code := eurekaSrv.doBatchReplicate(replicateInstances, "")
+		assert.Equal(t, api.ExecuteSuccess, code)
+	}
+}

apiserver/eurekaserver/server.go

@@ -142,6 +142,8 @@ type EurekaServer struct {
 	enableSelfPreservation bool
 	replicateWorker        *ReplicateWorker
 	eventHandlerHandler    *EurekaInstanceEventHandler
+
+	replicatePeers []string
 }
 
 // GetPort 获取端口
@@ -192,30 +194,25 @@ func (h *EurekaServer) Initialize(ctx context.Context, option map[string]interfa
 				replicatePeers = append(replicatePeers, replicatePeerObj.(string))
 			}
 		}
-		if len(replicatePeers) > 0 {
-			h.replicateWorker = NewReplicateWorker(ctx, replicatePeers)
-			h.eventHandlerHandler = &EurekaInstanceEventHandler{
-				BaseInstanceEventHandler: service.NewBaseInstanceEventHandler(h.namingServer), svr: h}
-			if err := eventhub.Subscribe(
-				eventhub.InstanceEventTopic, "eureka-replication", h.eventHandlerHandler); nil != err {
-				return err
-			}
+		h.replicatePeers = replicatePeers
+		if len(h.replicatePeers) > 0 {
+			h.replicateWorker = NewReplicateWorker(ctx, h.replicatePeers)
 		}
 	}
 
 	var refreshInterval int
 	if value, ok := option[optionRefreshInterval]; ok {
 		refreshInterval = value.(int)
 	}
-	if refreshInterval == 0 {
+	if refreshInterval <= 0 {
 		refreshInterval = DefaultRefreshInterval
 	}
 
 	var deltaExpireInterval int
 	if value, ok := option[optionDeltaExpireInterval]; ok {
 		deltaExpireInterval = value.(int)
 	}
-	if deltaExpireInterval == 0 {
+	if deltaExpireInterval <= 0 {
 		deltaExpireInterval = DefaultDetailExpireInterval
 	}
 
@@ -283,6 +280,15 @@ func (h *EurekaServer) Run(errCh chan error) {
 		errCh <- err
 		return
 	}
+	if len(h.replicatePeers) > 0 {
+		h.eventHandlerHandler = &EurekaInstanceEventHandler{
+			BaseInstanceEventHandler: service.NewBaseInstanceEventHandler(h.namingServer), svr: h}
+		if err = eventhub.Subscribe(
+			eventhub.InstanceEventTopic, "eureka-replication", h.eventHandlerHandler); nil != err {
+			errCh <- err
+			return
+		}
+	}
 	h.worker = NewApplicationsWorker(h.refreshInterval, h.deltaExpireInterval, h.enableSelfPreservation,
 		h.namingServer, h.healthCheckServer, h.namespace)
 	h.statis = plugin.GetStatis()

common/model/instance_event.go

@@ -45,6 +45,8 @@ const (
 	EventInstanceOffline InstanceEventType = "InstanceOffline"
 	// EventInstanceSendHeartbeat Instance send heartbeat package to server
 	EventInstanceSendHeartbeat InstanceEventType = "InstanceSendHeartbeat"
+	// EventInstanceUpdate Instance metadata and info update event
+	EventInstanceUpdate InstanceEventType = "InstanceUpdate"
 )
 
 // CtxEventKeyMetadata 用于将metadata从Context中传入并取出

service/instance.go

@@ -429,11 +429,10 @@ func (s *Server) UpdateInstance(ctx context.Context, req *apiservice.Instance) *
 	platformID := utils.ParsePlatformID(ctx)
 	log.Info(fmt.Sprintf("old instance: %+v", instance), utils.ZapRequestID(requestID), utils.ZapPlatformID(platformID))
 
-	// 比对下更新前后的 isolate 状态
-	eventTypes := diffInstanceEvent(req, instance)
-
+	var eventTypes map[model.InstanceEventType]bool
+	var needUpdate bool
 	// 存储层操作
-	if needUpdate := s.updateInstanceAttribute(req, instance); !needUpdate {
+	if needUpdate, eventTypes = s.updateInstanceAttribute(req, instance); !needUpdate {
 		log.Info("update instance no data change, no need update",
 			utils.ZapRequestID(requestID), utils.ZapPlatformID(platformID), zap.String("instance", req.String()))
 		return api.NewInstanceResponse(apimodel.Code_NoNeedUpdate, req)
@@ -449,13 +448,13 @@ func (s *Server) UpdateInstance(ctx context.Context, req *apiservice.Instance) *
 	log.Info(msg, utils.ZapRequestID(requestID), utils.ZapPlatformID(platformID))
 	s.RecordHistory(ctx, instanceRecordEntry(ctx, req, service, instance, model.OUpdate))
 
-	for i := range eventTypes {
+	for eventType := range eventTypes {
 		event := &model.InstanceEvent{
 			Id:         instance.ID(),
 			Namespace:  service.Namespace,
 			Service:    service.Name,
 			Instance:   instance.Proto,
-			EType:      eventTypes[i],
+			EType:      eventType,
 			CreateTime: time.Time{},
 		}
 		event.InjectMetadata(ctx)
@@ -604,71 +603,86 @@ func (s *Server) getInstancesMainByService(ctx context.Context, req *apiservice.
 /**
  * @brief 修改服务属性
  */
-func (s *Server) updateInstanceAttribute(req *apiservice.Instance, instance *model.Instance) bool {
+func (s *Server) updateInstanceAttribute(
+	req *apiservice.Instance, instance *model.Instance) (bool, map[model.InstanceEventType]bool) {
 	// #lizard forgives
 	instance.MallocProto()
 	needUpdate := false
 	insProto := instance.Proto
+	var updateEvents = make(map[model.InstanceEventType]bool)
 	if ok := instanceMetaNeedUpdate(req.GetMetadata(), instance.Metadata()); ok {
 		insProto.Metadata = req.GetMetadata()
 		needUpdate = true
+		updateEvents[model.EventInstanceUpdate] = true
 	}
 
 	if ok := instanceLocationNeedUpdate(req.GetLocation(), instance.Proto.GetLocation()); ok {
 		insProto.Location = req.Location
 		needUpdate = true
+		updateEvents[model.EventInstanceUpdate] = true
 	}
 
-	// if !needUpdate {
-	// 	// 不需要更新metadata，则置空
-	// 	insProto.Metadata = nil
-	// }
-
 	if req.GetProtocol() != nil && req.GetProtocol().GetValue() != instance.Protocol() {
 		insProto.Protocol = req.GetProtocol()
 		needUpdate = true
+		updateEvents[model.EventInstanceUpdate] = true
 	}
 
 	if req.GetVersion() != nil && req.GetVersion().GetValue() != instance.Version() {
 		insProto.Version = req.GetVersion()
 		needUpdate = true
+		updateEvents[model.EventInstanceUpdate] = true
 	}
 
 	if req.GetPriority() != nil && req.GetPriority().GetValue() != instance.Priority() {
 		insProto.Priority = req.GetPriority()
 		needUpdate = true
+		updateEvents[model.EventInstanceUpdate] = true
 	}
 
 	if req.GetWeight() != nil && req.GetWeight().GetValue() != instance.Weight() {
 		insProto.Weight = req.GetWeight()
 		needUpdate = true
+		updateEvents[model.EventInstanceUpdate] = true
 	}
 
 	if req.GetHealthy() != nil && req.GetHealthy().GetValue() != instance.Healthy() {
 		insProto.Healthy = req.GetHealthy()
 		needUpdate = true
+		if req.Healthy.GetValue() {
+			updateEvents[model.EventInstanceTurnHealth] = true
+		} else {
+			updateEvents[model.EventInstanceTurnUnHealth] = true
+		}
 	}
 
 	if req.GetIsolate() != nil && req.GetIsolate().GetValue() != instance.Isolate() {
 		insProto.Isolate = req.GetIsolate()
 		needUpdate = true
+		if req.Isolate.GetValue() {
+			updateEvents[model.EventInstanceOpenIsolate] = true
+		} else {
+			updateEvents[model.EventInstanceCloseIsolate] = true
+		}
 	}
 
 	if req.GetLogicSet() != nil && req.GetLogicSet().GetValue() != instance.LogicSet() {
 		insProto.LogicSet = req.GetLogicSet()
 		needUpdate = true
+		updateEvents[model.EventInstanceUpdate] = true
 	}
 
 	if ok := updateHealthCheck(req, instance); ok {
 		needUpdate = true
+		updateEvents[model.EventInstanceUpdate] = true
 	}
 
 	// 每次更改，都要生成一个新的uuid
 	if needUpdate {
 		insProto.Revision = utils.NewStringValue(utils.NewUUID())
 	}
 
-	return needUpdate
+	return needUpdate, updateEvents
 }
 
 func instanceLocationNeedUpdate(req *apimodel.Location, old *apimodel.Location) bool {
@@ -1369,24 +1383,3 @@ func CheckDbInstanceFieldLen(req *apiservice.Instance) (*apiservice.Response, bo
 	}
 	return nil, false
 }
-
-func diffInstanceEvent(req *apiservice.Instance, save *model.Instance) []model.InstanceEventType {
-	eventTypes := make([]model.InstanceEventType, 0)
-	if req.Isolate != nil && save.Isolate() != req.Isolate.GetValue() {
-		if req.Isolate.GetValue() {
-			eventTypes = append(eventTypes, model.EventInstanceOpenIsolate)
-		} else {
-			eventTypes = append(eventTypes, model.EventInstanceCloseIsolate)
-		}
-	}
-
-	if req.Healthy != nil && save.Healthy() != req.Healthy.GetValue() {
-		if req.Healthy.GetValue() {
-			eventTypes = append(eventTypes, model.EventInstanceTurnHealth)
-		} else {
-			eventTypes = append(eventTypes, model.EventInstanceTurnUnHealth)
-		}
-	}
-
-	return eventTypes
-}