* Added retries to initial discovering (#409)

Author: asmyasnikov

CHANGELOG.md

@@ -1,3 +1,5 @@
+* Added retries to initial discovering
+
 ## v3.38.2
 * Added missing `RetentionPeriod` parameter for topic description
 * Fixed reconnect problem for topic client

connection.go

@@ -413,7 +413,7 @@ func open(ctx context.Context, opts ...Option) (_ Connection, err error) {
 	}()
 
 	if c.pool == nil {
-		c.pool = conn.NewPool(ctx, c.config)
+		c.pool = conn.NewPool(c.config)
 	}
 
 	if c.userInfo != nil {
@@ -425,6 +425,12 @@ func open(ctx context.Context, opts ...Option) (_ Connection, err error) {
 		))
 	}
 
+	if t := c.config.DialTimeout(); t > 0 {
+		var cancel context.CancelFunc
+		ctx, cancel = context.WithTimeout(ctx, t)
+		defer cancel()
+	}
+
 	c.balancer, err = balancer.New(ctx,
 		c.config, c.pool,
 		append(

internal/balancer/balancer.go

@@ -17,6 +17,7 @@ import (
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/repeater"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xerrors"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xsync"
+	"github.com/ydb-platform/ydb-go-sdk/v3/retry"
 	"github.com/ydb-platform/ydb-go-sdk/v3/trace"
 )
 
@@ -146,6 +147,12 @@ func New(
 	pool *conn.Pool,
 	opts ...discoveryConfig.Option,
 ) (b *Balancer, err error) {
+	if t := c.DialTimeout(); t > 0 {
+		var cancel context.CancelFunc
+		ctx, cancel = context.WithTimeout(ctx, t)
+		defer cancel()
+	}
+
 	onDone := trace.DriverOnBalancerInit(
 		c.Trace(),
 		&ctx,
@@ -181,9 +188,17 @@ func New(
 			endpointsToConnections(pool, []endpoint.Endpoint{discoveryEndpoint}),
 			nil, balancerConfig.Info{}, false)
 	} else {
-		if err = b.clusterDiscovery(ctx); err != nil {
+		// initialization of balancer state
+		if err = retry.Retry(ctx, func(ctx context.Context) (err error) {
+			if err = b.clusterDiscovery(ctx); err != nil {
+				return xerrors.WithStackTrace(err)
+			}
+			return nil
+		}, retry.WithIdempotent(true)); err != nil {
 			return nil, xerrors.WithStackTrace(err)
 		}
+
+		// run background discovering
 		if d := discoveryConfig.Interval(); d > 0 {
 			b.discoveryRepeater = repeater.New(d, func(ctx context.Context) (err error) {
 				ctx, cancel := context.WithTimeout(ctx, d)
@@ -197,14 +212,6 @@ func New(
 		}
 	}
 
-	var cancel context.CancelFunc
-	if t := c.DialTimeout(); t > 0 {
-		ctx, cancel = context.WithTimeout(ctx, c.DialTimeout())
-	} else {
-		ctx, cancel = context.WithCancel(ctx)
-	}
-	defer cancel()
-
 	return b, nil
 }
 

internal/balancer/local_dc_test.go

@@ -144,7 +144,7 @@ func TestLocalDCDiscovery(t *testing.T) {
 	r := &Balancer{
 		driverConfig:   cfg,
 		balancerConfig: *cfg.Balancer(),
-		pool:           conn.NewPool(ctx, cfg),
+		pool:           conn.NewPool(cfg),
 		discovery: discoveryMock{endpoints: []endpoint.Endpoint{
 			&mock.Endpoint{AddrField: "a:123", LocationField: "a"},
 			&mock.Endpoint{AddrField: "b:234", LocationField: "b"},

internal/conn/pool.go

@@ -144,7 +144,7 @@ func (p *Pool) Release(ctx context.Context) error {
 	return nil
 }
 
-func (p *Pool) connParker(ctx context.Context, ttl, interval time.Duration) {
+func (p *Pool) connParker(ttl, interval time.Duration) {
 	ticker := time.NewTicker(interval)
 	defer ticker.Stop()
 	for {
@@ -156,7 +156,7 @@ func (p *Pool) connParker(ctx context.Context, ttl, interval time.Duration) {
 				if time.Since(c.LastUsage()) > ttl {
 					switch c.GetState() {
 					case Online, Banned:
-						_ = c.park(ctx)
+						_ = c.park(context.Background())
 					default:
 						// nop
 					}
@@ -176,10 +176,7 @@ func (p *Pool) collectConns() []*conn {
 	return conns
 }
 
-func NewPool(
-	ctx context.Context,
-	config Config,
-) *Pool {
+func NewPool(config Config) *Pool {
 	p := &Pool{
 		usages: 1,
 		config: config,
@@ -188,7 +185,7 @@ func NewPool(
 		done:   make(chan struct{}),
 	}
 	if ttl := config.ConnectionTTL(); ttl > 0 {
-		go p.connParker(ctx, ttl, ttl/2)
+		go p.connParker(ttl, ttl/2)
 	}
 	return p
 }