From 768ffd71d4369b99424458ffbbf287df342a8424 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=8C=85=E5=AD=90?= Date: Mon, 13 Mar 2023 16:05:09 +0800 Subject: [PATCH] fix(grpc/balancer): fix the problem that the watch log cannot be closed (#2726) --- transport/grpc/client.go | 37 +++++++++------ transport/grpc/resolver/discovery/builder.go | 46 +++++++++++-------- .../grpc/resolver/discovery/builder_test.go | 4 +- transport/grpc/resolver/discovery/resolver.go | 10 ++-- 4 files changed, 57 insertions(+), 40 deletions(-) diff --git a/transport/grpc/client.go b/transport/grpc/client.go index b2b5c0c2b..9c8dfa68a 100644 --- a/transport/grpc/client.go +++ b/transport/grpc/client.go @@ -109,19 +109,26 @@ func WithLogger(log log.Logger) ClientOption { return func(o *clientOptions) {} } +func WithPrintDiscoveryDebugLog(p bool) ClientOption { + return func(o *clientOptions) { + o.printDiscoveryDebugLog = p + } +} + // clientOptions is gRPC Client type clientOptions struct { - endpoint string - subsetSize int - tlsConf *tls.Config - timeout time.Duration - discovery registry.Discovery - middleware []middleware.Middleware - ints []grpc.UnaryClientInterceptor - streamInts []grpc.StreamClientInterceptor - grpcOpts []grpc.DialOption - balancerName string - filters []selector.NodeFilter + endpoint string + subsetSize int + tlsConf *tls.Config + timeout time.Duration + discovery registry.Discovery + middleware []middleware.Middleware + ints []grpc.UnaryClientInterceptor + streamInts []grpc.StreamClientInterceptor + grpcOpts []grpc.DialOption + balancerName string + filters []selector.NodeFilter + printDiscoveryDebugLog bool } // Dial returns a GRPC connection. @@ -136,9 +143,10 @@ func DialInsecure(ctx context.Context, opts ...ClientOption) (*grpc.ClientConn, func dial(ctx context.Context, insecure bool, opts ...ClientOption) (*grpc.ClientConn, error) { options := clientOptions{ - timeout: 2000 * time.Millisecond, - balancerName: balancerName, - subsetSize: 25, + timeout: 2000 * time.Millisecond, + balancerName: balancerName, + subsetSize: 25, + printDiscoveryDebugLog: true, } for _, o := range opts { o(&options) @@ -169,6 +177,7 @@ func dial(ctx context.Context, insecure bool, opts ...ClientOption) (*grpc.Clien options.discovery, discovery.WithInsecure(insecure), discovery.WithSubset(options.subsetSize), + discovery.PrintDebugLog(options.printDiscoveryDebugLog), ))) } if insecure { diff --git a/transport/grpc/resolver/discovery/builder.go b/transport/grpc/resolver/discovery/builder.go index 6186cdf1d..37a9b560c 100644 --- a/transport/grpc/resolver/discovery/builder.go +++ b/transport/grpc/resolver/discovery/builder.go @@ -38,29 +38,37 @@ func WithSubset(size int) Option { } } +// Deprecated: please use PrintDebugLog // DisableDebugLog disables update instances log. func DisableDebugLog() Option { return func(b *builder) { - b.debugLogDisabled = true + b.debugLog = false + } +} + +// PrintDebugLog print grpc resolver watch service log +func PrintDebugLog(p bool) Option { + return func(b *builder) { + b.debugLog = p } } type builder struct { - discoverer registry.Discovery - timeout time.Duration - insecure bool - subsetSize int - debugLogDisabled bool + discoverer registry.Discovery + timeout time.Duration + insecure bool + subsetSize int + debugLog bool } // NewBuilder creates a builder which is used to factory registry resolvers. func NewBuilder(d registry.Discovery, opts ...Option) resolver.Builder { b := &builder{ - discoverer: d, - timeout: time.Second * 10, - insecure: false, - debugLogDisabled: false, - subsetSize: 25, + discoverer: d, + timeout: time.Second * 10, + insecure: false, + debugLog: true, + subsetSize: 25, } for _, o := range opts { o(b) @@ -96,14 +104,14 @@ func (b *builder) Build(target resolver.Target, cc resolver.ClientConn, opts res } r := &discoveryResolver{ - w: watchRes.w, - cc: cc, - ctx: ctx, - cancel: cancel, - insecure: b.insecure, - debugLogDisabled: b.debugLogDisabled, - subsetSize: b.subsetSize, - selecterKey: uuid.New().String(), + w: watchRes.w, + cc: cc, + ctx: ctx, + cancel: cancel, + insecure: b.insecure, + debugLog: b.debugLog, + subsetSize: b.subsetSize, + selecterKey: uuid.New().String(), } go r.watch() return r, nil diff --git a/transport/grpc/resolver/discovery/builder_test.go b/transport/grpc/resolver/discovery/builder_test.go index 664b5d44d..21d5585c0 100644 --- a/transport/grpc/resolver/discovery/builder_test.go +++ b/transport/grpc/resolver/discovery/builder_test.go @@ -33,8 +33,8 @@ func TestWithTimeout(t *testing.T) { func TestDisableDebugLog(t *testing.T) { o := &builder{} DisableDebugLog()(o) - if !o.debugLogDisabled { - t.Errorf("expected debugLogDisabled true, got %v", o.debugLogDisabled) + if o.debugLog { + t.Errorf("expected debugLog true, got %v", o.debugLog) } } diff --git a/transport/grpc/resolver/discovery/resolver.go b/transport/grpc/resolver/discovery/resolver.go index a66a9b4e7..a4bd05d5b 100644 --- a/transport/grpc/resolver/discovery/resolver.go +++ b/transport/grpc/resolver/discovery/resolver.go @@ -22,10 +22,10 @@ type discoveryResolver struct { ctx context.Context cancel context.CancelFunc - insecure bool - debugLogDisabled bool - selecterKey string - subsetSize int + insecure bool + debugLog bool + selecterKey string + subsetSize int } func (r *discoveryResolver) watch() { @@ -90,7 +90,7 @@ func (r *discoveryResolver) update(ins []*registry.ServiceInstance) { log.Errorf("[resolver] failed to update state: %s", err) } - if !r.debugLogDisabled { + if r.debugLog { b, _ := json.Marshal(filtered) log.Infof("[resolver] update instances: %s", b) }