等待 apiserver 都启动完成，再自注册。 (#785)

* 等待 apiserver 都启动完成，再自注册。+ 格式化代码等待 apiserver 都启动完成，再自注册。 1.完善代码规范 2.将cond.add 放到 apiserver的listen后。完善代码规范 * import_format 格式化 * 添加Licensed * 改用waitgroup Co-authored-by: zhanglei25 <[email protected]>
polarismesh · Nov 9, 2022 · 9679e45 · 9679e45
1 parent db611ac
commit 9679e45
Show file tree

Hide file tree

Showing 12 changed files with 41 additions and 22 deletions.
diff --git a/apiserver/eurekaserver/eureka_suit_test.go b/apiserver/eurekaserver/eureka_suit_test.go
@@ -33,15 +33,6 @@ import (
 	"github.com/polarismesh/polaris/common/utils"
 	"github.com/polarismesh/polaris/namespace"
 	"github.com/polarismesh/polaris/plugin"
-	"github.com/polarismesh/polaris/service"
-	"github.com/polarismesh/polaris/service/batch"
-	"github.com/polarismesh/polaris/service/healthcheck"
-	"github.com/polarismesh/polaris/store"
-	storemock "github.com/polarismesh/polaris/store/mock"
-
-	"github.com/polarismesh/polaris/testdata"
-
-	// 注册相关默认插件
 	_ "github.com/polarismesh/polaris/plugin/cmdb/memory"
 	_ "github.com/polarismesh/polaris/plugin/discoverevent/local"
 	_ "github.com/polarismesh/polaris/plugin/discoverstat/discoverlocal"
@@ -52,6 +43,12 @@ import (
 	_ "github.com/polarismesh/polaris/plugin/ratelimit/lrurate"
 	_ "github.com/polarismesh/polaris/plugin/ratelimit/token"
 	_ "github.com/polarismesh/polaris/plugin/statis/local"
+	"github.com/polarismesh/polaris/service"
+	"github.com/polarismesh/polaris/service/batch"
+	"github.com/polarismesh/polaris/service/healthcheck"
+	"github.com/polarismesh/polaris/store"
+	storemock "github.com/polarismesh/polaris/store/mock"
+	"github.com/polarismesh/polaris/testdata"
 )
 
 type Bootstrap struct {

diff --git a/apiserver/eurekaserver/server.go b/apiserver/eurekaserver/server.go
@@ -27,10 +27,11 @@ import (
 	"strings"
 	"time"
 
-	"github.com/emicklei/go-restful/v3"
+	restful "github.com/emicklei/go-restful/v3"
 	"go.uber.org/zap"
 
 	"github.com/polarismesh/polaris/apiserver"
+	"github.com/polarismesh/polaris/bootstrap"
 	"github.com/polarismesh/polaris/common/connlimit"
 	"github.com/polarismesh/polaris/common/secure"
 	"github.com/polarismesh/polaris/common/utils"
@@ -253,6 +254,8 @@ func (h *EurekaServer) Run(errCh chan error) {
 		errCh <- err
 		return
 	}
+	bootstrap.ApiServerWaitGroup.Done()
+
 	ln = &tcpKeepAliveListener{ln.(*net.TCPListener)}
 	// 开启最大连接数限制
 	if h.connLimitConfig != nil && h.connLimitConfig.OpenConnLimit {

diff --git a/apiserver/grpcserver/config/server.go b/apiserver/grpcserver/config/server.go
@@ -26,6 +26,7 @@ import (
 
 	"github.com/polarismesh/polaris/apiserver"
 	"github.com/polarismesh/polaris/apiserver/grpcserver"
+	"github.com/polarismesh/polaris/bootstrap"
 	api "github.com/polarismesh/polaris/common/api/v1"
 	commonlog "github.com/polarismesh/polaris/common/log"
 	"github.com/polarismesh/polaris/config"
@@ -65,6 +66,7 @@ func (g *ConfigGRPCServer) Initialize(ctx context.Context, option map[string]int
 // Run 启动GRPC API服务器
 func (g *ConfigGRPCServer) Run(errCh chan error) {
 	g.BaseGrpcServer.Run(errCh, g.GetProtocol(), func(server *grpc.Server) error {
+		defer bootstrap.ApiServerWaitGroup.Done()
 		for name, apiConfig := range g.openAPI {
 			switch name {
 			case "client":

diff --git a/apiserver/grpcserver/discover/server.go b/apiserver/grpcserver/discover/server.go
@@ -28,6 +28,7 @@ import (
 	"github.com/polarismesh/polaris/apiserver/grpcserver"
 	v1 "github.com/polarismesh/polaris/apiserver/grpcserver/discover/v1"
 	v2 "github.com/polarismesh/polaris/apiserver/grpcserver/discover/v2"
+	"github.com/polarismesh/polaris/bootstrap"
 	apiv1 "github.com/polarismesh/polaris/common/api/v1"
 	apiv2 "github.com/polarismesh/polaris/common/api/v2"
 	commonlog "github.com/polarismesh/polaris/common/log"
@@ -106,6 +107,7 @@ func (g *GRPCServer) Initialize(ctx context.Context, option map[string]interface
 // Run 启动GRPC API服务器
 func (g *GRPCServer) Run(errCh chan error) {
 	g.BaseGrpcServer.Run(errCh, g.GetProtocol(), func(server *grpc.Server) error {
+		defer bootstrap.ApiServerWaitGroup.Done()
 		for name, config := range g.openAPI {
 			switch name {
 			case "client":

diff --git a/apiserver/httpserver/server.go b/apiserver/httpserver/server.go
@@ -26,7 +26,7 @@ import (
 	"strings"
 	"time"
 
-	"github.com/emicklei/go-restful/v3"
+	restful "github.com/emicklei/go-restful/v3"
 	"github.com/pkg/errors"
 	"go.uber.org/zap"
 
@@ -35,6 +35,7 @@ import (
 	v1 "github.com/polarismesh/polaris/apiserver/httpserver/v1"
 	v2 "github.com/polarismesh/polaris/apiserver/httpserver/v2"
 	"github.com/polarismesh/polaris/auth"
+	"github.com/polarismesh/polaris/bootstrap"
 	api "github.com/polarismesh/polaris/common/api/v1"
 	"github.com/polarismesh/polaris/common/connlimit"
 	commonlog "github.com/polarismesh/polaris/common/log"
@@ -220,6 +221,7 @@ func (h *HTTPServer) Run(errCh chan error) {
 		errCh <- err
 		return
 	}
+	bootstrap.ApiServerWaitGroup.Done()
 
 	ln = &tcpKeepAliveListener{ln.(*net.TCPListener)}
 	// 开启最大连接数限制
@@ -318,7 +320,7 @@ func (h *HTTPServer) createRestfulContainer() (*restful.Container, error) {
 		Container:      wsContainer}
 	wsContainer.Filter(cors.Filter)
 
-	// Add container filter to respond to OPTIONS
+	// Incr container filter to respond to OPTIONS
 	wsContainer.Filter(wsContainer.OPTIONSFilter)
 
 	wsContainer.Filter(h.process)

diff --git a/apiserver/l5pbserver/server.go b/apiserver/l5pbserver/server.go
@@ -26,6 +26,7 @@ import (
 	"go.uber.org/zap"
 
 	"github.com/polarismesh/polaris/apiserver"
+	"github.com/polarismesh/polaris/bootstrap"
 	"github.com/polarismesh/polaris/common/api/l5"
 	"github.com/polarismesh/polaris/plugin"
 	"github.com/polarismesh/polaris/service"
@@ -105,6 +106,7 @@ func (l *L5pbserver) Run(errCh chan error) {
 		errCh <- err
 		return
 	}
+	bootstrap.ApiServerWaitGroup.Done()
 	l.listener = listener
 
 	for {

diff --git a/apiserver/prometheussd/server.go b/apiserver/prometheussd/server.go
@@ -25,10 +25,11 @@ import (
 	"strings"
 	"time"
 
-	"github.com/emicklei/go-restful/v3"
+	restful "github.com/emicklei/go-restful/v3"
 	"go.uber.org/zap"
 
 	"github.com/polarismesh/polaris/apiserver"
+	"github.com/polarismesh/polaris/bootstrap"
 	"github.com/polarismesh/polaris/common/connlimit"
 	"github.com/polarismesh/polaris/common/log"
 	"github.com/polarismesh/polaris/common/utils"
@@ -115,6 +116,7 @@ func (h *PrometheusServer) Run(errCh chan error) {
 		errCh <- err
 		return
 	}
+	bootstrap.ApiServerWaitGroup.Done()
 
 	ln = &tcpKeepAliveListener{ln.(*net.TCPListener)}
 	// 开启最大连接数限制
@@ -207,7 +209,7 @@ func (h *PrometheusServer) createRestfulContainer() (*restful.Container, error)
 		Container:      wsContainer}
 	wsContainer.Filter(cors.Filter)
 
-	// Add container filter to respond to OPTIONS
+	// Incr container filter to respond to OPTIONS
 	wsContainer.Filter(wsContainer.OPTIONSFilter)
 
 	wsContainer.Filter(h.process)

diff --git a/apiserver/xdsserverv3/server.go b/apiserver/xdsserverv3/server.go
@@ -44,7 +44,7 @@ import (
 	envoy_type_v3 "github.com/envoyproxy/go-control-plane/envoy/type/v3"
 	"github.com/envoyproxy/go-control-plane/pkg/cache/types"
 	cachev3 "github.com/envoyproxy/go-control-plane/pkg/cache/v3"
-	"github.com/envoyproxy/go-control-plane/pkg/resource/v3"
+	resource "github.com/envoyproxy/go-control-plane/pkg/resource/v3"
 	serverv3 "github.com/envoyproxy/go-control-plane/pkg/server/v3"
 	"github.com/golang/protobuf/ptypes"
 	_struct "github.com/golang/protobuf/ptypes/struct"
@@ -54,6 +54,7 @@ import (
 	"google.golang.org/protobuf/types/known/anypb"
 
 	"github.com/polarismesh/polaris/apiserver"
+	"github.com/polarismesh/polaris/bootstrap"
 	"github.com/polarismesh/polaris/cache"
 	api "github.com/polarismesh/polaris/common/api/v1"
 	"github.com/polarismesh/polaris/common/connlimit"
@@ -160,6 +161,7 @@ func (x *XDSServer) Run(errCh chan error) {
 		errCh <- err
 		return
 	}
+	bootstrap.ApiServerWaitGroup.Done()
 
 	if x.connLimitConfig != nil && x.connLimitConfig.OpenConnLimit {
 		log.Infof("grpc server use max connection limit: %d, grpc max limit: %d",

diff --git a/bootstrap/server.go b/bootstrap/server.go
@@ -23,10 +23,11 @@ import (
 	"fmt"
 	"net"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/golang/protobuf/ptypes/wrappers"
-	"gopkg.in/yaml.v2"
+	yaml "gopkg.in/yaml.v2"
 
 	"github.com/polarismesh/polaris/apiserver"
 	"github.com/polarismesh/polaris/auth"
@@ -52,6 +53,7 @@ var (
 	SelfServiceInstance = make([]*api.Instance, 0)
 	ConfigFilePath      = ""
 	selfHeathChecker    *SelfHeathChecker
+	ApiServerWaitGroup  = new(sync.WaitGroup)
 )
 
 // Start 启动
@@ -292,6 +294,9 @@ func StartServers(ctx context.Context, cfg *boot_config.Config, errCh chan error
 	[]apiserver.Apiserver, error) {
 	// 启动API服务器
 	var servers []apiserver.Apiserver
+
+	// 等待所有ApiServer都监听完成
+
 	for _, protocol := range cfg.APIServers {
 		slot, exist := apiserver.Slots[protocol.Name]
 		if !exist {
@@ -306,9 +311,10 @@ func StartServers(ctx context.Context, cfg *boot_config.Config, errCh chan error
 		}
 
 		servers = append(servers, slot)
+		ApiServerWaitGroup.Add(1)
 		go slot.Run(errCh)
 	}
-
+	ApiServerWaitGroup.Wait()
 	return servers, nil
 }
 

diff --git a/main.go b/main.go
@@ -18,9 +18,9 @@
 package main
 
 import (
-	"github.com/polarismesh/polaris/cmd"
-
 	_ "go.uber.org/automaxprocs"
+
+	"github.com/polarismesh/polaris/cmd"
 )
 
 func main() {

diff --git a/plugin/whitelist/ip_whitelist_test.go b/plugin/whitelist/ip_whitelist_test.go
@@ -21,8 +21,9 @@ import (
 	"errors"
 	"testing"
 
-	"github.com/polarismesh/polaris/plugin"
 	"github.com/stretchr/testify/assert"
+
+	"github.com/polarismesh/polaris/plugin"
 )
 
 func Test_ipWhitelist_Name(t *testing.T) {

diff --git a/store/sqldb/default.go b/store/sqldb/default.go
@@ -21,10 +21,10 @@ import (
 	"errors"
 	"fmt"
 
+	_ "github.com/go-sql-driver/mysql"
+
 	"github.com/polarismesh/polaris/plugin"
 	"github.com/polarismesh/polaris/store"
-
-	_ "github.com/go-sql-driver/mysql"
 )
 
 const (