[patch] Bug fix on StreamGetObject API and DNS cache expiration / refactor net connection

internal/config/meta.go

@@ -17,7 +17,11 @@
 // Package config providers configuration type and load configuration logic
 package config
 
-import "fmt"
+import (
+	"strconv"
+
+	"github.com/vdaas/vald/internal/net"
+)
 
 type Meta struct {
 	Host                      string      `json:"host" yaml:"host"`
@@ -39,7 +43,7 @@ func (m *Meta) Bind() *Meta {
 		m.Client = newGRPCClientConfig()
 	}
 	if len(m.Host) != 0 {
-		m.Client.Addrs = append(m.Client.Addrs, fmt.Sprintf("%s:%d", m.Host, m.Port))
+		m.Client.Addrs = append(m.Client.Addrs, net.JoinHostPort(m.Host, strconv.FormatInt(int64(m.Port), 10)))
 	}
 	return m
 }

internal/config/tcp.go

@@ -32,6 +32,7 @@ type TCP struct {
 type Dialer struct {
 	Timeout          string `yaml:"timeout" json:"timeout"`
 	KeepAlive        string `yaml:"keep_alive" json:"keep_alive"`
+	FallbackDelay    string `yaml:"fallback_delay" json:"fallback_delay"`
 	DualStackEnabled bool   `yaml:"dual_stack_enabled" json:"dual_stack_enabled"`
 }
 
@@ -50,6 +51,7 @@ func (d *DNS) Bind() *DNS {
 func (d *Dialer) Bind() *Dialer {
 	d.Timeout = GetActualValue(d.Timeout)
 	d.KeepAlive = GetActualValue(d.KeepAlive)
+	d.FallbackDelay = GetActualValue(d.FallbackDelay)
 	return d
 }
 
@@ -84,6 +86,7 @@ func (t *TCP) Opts() []tcp.DialerOption {
 		opts = append(opts,
 			tcp.WithDialerKeepAlive(t.Dialer.KeepAlive),
 			tcp.WithDialerTimeout(t.Dialer.Timeout),
+			tcp.WithDialerFallbackDelay(t.Dialer.FallbackDelay),
 		)
 		if t.Dialer.DualStackEnabled {
 			opts = append(opts,

internal/db/nosql/cassandra/cassandra.go

@@ -27,6 +27,7 @@ import (
 	"github.com/scylladb/gocqlx/qb"
 	"github.com/vdaas/vald/internal/errors"
 	"github.com/vdaas/vald/internal/log"
+	"github.com/vdaas/vald/internal/net/tcp"
 )
 
 var (
@@ -156,6 +157,7 @@ type (
 		connectObserver          ConnectObserver
 		frameHeaderObserver      FrameHeaderObserver
 		defaultIdempotence       bool
+		rawDialer                tcp.Dialer
 		dialer                   gocql.Dialer
 		writeCoalesceWaitTime    time.Duration
 
@@ -305,6 +307,9 @@ func (c *client) Open(ctx context.Context) (err error) {
 		log.Debugf("failed to create session %#v", c)
 		return errors.ErrCassandraFailedToCreateSession(err, c.hosts, c.port, c.cqlVersion)
 	}
+	if c.rawDialer != nil {
+		c.rawDialer.StartDialerCache(ctx)
+	}
 	return nil
 }
 

internal/db/nosql/cassandra/cassandra_mock_test.go

@@ -34,6 +34,11 @@ func (dm *DialerMock) DialContext(ctx context.Context, network, addr string) (ne
 	return dm.DialContextFunc(ctx, network, addr)
 }
 
+func (dm *DialerMock) GetDialer() func(ctx context.Context, network, addr string) (net.Conn, error) {
+	return dm.DialContextFunc
+}
+func (dm *DialerMock) StartDialerCache(ctx context.Context) {}
+
 func TestMockClusterConfig_CreateSession(t *testing.T) {
 	t.Parallel()
 	type fields struct {

internal/db/nosql/cassandra/option.go

@@ -25,6 +25,7 @@ import (
 
 	"github.com/gocql/gocql"
 	"github.com/vdaas/vald/internal/errors"
+	"github.com/vdaas/vald/internal/net/tcp"
 	"github.com/vdaas/vald/internal/timeutil"
 )
 
@@ -79,11 +80,12 @@ func WithHosts(hosts ...string) Option {
 }
 
 // WithDialer returns the option to set the dialer.
-func WithDialer(der gocql.Dialer) Option {
+func WithDialer(der tcp.Dialer) Option {
 	return func(c *client) error {
 		if der == nil {
 			return errors.NewErrInvalidOption("dialer", der)
 		}
+		c.rawDialer = der
 		c.dialer = der
 		return nil
 	}

internal/db/nosql/cassandra/option_test.go

@@ -26,6 +26,7 @@ import (
 
 	"github.com/gocql/gocql"
 	"github.com/vdaas/vald/internal/errors"
+	"github.com/vdaas/vald/internal/net/tcp"
 	"github.com/vdaas/vald/internal/test/comparator"
 	"go.uber.org/goleak"
 )
@@ -147,7 +148,7 @@ func TestWithHosts(t *testing.T) {
 func TestWithDialer(t *testing.T) {
 	type T = client
 	type args struct {
-		der gocql.Dialer
+		der tcp.Dialer
 	}
 	type want struct {
 		obj *T

internal/net/grpc/client.go

@@ -30,6 +30,7 @@ import (
 	"github.com/vdaas/vald/internal/errors"
 	"github.com/vdaas/vald/internal/log"
 	"github.com/vdaas/vald/internal/net/grpc/pool"
+	"github.com/vdaas/vald/internal/net/tcp"
 	"github.com/vdaas/vald/internal/observability/trace"
 	"github.com/vdaas/vald/internal/safety"
 	"github.com/vdaas/vald/internal/singleflight"
@@ -93,6 +94,7 @@ type gRPCClient struct {
 	conns               grpcConns
 	hcDur               time.Duration
 	prDur               time.Duration
+	dialer              tcp.Dialer
 	enablePoolRebalance bool
 	resolveDNS          bool
 	dopts               []DialOption
@@ -138,6 +140,10 @@ func (g *gRPCClient) StartConnectionMonitor(ctx context.Context) (<-chan error,
 		return nil, errors.ErrGRPCTargetAddrNotFound
 	}
 
+	if g.dialer != nil {
+		g.dialer.StartDialerCache(ctx)
+	}
+
 	ech := make(chan error, len(addrs))
 	for _, addr := range addrs {
 		if len(addr) != 0 {

internal/net/grpc/option.go

@@ -322,6 +322,7 @@ func WithKeepaliveParams(t, to string, permitWithoutStream bool) Option {
 func WithDialer(der tcp.Dialer) Option {
 	return func(g *gRPCClient) {
 		if der != nil {
+			g.dialer = der
 			g.dopts = append(g.dopts,
 				grpc.WithContextDialer(func(ctx context.Context, addr string) (net.Conn, error) {
 					return der.GetDialer()(ctx, "tcp", addr)

internal/net/grpc/pool/pool.go

@@ -21,6 +21,7 @@ import (
 	"context"
 	"fmt"
 	"sort"
+	"strconv"
 	"strings"
 	"sync/atomic"
 	"time"
@@ -97,7 +98,7 @@ func New(ctx context.Context, opts ...Option) (c Conn, err error) {
 		if err != nil {
 			return nil, err
 		}
-		p.addr = fmt.Sprintf("%s:%d", p.host, p.port)
+		p.addr = net.JoinHostPort(p.host, strconv.FormatInt(int64(p.port), 10))
 	}
 
 	conn, err := grpc.DialContext(ctx, p.addr, p.dopts...)
@@ -107,7 +108,7 @@ func New(ctx context.Context, opts ...Option) (c Conn, err error) {
 		if err != nil {
 			return nil, err
 		}
-		p.addr = fmt.Sprintf("%s:%d", p.host, p.port)
+		p.addr = net.JoinHostPort(p.host, strconv.FormatInt(int64(p.port), 10))
 	}
 	if conn != nil {
 		err = conn.Close()
@@ -144,7 +145,7 @@ func (p *pool) Connect(ctx context.Context) (c Conn, err error) {
 		default:
 			var (
 				conn   *ClientConn
-				addr   = fmt.Sprintf("%s:%d", ips[i%len(ips)], p.port)
+				addr   = net.JoinHostPort(ips[i%len(ips)], strconv.FormatInt(int64(p.port), 10))
 				pc, ok = p.load(i)
 			)
 			if ok && pc != nil && pc.addr == addr && isHealthy(pc.conn) {
@@ -387,7 +388,7 @@ func (p *pool) lookupIPAddr(ctx context.Context) (ips []string, err error) {
 			ipStr = fmt.Sprintf("[%s]", ipStr)
 		}
 		var conn net.Conn
-		addr := fmt.Sprintf("%s:%d", ipStr, p.port)
+		addr := net.JoinHostPort(ipStr, strconv.FormatInt(int64(p.port), 10))
 		if net.DefaultResolver.Dial != nil {
 			ctx, cancel := context.WithTimeout(ctx, time.Millisecond*10)
 			conn, err = net.DefaultResolver.Dial(ctx, network, addr)
@@ -470,7 +471,8 @@ func (p *pool) scanGRPCPort(ctx context.Context) (err error) {
 func isGRPCPort(ctx context.Context, host string, port uint16) bool {
 	ctx, cancel := context.WithTimeout(ctx, time.Millisecond*5)
 	defer cancel()
-	conn, err := grpc.DialContext(ctx, fmt.Sprintf("%s:%d", host, port),
+	conn, err := grpc.DialContext(ctx,
+		net.JoinHostPort(host, strconv.FormatInt(int64(port), 10)),
 		grpc.WithInsecure(),
 		grpc.WithBlock(),
 	)

internal/net/net.go

@@ -19,7 +19,6 @@ package net
 
 import (
 	"context"
-	"fmt"
 	"net"
 	"strconv"
 	"strings"
@@ -29,6 +28,7 @@ import (
 	"github.com/vdaas/vald/internal/errgroup"
 	"github.com/vdaas/vald/internal/errors"
 	"github.com/vdaas/vald/internal/log"
+	"github.com/vdaas/vald/internal/safety"
 )
 
 const (
@@ -97,6 +97,11 @@ func IsIPv4(addr string) bool {
 	return net.ParseIP(addr) != nil && strings.Count(addr, ":") < 2
 }
 
+// JoinHostPort joins the host/IP address and the port number,
+func JoinHostPort(host, port string) string {
+	return net.JoinHostPort(host, port)
+}
+
 // SplitHostPort splits the address, and return the host/IP address and the port number,
 // and any error occurred.
 // If it is the loopback address, it will return the loopback address and corresponding port number.
@@ -136,14 +141,15 @@ func ScanPorts(ctx context.Context, start, end uint16, host string) (ports []uin
 
 	var mu sync.Mutex
 
+	var der net.Dialer
 	for i := start; i >= start && i <= end; i++ {
 		port := i
-		eg.Go(func() error {
+		eg.Go(safety.RecoverFunc(func() error {
 			select {
 			case <-egctx.Done():
 				return egctx.Err()
 			default:
-				conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", host, port))
+				conn, err := der.DialContext(ctx, "tcp", JoinHostPort(host, strconv.FormatInt(int64(port), 10)))
 				if err != nil {
 					log.Warn(err)
 					return nil
@@ -158,7 +164,7 @@ func ScanPorts(ctx context.Context, start, end uint16, host string) (ports []uin
 				}
 				return nil
 			}
-		})
+		}))
 	}
 
 	if err = eg.Wait(); err != nil {