fix

Signed-off-by: kpango <[email protected]>

Author: kpango

internal/net/grpc/pool/pool.go

@@ -30,6 +30,8 @@ import (
 	"github.com/vdaas/vald/internal/errors"
 	"github.com/vdaas/vald/internal/log"
 	"github.com/vdaas/vald/internal/net"
+	"github.com/vdaas/vald/internal/net/grpc/codes"
+	"github.com/vdaas/vald/internal/net/grpc/status"
 	"github.com/vdaas/vald/internal/safety"
 	"github.com/vdaas/vald/internal/strings"
 	"github.com/vdaas/vald/internal/sync"
@@ -131,7 +133,7 @@ func New(ctx context.Context, opts ...Option) (c Conn, err error) {
 		p.addr = net.JoinHostPort(p.host, p.port)
 	}
 
-	conn, err := grpc.NewClient(p.addr, p.dopts...)
+	conn, err := p.dial(ctx, p.addr)
 	if err != nil {
 		log.Warnf("grpc.New initial Dial check to %s returned error: %v", p.addr, err)
 		if conn != nil {
@@ -147,7 +149,7 @@ func New(ctx context.Context, opts ...Option) (c Conn, err error) {
 		}
 		p.port = port
 		p.addr = net.JoinHostPort(p.host, p.port)
-		conn, err = grpc.NewClient(p.addr, p.dopts...)
+		conn, err := p.dial(ctx, p.addr)
 		if err != nil {
 			if conn != nil {
 				cerr := conn.Close()
@@ -294,9 +296,9 @@ func (p *pool) refreshConn(ctx context.Context, idx int, pc *poolConn, addr stri
 		return nil
 	}
 	if pc != nil {
-		log.Debugf("connection for %s pool %d/%d is unhealthy trying to establish new pool member connection to %s", pc.addr, idx+1, p.Size(), addr)
+		log.Debugf("connection for %s pool %d/%d len %d is unhealthy trying to establish new pool member connection to %s", pc.addr, idx+1, p.Size(), p.Len(), addr)
 	} else {
-		log.Debugf("connection pool %d/%d is empty, establish new pool member connection to %s", idx+1, p.Size(), addr)
+		log.Debugf("connection pool %d/%d len %d is empty, establish new pool member connection to %s", idx+1, p.Size(), p.Len(), addr)
 	}
 	conn, err := p.dial(ctx, addr)
 	if err != nil {
@@ -580,20 +582,27 @@ func (p *pool) Do(ctx context.Context, f func(conn *ClientConn) error) (err erro
 	}
 	conn := pc.conn
 	err = f(conn)
-	if errors.Is(err, grpc.ErrClientConnClosing) {
-		if conn != nil {
-			if cerr := conn.Close(); cerr != nil && !errors.Is(cerr, grpc.ErrClientConnClosing) {
-				log.Warnf("Failed to close connection: %v", cerr)
+	if err != nil {
+		st, ok := status.FromError(err)
+		if ok && st != nil && st.Code() != codes.Canceled { // connection closing or closed
+			if conn != nil {
+				cerr := conn.Close()
+				if cerr != nil {
+					st, ok := status.FromError(cerr)
+					if ok && st != nil && st.Code() != codes.Canceled { // connection closing or closed
+						log.Warnf("Failed to close connection: %v", cerr)
+					}
+				}
 			}
-		}
-		rerr := p.refreshConn(ctx, idx, pc, p.addr)
-		if rerr == nil {
-			if newErr := f(p.load(idx).conn); newErr != nil {
-				return errors.Join(err, newErr)
+			rerr := p.refreshConn(ctx, idx, pc, p.addr)
+			if rerr == nil {
+				if newErr := f(p.load(idx).conn); newErr != nil {
+					return errors.Join(err, newErr)
+				}
+				return nil
 			}
-			return nil
+			err = errors.Join(err, rerr)
 		}
-		err = errors.Join(err, rerr)
 	}
 	return err
 }
@@ -769,13 +778,16 @@ func (pc *poolConn) Close(ctx context.Context, delay time.Duration) error {
 		select {
 		case <-ctx.Done():
 			err := pc.conn.Close()
-			if err != nil && !errors.Is(err, grpc.ErrClientConnClosing) {
-				if ctx.Err() != nil &&
-					!errors.Is(ctx.Err(), context.DeadlineExceeded) &&
-					!errors.Is(ctx.Err(), context.Canceled) {
-					return errors.Join(err, ctx.Err())
+			if err != nil {
+				st, ok := status.FromError(err)
+				if ok && st != nil && st.Code() != codes.Canceled { // connection closing or closed
+					if ctx.Err() != nil &&
+						!errors.Is(ctx.Err(), context.DeadlineExceeded) &&
+						!errors.Is(ctx.Err(), context.Canceled) {
+						return errors.Join(err, ctx.Err())
+					}
+					return err
 				}
-				return err
 			}
 			if ctx.Err() != nil &&
 				!errors.Is(ctx.Err(), context.DeadlineExceeded) &&
@@ -787,8 +799,11 @@ func (pc *poolConn) Close(ctx context.Context, delay time.Duration) error {
 			switch pc.conn.GetState() {
 			case connectivity.Idle, connectivity.Connecting, connectivity.TransientFailure:
 				err := pc.conn.Close()
-				if err != nil && !errors.Is(err, grpc.ErrClientConnClosing) {
-					return err
+				if err != nil {
+					st, ok := status.FromError(err)
+					if ok && st != nil && st.Code() != codes.Canceled {
+						return err
+					}
 				}
 				return nil
 			case connectivity.Shutdown: