Feature: add sync readiness checker (#1499)

* Feature: add sync readiness checker --------- Co-authored-by: sunhaidong <[email protected]>
apache · Feb 20, 2025 · 47b2e16 · 47b2e16
1 parent ff84385
commit 47b2e16
Show file tree

Hide file tree

Showing 5 changed files with 139 additions and 3 deletions.
diff --git a/server/health/health.go b/server/health/health.go
@@ -19,12 +19,25 @@ package health
 
 import (
 	"errors"
+	"fmt"
+	"time"
 
+	"github.com/apache/servicecomb-service-center/pkg/log"
 	"github.com/apache/servicecomb-service-center/server/alarm"
+	"github.com/apache/servicecomb-service-center/syncer/config"
+	"github.com/apache/servicecomb-service-center/syncer/rpc"
 )
 
-var healthChecker Checker = &NullChecker{}
-var readinessChecker Checker = &DefaultHealthChecker{}
+var (
+	healthChecker        Checker = &NullChecker{}
+	readinessChecker     Checker = &DefaultHealthChecker{}
+	syncReadinessChecker Checker = &SyncReadinessChecker{}
+)
+
+var (
+	scNotReadyError     = errors.New("sc api server is not ready")
+	syncerNotReadyError = errors.New("the syncer module is not ready")
+)
 
 type Checker interface {
 	Healthy() error
@@ -40,7 +53,41 @@ func (n NullChecker) Healthy() error {
 type DefaultHealthChecker struct {
 }
 
+type SyncReadinessChecker struct {
+	startupTime time.Time
+}
+
+func SetStartupTime(startupTime time.Time) {
+	syncReadinessChecker.(*SyncReadinessChecker).startupTime = startupTime
+}
+
+func (src *SyncReadinessChecker) Healthy() error {
+	err := defaultHealth()
+	if err != nil {
+		return err
+	}
+	if src.startupTime.IsZero() {
+		return scNotReadyError
+	}
+	passTime := src.startupTime.Add(30 * time.Second)
+	if !rpc.IsNotReceiveSyncRequest() && rpc.GetFirstReceiveTime().Sub(src.startupTime) < 30*time.Second {
+		passTime = passTime.Add(rpc.GetFirstReceiveTime().Sub(src.startupTime))
+	} else {
+		log.Warn(fmt.Sprintf("first sync request is not received or received 30 seconds after startup,%s,%s", rpc.GetFirstReceiveTime(), src.startupTime))
+		passTime = passTime.Add(30 * time.Second)
+	}
+	nowTime := time.Now()
+	if nowTime.After(passTime) {
+		return nil
+	}
+	return syncerNotReadyError
+}
+
 func (hc *DefaultHealthChecker) Healthy() error {
+	return defaultHealth()
+}
+
+func defaultHealth() error {
 	for _, a := range alarm.ListAll() {
 		if a.Status == alarm.Cleared {
 			continue
@@ -65,5 +112,8 @@ func SetGlobalReadinessChecker(hc Checker) {
 }
 
 func GlobalReadinessChecker() Checker {
+	if config.GetConfig().Sync.EnableOnStart {
+		return syncReadinessChecker
+	}
 	return readinessChecker
 }
diff --git a/server/health/health_test.go b/server/health/health_test.go
@@ -21,8 +21,11 @@ import (
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/assert"
+
 	"github.com/apache/servicecomb-service-center/server/alarm"
 	"github.com/apache/servicecomb-service-center/server/event"
+	"github.com/apache/servicecomb-service-center/syncer/rpc"
 )
 
 func TestDefaultHealthChecker_Healthy(t *testing.T) {
@@ -55,3 +58,40 @@ func TestDefaultHealthChecker_Healthy(t *testing.T) {
 		t.Fatal("TestDefaultHealthChecker_Healthy failed")
 	}
 }
+
+func TestHealthy(t *testing.T) {
+	now := time.Now()
+	t.Run("sync_not_start", func(t *testing.T) {
+		err := syncReadinessChecker.Healthy()
+		assert.ErrorIs(t, err, scNotReadyError)
+	})
+
+	t.Run("no_sync", func(t *testing.T) {
+		// 未接受到同步请求，并且nowTime.Before(passTime)
+		src := &SyncReadinessChecker{startupTime: now}
+		err := src.Healthy()
+		assert.ErrorIs(t, err, syncerNotReadyError)
+	})
+
+	t.Run("no_sync_but_exceeds_60s", func(t *testing.T) {
+		// 未接受到同步请求，但是超出最大等待时间
+		src := &SyncReadinessChecker{startupTime: now.Add(-60 * time.Second)}
+		err := src.Healthy()
+		assert.Nil(t, err)
+	})
+
+	t.Run("sync_and_before", func(t *testing.T) {
+		// 30s内接受到第一次sync请求，并且nowTime.Before(passTime)
+		src := &SyncReadinessChecker{startupTime: now}
+		err := src.Healthy()
+		rpc.RecordFirstReceivedRequestTime()
+		assert.ErrorIs(t, err, syncerNotReadyError)
+	})
+
+	t.Run("31s_sync_and_before", func(t *testing.T) {
+		// 30s后接受到第一次sync请求，并且nowTime.Before(passTime)
+		src := &SyncReadinessChecker{startupTime: now.Add(-31 * time.Second)}
+		err := src.Healthy()
+		assert.ErrorIs(t, err, syncerNotReadyError)
+	})
+}
diff --git a/server/server.go b/server/server.go
@@ -20,8 +20,11 @@ package server
 import (
 	"context"
 	"crypto/tls"
-	"github.com/apache/servicecomb-service-center/pkg/protect"
 	"os"
+	"time"
+
+	"github.com/apache/servicecomb-service-center/pkg/protect"
+	"github.com/apache/servicecomb-service-center/server/health"
 
 	"github.com/gofiber/fiber/v2"
 
@@ -47,6 +50,11 @@ import (
 	"github.com/apache/servicecomb-service-center/server/service/rbac"
 )
 
+const (
+	apiServerStartCheckInterval = 1 * time.Second
+	apiServerStartCheckTimes    = 120 // 共检查2分钟
+)
+
 var sc ServiceCenterServer
 
 func Run() {
@@ -76,9 +84,23 @@ func (s *ServiceCenterServer) Run() {
 
 	signal.RegisterListener()
 
+	go initScStartupTime()
+
 	s.waitForQuit()
 }
 
+func initScStartupTime() {
+	i := 1
+	for ; i <= apiServerStartCheckTimes; i++ {
+		time.Sleep(apiServerStartCheckInterval)
+		// 等待sc api server初始化完成
+		if !GetAPIServer().IsClose() {
+			health.SetStartupTime(time.Now())
+			break
+		}
+	}
+}
+
 func (s *ServiceCenterServer) startChassis() {
 	go func() {
 		mask := make([]string, 0)

diff --git a/syncer/rpc/server.go b/syncer/rpc/server.go
@@ -40,6 +40,8 @@ const (
 	RbacAllowedRoleName    = "sync-admin"
 )
 
+var firstReceiveTime time.Time
+
 func NewServer() *Server {
 	return &Server{
 		replicator: replicator.Manager(),
@@ -53,6 +55,7 @@ type Server struct {
 }
 
 func (s *Server) Sync(ctx context.Context, events *v1sync.EventList) (*v1sync.Results, error) {
+	RecordFirstReceivedRequestTime()
 	err := auth(ctx)
 	if err != nil {
 		log.Error("auth failed", err)
@@ -66,6 +69,21 @@ func (s *Server) Sync(ctx context.Context, events *v1sync.EventList) (*v1sync.Re
 	return s.toResults(res), nil
 }
 
+func RecordFirstReceivedRequestTime() {
+	if IsNotReceiveSyncRequest() {
+		firstReceiveTime = time.Now()
+		log.Info(fmt.Sprintf("receive first received request time: %s", firstReceiveTime))
+	}
+}
+
+func IsNotReceiveSyncRequest() bool {
+	return firstReceiveTime.IsZero()
+}
+
+func GetFirstReceiveTime() time.Time {
+	return firstReceiveTime
+}
+
 func generateFailedResults(events *v1sync.EventList, err error) (*v1sync.Results, error) {
 	if events == nil || len(events.Events) == 0 {
 		return &v1sync.Results{Results: map[string]*v1sync.Result{}}, nil

diff --git a/syncer/service/sync/sync.go b/syncer/service/sync/sync.go
@@ -19,13 +19,19 @@ package sync
 
 import (
 	"fmt"
+	"time"
 
 	"github.com/apache/servicecomb-service-center/pkg/log"
 	"github.com/apache/servicecomb-service-center/syncer/service/event"
 	"github.com/apache/servicecomb-service-center/syncer/service/replicator"
 	"github.com/apache/servicecomb-service-center/syncer/service/task"
 )
 
+const (
+	apiServerStartCheckInterval = 1 * time.Second
+	apiServerStartCheckTimes    = 120 // 共检查2分钟
+)
+
 func Init() {
 	err := replicator.Work()
 	if err != nil {