|
|
|
@ -19,7 +19,6 @@ import (
|
|
|
|
"errors"
|
|
|
|
"errors"
|
|
|
|
"fmt"
|
|
|
|
"fmt"
|
|
|
|
"net"
|
|
|
|
"net"
|
|
|
|
"net/http"
|
|
|
|
|
|
|
|
"os"
|
|
|
|
"os"
|
|
|
|
"os/signal"
|
|
|
|
"os/signal"
|
|
|
|
"reflect"
|
|
|
|
"reflect"
|
|
|
|
@ -28,21 +27,18 @@ import (
|
|
|
|
"time"
|
|
|
|
"time"
|
|
|
|
|
|
|
|
|
|
|
|
conf "github.com/openimsdk/open-im-server/v3/pkg/common/config"
|
|
|
|
conf "github.com/openimsdk/open-im-server/v3/pkg/common/config"
|
|
|
|
disetcd "github.com/openimsdk/open-im-server/v3/pkg/common/discovery/etcd"
|
|
|
|
|
|
|
|
"github.com/openimsdk/tools/discovery/etcd"
|
|
|
|
|
|
|
|
"github.com/openimsdk/tools/utils/datautil"
|
|
|
|
"github.com/openimsdk/tools/utils/datautil"
|
|
|
|
"github.com/openimsdk/tools/utils/jsonutil"
|
|
|
|
"github.com/openimsdk/tools/utils/jsonutil"
|
|
|
|
|
|
|
|
"github.com/openimsdk/tools/utils/network"
|
|
|
|
"google.golang.org/grpc/status"
|
|
|
|
"google.golang.org/grpc/status"
|
|
|
|
|
|
|
|
|
|
|
|
"github.com/openimsdk/tools/utils/runtimeenv"
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
kdisc "github.com/openimsdk/open-im-server/v3/pkg/common/discovery"
|
|
|
|
kdisc "github.com/openimsdk/open-im-server/v3/pkg/common/discovery"
|
|
|
|
"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
|
|
|
|
"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
|
|
|
|
"github.com/openimsdk/tools/discovery"
|
|
|
|
"github.com/openimsdk/tools/discovery"
|
|
|
|
"github.com/openimsdk/tools/errs"
|
|
|
|
"github.com/openimsdk/tools/errs"
|
|
|
|
"github.com/openimsdk/tools/log"
|
|
|
|
"github.com/openimsdk/tools/log"
|
|
|
|
"github.com/openimsdk/tools/mw"
|
|
|
|
grpccli "github.com/openimsdk/tools/mw/grpc/client"
|
|
|
|
"github.com/openimsdk/tools/utils/network"
|
|
|
|
grpcsrv "github.com/openimsdk/tools/mw/grpc/server"
|
|
|
|
"google.golang.org/grpc"
|
|
|
|
"google.golang.org/grpc"
|
|
|
|
"google.golang.org/grpc/credentials/insecure"
|
|
|
|
"google.golang.org/grpc/credentials/insecure"
|
|
|
|
)
|
|
|
|
)
|
|
|
|
@ -81,31 +77,59 @@ func getConfigRpcMaxRequestBody(value reflect.Value) *conf.MaxRequestBody {
|
|
|
|
return nil
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
func getConfigShare(value reflect.Value) *conf.Share {
|
|
|
|
|
|
|
|
for value.Kind() == reflect.Pointer {
|
|
|
|
|
|
|
|
value = value.Elem()
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
if value.Kind() == reflect.Struct {
|
|
|
|
|
|
|
|
num := value.NumField()
|
|
|
|
|
|
|
|
for i := 0; i < num; i++ {
|
|
|
|
|
|
|
|
field := value.Field(i)
|
|
|
|
|
|
|
|
if !field.CanInterface() {
|
|
|
|
|
|
|
|
continue
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
for field.Kind() == reflect.Pointer {
|
|
|
|
|
|
|
|
field = field.Elem()
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
switch elem := field.Interface().(type) {
|
|
|
|
|
|
|
|
case conf.Share:
|
|
|
|
|
|
|
|
return &elem
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
if field.Kind() == reflect.Struct {
|
|
|
|
|
|
|
|
if elem := getConfigShare(field); elem != nil {
|
|
|
|
|
|
|
|
return elem
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
return nil
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
func Start[T any](ctx context.Context, disc *conf.Discovery, prometheusConfig *conf.Prometheus, listenIP,
|
|
|
|
func Start[T any](ctx context.Context, disc *conf.Discovery, prometheusConfig *conf.Prometheus, listenIP,
|
|
|
|
registerIP string, autoSetPorts bool, rpcPorts []int, index int, rpcRegisterName string, notification *conf.Notification, config T,
|
|
|
|
registerIP string, autoSetPorts bool, rpcPorts []int, index int, rpcRegisterName string, notification *conf.Notification, config T,
|
|
|
|
watchConfigNames []string, watchServiceNames []string,
|
|
|
|
watchConfigNames []string, watchServiceNames []string,
|
|
|
|
rpcFn func(ctx context.Context, config T, client discovery.SvcDiscoveryRegistry, server *grpc.Server) error,
|
|
|
|
rpcFn func(ctx context.Context, config T, client discovery.Conn, server grpc.ServiceRegistrar) error,
|
|
|
|
options ...grpc.ServerOption) error {
|
|
|
|
options ...grpc.ServerOption) error {
|
|
|
|
|
|
|
|
|
|
|
|
watchConfigNames = append(watchConfigNames, conf.LogConfigFileName)
|
|
|
|
|
|
|
|
var (
|
|
|
|
|
|
|
|
rpcTcpAddr string
|
|
|
|
|
|
|
|
netDone = make(chan struct{}, 2)
|
|
|
|
|
|
|
|
netErr error
|
|
|
|
|
|
|
|
prometheusPort int
|
|
|
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
if notification != nil {
|
|
|
|
if notification != nil {
|
|
|
|
conf.InitNotification(notification)
|
|
|
|
conf.InitNotification(notification)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
maxRequestBody := getConfigRpcMaxRequestBody(reflect.ValueOf(config))
|
|
|
|
maxRequestBody := getConfigRpcMaxRequestBody(reflect.ValueOf(config))
|
|
|
|
|
|
|
|
shareConfig := getConfigShare(reflect.ValueOf(config))
|
|
|
|
|
|
|
|
|
|
|
|
log.ZDebug(ctx, "rpc start", "rpcMaxRequestBody", maxRequestBody, "rpcRegisterName", rpcRegisterName, "registerIP", registerIP, "listenIP", listenIP)
|
|
|
|
log.ZDebug(ctx, "rpc start", "rpcMaxRequestBody", maxRequestBody, "rpcRegisterName", rpcRegisterName, "registerIP", registerIP, "listenIP", listenIP)
|
|
|
|
|
|
|
|
|
|
|
|
options = append(options,
|
|
|
|
options = append(options,
|
|
|
|
mw.GrpcServer(),
|
|
|
|
grpcsrv.GrpcServerMetadataContext(),
|
|
|
|
|
|
|
|
grpcsrv.GrpcServerLogger(),
|
|
|
|
|
|
|
|
grpcsrv.GrpcServerErrorConvert(),
|
|
|
|
|
|
|
|
grpcsrv.GrpcServerRequestValidate(),
|
|
|
|
|
|
|
|
grpcsrv.GrpcServerPanicCapture(),
|
|
|
|
)
|
|
|
|
)
|
|
|
|
|
|
|
|
if shareConfig != nil && len(shareConfig.IMAdminUserID) > 0 {
|
|
|
|
|
|
|
|
options = append(options, grpcServerIMAdminUserID(shareConfig.IMAdminUserID))
|
|
|
|
|
|
|
|
}
|
|
|
|
var clientOptions []grpc.DialOption
|
|
|
|
var clientOptions []grpc.DialOption
|
|
|
|
if maxRequestBody != nil {
|
|
|
|
if maxRequestBody != nil {
|
|
|
|
if maxRequestBody.RequestMaxBodySize > 0 {
|
|
|
|
if maxRequestBody.RequestMaxBodySize > 0 {
|
|
|
|
@ -122,41 +146,32 @@ func Start[T any](ctx context.Context, disc *conf.Discovery, prometheusConfig *c
|
|
|
|
if err != nil {
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
var prometheusListenAddr string
|
|
|
|
runTimeEnv := runtimeenv.RuntimeEnvironment()
|
|
|
|
if autoSetPorts {
|
|
|
|
|
|
|
|
prometheusListenAddr = net.JoinHostPort(listenIP, "0")
|
|
|
|
if !autoSetPorts {
|
|
|
|
} else {
|
|
|
|
rpcPort, err := datautil.GetElemByIndex(rpcPorts, index)
|
|
|
|
prometheusPort, err := datautil.GetElemByIndex(prometheusConfig.Ports, index)
|
|
|
|
if err != nil {
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
}
|
|
|
|
rpcTcpAddr = net.JoinHostPort(network.GetListenIP(listenIP), strconv.Itoa(rpcPort))
|
|
|
|
prometheusListenAddr = net.JoinHostPort(listenIP, strconv.Itoa(prometheusPort))
|
|
|
|
} else {
|
|
|
|
|
|
|
|
rpcTcpAddr = net.JoinHostPort(network.GetListenIP(listenIP), "0")
|
|
|
|
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
getAutoPort := func() (net.Listener, int, error) {
|
|
|
|
watchConfigNames = append(watchConfigNames, conf.LogConfigFileName)
|
|
|
|
listener, err := net.Listen("tcp", rpcTcpAddr)
|
|
|
|
|
|
|
|
if err != nil {
|
|
|
|
|
|
|
|
return nil, 0, errs.WrapMsg(err, "listen err", "rpcTcpAddr", rpcTcpAddr)
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
_, portStr, _ := net.SplitHostPort(listener.Addr().String())
|
|
|
|
|
|
|
|
port, _ := strconv.Atoi(portStr)
|
|
|
|
|
|
|
|
return listener, port, nil
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
if autoSetPorts && discovery.Enable != conf.ETCD {
|
|
|
|
client, err := kdisc.NewDiscoveryRegister(disc, watchServiceNames)
|
|
|
|
return errs.New("only etcd support autoSetPorts", "rpcRegisterName", rpcRegisterName).Wrap()
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
client, err := kdisc.NewDiscoveryRegister(discovery, runTimeEnv, watchServiceNames)
|
|
|
|
|
|
|
|
if err != nil {
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
defer client.Close()
|
|
|
|
defer client.Close()
|
|
|
|
client.AddOption(
|
|
|
|
client.AddOption(
|
|
|
|
mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials()),
|
|
|
|
grpc.WithTransportCredentials(insecure.NewCredentials()),
|
|
|
|
grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, "round_robin")),
|
|
|
|
grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, "round_robin")),
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
grpccli.GrpcClientLogger(),
|
|
|
|
|
|
|
|
grpccli.GrpcClientContext(),
|
|
|
|
|
|
|
|
grpccli.GrpcClientErrorConvert(),
|
|
|
|
)
|
|
|
|
)
|
|
|
|
if len(clientOptions) > 0 {
|
|
|
|
if len(clientOptions) > 0 {
|
|
|
|
client.AddOption(clientOptions...)
|
|
|
|
client.AddOption(clientOptions...)
|
|
|
|
@ -178,122 +193,111 @@ func Start[T any](ctx context.Context, disc *conf.Discovery, prometheusConfig *c
|
|
|
|
|
|
|
|
|
|
|
|
if prometheusListenAddr != "" {
|
|
|
|
if prometheusListenAddr != "" {
|
|
|
|
options = append(
|
|
|
|
options = append(
|
|
|
|
options, mw.GrpcServer(),
|
|
|
|
options,
|
|
|
|
prommetricsUnaryInterceptor(rpcRegisterName),
|
|
|
|
prommetricsUnaryInterceptor(rpcRegisterName),
|
|
|
|
prommetricsStreamInterceptor(rpcRegisterName),
|
|
|
|
prommetricsStreamInterceptor(rpcRegisterName),
|
|
|
|
)
|
|
|
|
)
|
|
|
|
|
|
|
|
prometheusListener, prometheusPort, err := listenTCP(prometheusListenAddr)
|
|
|
|
var (
|
|
|
|
if err != nil {
|
|
|
|
listener net.Listener
|
|
|
|
return err
|
|
|
|
)
|
|
|
|
}
|
|
|
|
|
|
|
|
log.ZDebug(ctx, "prometheus start", "addr", prometheusListener.Addr(), "rpcRegisterName", rpcRegisterName)
|
|
|
|
if autoSetPorts {
|
|
|
|
target, err := jsonutil.JsonMarshal(prommetrics.BuildDefaultTarget(registerIP, prometheusPort))
|
|
|
|
listener, prometheusPort, err = getAutoPort()
|
|
|
|
if err != nil {
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
}
|
|
|
|
if err := client.SetKey(ctx, prommetrics.BuildDiscoveryKey(prommetrics.APIKeyName), target); err != nil {
|
|
|
|
|
|
|
|
if !errors.Is(err, discovery.ErrNotSupportedKeyValue) {
|
|
|
|
etcdClient := client.(*etcd.SvcDiscoveryRegistryImpl).GetClient()
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
_, err = etcdClient.Put(ctx, prommetrics.BuildDiscoveryKey(rpcRegisterName), jsonutil.StructToJsonString(prommetrics.BuildDefaultTarget(registerIP, prometheusPort)))
|
|
|
|
|
|
|
|
if err != nil {
|
|
|
|
|
|
|
|
return errs.WrapMsg(err, "etcd put err")
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
} else {
|
|
|
|
|
|
|
|
prometheusPort, err = datautil.GetElemByIndex(prometheusConfig.Ports, index)
|
|
|
|
|
|
|
|
if err != nil {
|
|
|
|
|
|
|
|
return err
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
}
|
|
|
|
listener, err = net.Listen("tcp", fmt.Sprintf(":%d", prometheusPort))
|
|
|
|
|
|
|
|
if err != nil {
|
|
|
|
|
|
|
|
return errs.WrapMsg(err, "listen err", "rpcTcpAddr", rpcTcpAddr)
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
}
|
|
|
|
}
|
|
|
|
cs := prommetrics.GetGrpcCusMetrics(rpcRegisterName, discovery)
|
|
|
|
|
|
|
|
go func() {
|
|
|
|
go func() {
|
|
|
|
if err := prommetrics.RpcInit(cs, listener); err != nil && !errors.Is(err, http.ErrServerClosed) {
|
|
|
|
err := prommetrics.Start(prometheusListener)
|
|
|
|
netErr = errs.WrapMsg(err, fmt.Sprintf("rpc %s prometheus start err: %d", rpcRegisterName, prometheusPort))
|
|
|
|
if err == nil {
|
|
|
|
netDone <- struct{}{}
|
|
|
|
err = fmt.Errorf("listener done")
|
|
|
|
}
|
|
|
|
}
|
|
|
|
//metric.InitializeMetrics(srv)
|
|
|
|
cancel(fmt.Errorf("prommetrics %s %w", rpcRegisterName, err))
|
|
|
|
// Create a HTTP server for prometheus.
|
|
|
|
|
|
|
|
// httpServer = &http.Server{Handler: promhttp.HandlerFor(reg, promhttp.HandlerOpts{}), Addr: fmt.Sprintf("0.0.0.0:%d", prometheusPort)}
|
|
|
|
|
|
|
|
// if err := httpServer.ListenAndServe(); err != nil && err != http.ErrServerClosed {
|
|
|
|
|
|
|
|
// netErr = errs.WrapMsg(err, "prometheus start err", httpServer.Addr)
|
|
|
|
|
|
|
|
// netDone <- struct{}{}
|
|
|
|
|
|
|
|
// }
|
|
|
|
|
|
|
|
}()
|
|
|
|
}()
|
|
|
|
} else {
|
|
|
|
|
|
|
|
options = append(options, mw.GrpcServer())
|
|
|
|
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
listener, port, err := getAutoPort()
|
|
|
|
var (
|
|
|
|
if err != nil {
|
|
|
|
rpcServer *grpc.Server
|
|
|
|
return err
|
|
|
|
rpcGracefulStop chan struct{}
|
|
|
|
}
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
log.CInfo(ctx, "RPC server is initializing", "rpcRegisterName", rpcRegisterName, "rpcPort", port,
|
|
|
|
onGrpcServiceRegistrar := func(desc *grpc.ServiceDesc, impl any) {
|
|
|
|
"prometheusPort", prometheusPort)
|
|
|
|
if rpcServer != nil {
|
|
|
|
|
|
|
|
rpcServer.RegisterService(desc, impl)
|
|
|
|
|
|
|
|
return
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
var rpcListenAddr string
|
|
|
|
|
|
|
|
if autoSetPorts {
|
|
|
|
|
|
|
|
rpcListenAddr = net.JoinHostPort(listenIP, "0")
|
|
|
|
|
|
|
|
} else {
|
|
|
|
|
|
|
|
rpcPort, err := datautil.GetElemByIndex(rpcPorts, index)
|
|
|
|
|
|
|
|
if err != nil {
|
|
|
|
|
|
|
|
cancel(fmt.Errorf("rpcPorts index out of range %s %w", rpcRegisterName, err))
|
|
|
|
|
|
|
|
return
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
rpcListenAddr = net.JoinHostPort(listenIP, strconv.Itoa(rpcPort))
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
rpcListener, err := net.Listen("tcp", rpcListenAddr)
|
|
|
|
|
|
|
|
if err != nil {
|
|
|
|
|
|
|
|
cancel(fmt.Errorf("listen rpc %s %s %w", rpcRegisterName, rpcListenAddr, err))
|
|
|
|
|
|
|
|
return
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
defer listener.Close()
|
|
|
|
rpcServer = grpc.NewServer(options...)
|
|
|
|
srv := grpc.NewServer(options...)
|
|
|
|
rpcServer.RegisterService(desc, impl)
|
|
|
|
|
|
|
|
rpcGracefulStop = make(chan struct{})
|
|
|
|
|
|
|
|
rpcPort := rpcListener.Addr().(*net.TCPAddr).Port
|
|
|
|
|
|
|
|
log.ZDebug(ctx, "rpc start register", "rpcRegisterName", rpcRegisterName, "registerIP", registerIP, "rpcPort", rpcPort)
|
|
|
|
|
|
|
|
grpcOpt := grpc.WithTransportCredentials(insecure.NewCredentials())
|
|
|
|
|
|
|
|
rpcGracefulStop = make(chan struct{})
|
|
|
|
|
|
|
|
go func() {
|
|
|
|
|
|
|
|
<-ctx.Done()
|
|
|
|
|
|
|
|
rpcServer.GracefulStop()
|
|
|
|
|
|
|
|
close(rpcGracefulStop)
|
|
|
|
|
|
|
|
}()
|
|
|
|
|
|
|
|
if err := client.Register(ctx, rpcRegisterName, registerIP, rpcListener.Addr().(*net.TCPAddr).Port, grpcOpt); err != nil {
|
|
|
|
|
|
|
|
cancel(fmt.Errorf("rpc register %s %w", rpcRegisterName, err))
|
|
|
|
|
|
|
|
return
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
err = rpcFn(ctx, config, client, srv)
|
|
|
|
go func() {
|
|
|
|
if err != nil {
|
|
|
|
err := rpcServer.Serve(rpcListener)
|
|
|
|
return err
|
|
|
|
if err == nil {
|
|
|
|
|
|
|
|
err = fmt.Errorf("serve end")
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
cancel(fmt.Errorf("rpc %s %w", rpcRegisterName, err))
|
|
|
|
|
|
|
|
}()
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
err = client.Register(
|
|
|
|
err = rpcFn(ctx, config, client, &grpcServiceRegistrar{onRegisterService: onGrpcServiceRegistrar})
|
|
|
|
ctx,
|
|
|
|
|
|
|
|
rpcRegisterName,
|
|
|
|
|
|
|
|
registerIP,
|
|
|
|
|
|
|
|
port,
|
|
|
|
|
|
|
|
grpc.WithTransportCredentials(insecure.NewCredentials()),
|
|
|
|
|
|
|
|
)
|
|
|
|
|
|
|
|
if err != nil {
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
<-ctx.Done()
|
|
|
|
go func() {
|
|
|
|
log.ZDebug(ctx, "cmd wait done", "err", context.Cause(ctx))
|
|
|
|
err := srv.Serve(listener)
|
|
|
|
if rpcGracefulStop != nil {
|
|
|
|
if err != nil && !errors.Is(err, http.ErrServerClosed) {
|
|
|
|
timeout := time.NewTimer(time.Second * 15)
|
|
|
|
netErr = errs.WrapMsg(err, "rpc start err: ", rpcTcpAddr)
|
|
|
|
defer timeout.Stop()
|
|
|
|
netDone <- struct{}{}
|
|
|
|
select {
|
|
|
|
}
|
|
|
|
case <-timeout.C:
|
|
|
|
}()
|
|
|
|
log.ZWarn(ctx, "rcp graceful stop timeout", nil)
|
|
|
|
|
|
|
|
case <-rpcGracefulStop:
|
|
|
|
if discovery.Enable == conf.ETCD {
|
|
|
|
log.ZDebug(ctx, "rcp graceful stop done")
|
|
|
|
cm := disetcd.NewConfigManager(client.(*etcd.SvcDiscoveryRegistryImpl).GetClient(), watchConfigNames)
|
|
|
|
|
|
|
|
cm.Watch(ctx)
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
sigs := make(chan os.Signal, 1)
|
|
|
|
|
|
|
|
signal.Notify(sigs, syscall.SIGTERM)
|
|
|
|
|
|
|
|
select {
|
|
|
|
|
|
|
|
case <-sigs:
|
|
|
|
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
|
|
|
|
|
|
|
defer cancel()
|
|
|
|
|
|
|
|
if err := gracefulStopWithCtx(ctx, srv.GracefulStop); err != nil {
|
|
|
|
|
|
|
|
return err
|
|
|
|
|
|
|
|
}
|
|
|
|
}
|
|
|
|
return nil
|
|
|
|
|
|
|
|
case <-netDone:
|
|
|
|
|
|
|
|
return netErr
|
|
|
|
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
return context.Cause(ctx)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
func gracefulStopWithCtx(ctx context.Context, f func()) error {
|
|
|
|
func listenTCP(addr string) (net.Listener, int, error) {
|
|
|
|
done := make(chan struct{}, 1)
|
|
|
|
listener, err := net.Listen("tcp", addr)
|
|
|
|
go func() {
|
|
|
|
if err != nil {
|
|
|
|
f()
|
|
|
|
return nil, 0, errs.WrapMsg(err, "listen err", "addr", addr)
|
|
|
|
close(done)
|
|
|
|
|
|
|
|
}()
|
|
|
|
|
|
|
|
select {
|
|
|
|
|
|
|
|
case <-ctx.Done():
|
|
|
|
|
|
|
|
return errs.New("timeout, ctx graceful stop")
|
|
|
|
|
|
|
|
case <-done:
|
|
|
|
|
|
|
|
return nil
|
|
|
|
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
return listener, listener.Addr().(*net.TCPAddr).Port, nil
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
func prommetricsUnaryInterceptor(rpcRegisterName string) grpc.ServerOption {
|
|
|
|
func prommetricsUnaryInterceptor(rpcRegisterName string) grpc.ServerOption {
|
|
|
|
@ -317,3 +321,11 @@ func prommetricsUnaryInterceptor(rpcRegisterName string) grpc.ServerOption {
|
|
|
|
func prommetricsStreamInterceptor(rpcRegisterName string) grpc.ServerOption {
|
|
|
|
func prommetricsStreamInterceptor(rpcRegisterName string) grpc.ServerOption {
|
|
|
|
return grpc.ChainStreamInterceptor()
|
|
|
|
return grpc.ChainStreamInterceptor()
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
type grpcServiceRegistrar struct {
|
|
|
|
|
|
|
|
onRegisterService func(desc *grpc.ServiceDesc, impl any)
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
func (x *grpcServiceRegistrar) RegisterService(desc *grpc.ServiceDesc, impl any) {
|
|
|
|
|
|
|
|
x.onRegisterService(desc, impl)
|
|
|
|
|
|
|
|
}
|
|
|
|
|
chao
4 months ago
committed by
GitHub