Merge pull request #552 from ydb-platform/issue551

* Added `trace.Driver.{OnBalancerDialEntrypoint,OnBalancerClusterDiscoveryAttempt}` trace events

Author: asmyasnikov

CHANGELOG.md

@@ -3,6 +3,7 @@
 * Added check of port in connection string and erro throw
 * Fixed bug with initialization of connection pool before apply static credentials
 * Refactored of applying grpc dial options with defaults
+* Added `trace.Driver.{OnBalancerDialEntrypoint,OnBalancerClusterDiscoveryAttempt}` trace events
 
 ## v3.42.8
 * Fixed `internal/scheme/helpers/IsDirectoryExists(..)` recursive bug

connection.go

@@ -457,20 +457,7 @@ func connect(ctx context.Context, c *connection) error {
 		c.pool = conn.NewPool(c.config)
 	}
 
-	c.balancer, err = balancer.New(ctx,
-		c.config, c.pool,
-		append(
-			// prepend common params from root config
-			[]discoveryConfig.Option{
-				discoveryConfig.With(c.config.Common),
-				discoveryConfig.WithEndpoint(c.Endpoint()),
-				discoveryConfig.WithDatabase(c.Name()),
-				discoveryConfig.WithSecure(c.Secure()),
-				discoveryConfig.WithMeta(c.config.Meta()),
-			},
-			c.discoveryOptions...,
-		)...,
-	)
+	c.balancer, err = balancer.New(ctx, c.config, c.pool, c.discoveryOptions...)
 	if err != nil {
 		return xerrors.WithStackTrace(err)
 	}

internal/balancer/balancer.go

@@ -32,19 +32,19 @@ type Balancer struct {
 	driverConfig      config.Config
 	balancerConfig    balancerConfig.Config
 	pool              *conn.Pool
-	discoveryClient   func(ctx context.Context) (discoveryClient, error)
+	discoveryClient   func(ctx context.Context, address string) (discoveryClient, error)
 	discoveryRepeater repeater.Repeater
 	localDCDetector   func(ctx context.Context, endpoints []endpoint.Endpoint) (string, error)
 
 	mu               xsync.RWMutex
 	connectionsState *connectionsState
 
-	onDiscovery []func(ctx context.Context, endpoints []endpoint.Info)
+	onApplyDiscoveredEndpoints []func(ctx context.Context, endpoints []endpoint.Info)
 }
 
-func (b *Balancer) OnUpdate(onDiscovery func(ctx context.Context, endpoints []endpoint.Info)) {
+func (b *Balancer) OnUpdate(onApplyDiscoveredEndpoints func(ctx context.Context, endpoints []endpoint.Info)) {
 	b.mu.WithLock(func() {
-		b.onDiscovery = append(b.onDiscovery, onDiscovery)
+		b.onApplyDiscoveredEndpoints = append(b.onApplyDiscoveredEndpoints, onApplyDiscoveredEndpoints)
 	})
 }
 
@@ -66,22 +66,18 @@ func (b *Balancer) clusterDiscovery(ctx context.Context) (err error) {
 
 func (b *Balancer) clusterDiscoveryAttempt(ctx context.Context) (err error) {
 	var (
-		onDone = trace.DriverOnBalancerUpdate(
+		address = "ydb:///" + b.driverConfig.Endpoint()
+		onDone  = trace.DriverOnBalancerClusterDiscoveryAttempt(
 			b.driverConfig.Trace(),
 			&ctx,
-			b.balancerConfig.DetectlocalDC,
+			address,
 		)
 		endpoints []endpoint.Endpoint
 		localDC   string
 		cancel    context.CancelFunc
 	)
-
 	defer func() {
-		nodes := make([]trace.EndpointInfo, 0, len(endpoints))
-		for _, e := range endpoints {
-			nodes = append(nodes, e.Copy())
-		}
-		onDone(nodes, localDC, err)
+		onDone(err)
 	}()
 
 	if dialTimeout := b.driverConfig.DialTimeout(); dialTimeout > 0 {
@@ -91,7 +87,7 @@ func (b *Balancer) clusterDiscoveryAttempt(ctx context.Context) (err error) {
 	}
 	defer cancel()
 
-	client, err := b.discoveryClient(ctx)
+	client, err := b.discoveryClient(ctx, address)
 	if err != nil {
 		return xerrors.WithStackTrace(err)
 	}
@@ -117,6 +113,19 @@ func (b *Balancer) clusterDiscoveryAttempt(ctx context.Context) (err error) {
 }
 
 func (b *Balancer) applyDiscoveredEndpoints(ctx context.Context, endpoints []endpoint.Endpoint, localDC string) {
+	onDone := trace.DriverOnBalancerUpdate(
+		b.driverConfig.Trace(),
+		&ctx,
+		b.balancerConfig.DetectlocalDC,
+	)
+	defer func() {
+		nodes := make([]trace.EndpointInfo, 0, len(endpoints))
+		for _, e := range endpoints {
+			nodes = append(nodes, e.Copy())
+		}
+		onDone(nodes, localDC, nil)
+	}()
+
 	connections := endpointsToConnections(b.pool, endpoints)
 	for _, c := range connections {
 		b.pool.Allow(ctx, c)
@@ -133,8 +142,8 @@ func (b *Balancer) applyDiscoveredEndpoints(ctx context.Context, endpoints []end
 
 	b.mu.WithLock(func() {
 		b.connectionsState = state
-		for _, onDiscovery := range b.onDiscovery {
-			onDiscovery(ctx, endpointsInfo)
+		for _, onApplyDiscoveredEndpoints := range b.onApplyDiscoveredEndpoints {
+			onApplyDiscoveredEndpoints(ctx, endpointsInfo)
 		}
 	})
 }
@@ -166,7 +175,13 @@ func New(
 			driverConfig.Trace(),
 			&ctx,
 		)
-		discoveryConfig = discoveryConfig.New(opts...)
+		discoveryConfig = discoveryConfig.New(append(opts,
+			discoveryConfig.With(driverConfig.Common),
+			discoveryConfig.WithEndpoint(driverConfig.Endpoint()),
+			discoveryConfig.WithDatabase(driverConfig.Database()),
+			discoveryConfig.WithSecure(driverConfig.Secure()),
+			discoveryConfig.WithMeta(driverConfig.Meta()),
+		)...)
 	)
 	defer func() {
 		onDone(err)
@@ -176,11 +191,16 @@ func New(
 		driverConfig:    driverConfig,
 		pool:            pool,
 		localDCDetector: detectLocalDC,
-		discoveryClient: func(ctx context.Context) (_ discoveryClient, err error) {
-			cc, err := grpc.DialContext(ctx,
-				"dns:///"+b.driverConfig.Endpoint(),
-				b.driverConfig.GrpcDialOptions()...,
+		discoveryClient: func(ctx context.Context, address string) (_ discoveryClient, err error) {
+			onBalancerDialEntrypointDone := trace.DriverOnBalancerDialEntrypoint(
+				b.driverConfig.Trace(),
+				&ctx,
+				address,
 			)
+			defer func() {
+				onBalancerDialEntrypointDone(err)
+			}()
+			cc, err := grpc.DialContext(ctx, address, b.driverConfig.GrpcDialOptions()...)
 			if err != nil {
 				return nil, xerrors.WithStackTrace(err)
 			}
@@ -195,11 +215,9 @@ func New(
 	}
 
 	if b.balancerConfig.SingleConn {
-		b.connectionsState = newConnectionsState(
-			endpointsToConnections(pool, []endpoint.Endpoint{
-				endpoint.New(driverConfig.Endpoint()),
-			}),
-			nil, balancerConfig.Info{}, false)
+		b.applyDiscoveredEndpoints(ctx, []endpoint.Endpoint{
+			endpoint.New(driverConfig.Endpoint()),
+		}, "")
 	} else {
 		// initialization of balancer state
 		if err = b.clusterDiscovery(ctx); err != nil {

internal/balancer/local_dc_test.go

@@ -142,7 +142,7 @@ func TestLocalDCDiscovery(t *testing.T) {
 		driverConfig:   cfg,
 		balancerConfig: *cfg.Balancer(),
 		pool:           conn.NewPool(cfg),
-		discoveryClient: func(context.Context) (discoveryClient, error) {
+		discoveryClient: func(context.Context, string) (discoveryClient, error) {
 			return discoveryMock{endpoints: []endpoint.Endpoint{
 				&mock.Endpoint{AddrField: "a:123", LocationField: "a"},
 				&mock.Endpoint{AddrField: "b:234", LocationField: "b"},

log/driver.go

@@ -378,33 +378,67 @@ func Driver(l Logger, details trace.Details) (t trace.Driver) {
 				}
 			}
 		}
-		t.OnBalancerUpdate = func(
-			info trace.DriverBalancerUpdateStartInfo,
+		t.OnBalancerDialEntrypoint = func(
+			info trace.DriverBalancerDialEntrypointStartInfo,
 		) func(
-			trace.DriverBalancerUpdateDoneInfo,
+			trace.DriverBalancerDialEntrypointDoneInfo,
 		) {
-			l.Tracef(
-				`balancer discovery start {needLocalDC: "%v"}`,
-				info.NeedLocalDC,
-			)
+			l.Tracef(`trying to dial entrypoint {address:"%s"}`, info.Address)
 			start := time.Now()
-			return func(info trace.DriverBalancerUpdateDoneInfo) {
+			return func(info trace.DriverBalancerDialEntrypointDoneInfo) {
 				if info.Error == nil {
-					l.Infof(
-						`balancer discovery done {latency:"%v", endpoints: "%v", detectedLocalDC: "%v"}`,
+					l.Tracef(`dial entrypoint done {latency:"%v"}`,
 						time.Since(start),
-						info.Endpoints,
-						info.LocalDC,
 					)
 				} else {
-					l.Errorf(
-						`balancer discovery failed {latency:"%v", error: "%v"}`,
+					l.Errorf(`dial entrypoint failed {latency:"%v",error:"%s",version:"%s"}`,
 						time.Since(start),
 						info.Error,
+						meta.Version,
 					)
 				}
 			}
 		}
+		t.OnBalancerClusterDiscoveryAttempt = func(
+			info trace.DriverBalancerClusterDiscoveryAttemptStartInfo,
+		) func(
+			trace.DriverBalancerClusterDiscoveryAttemptDoneInfo,
+		) {
+			l.Tracef(`trying to cluster discovery {address:"%s"}`, info.Address)
+			start := time.Now()
+			return func(info trace.DriverBalancerClusterDiscoveryAttemptDoneInfo) {
+				if info.Error == nil {
+					l.Tracef(`cluster discovery done {latency:"%v"}`,
+						time.Since(start),
+					)
+				} else {
+					l.Errorf(`cluster discovery failed {latency:"%v",error:"%s",version:"%s"}`,
+						time.Since(start),
+						info.Error,
+						meta.Version,
+					)
+				}
+			}
+		}
+		t.OnBalancerUpdate = func(
+			info trace.DriverBalancerUpdateStartInfo,
+		) func(
+			trace.DriverBalancerUpdateDoneInfo,
+		) {
+			l.Tracef(
+				`balancer update start {needLocalDC: "%v"}`,
+				info.NeedLocalDC,
+			)
+			start := time.Now()
+			return func(info trace.DriverBalancerUpdateDoneInfo) {
+				l.Infof(
+					`balancer update done {latency:"%v", endpoints: "%v", detectedLocalDC: "%v"}`,
+					time.Since(start),
+					info.Endpoints,
+					info.LocalDC,
+				)
+			}
+		}
 	}
 	if details&trace.DriverCredentialsEvents != 0 {
 		//nolint:govet

trace/driver.go

@@ -53,9 +53,23 @@ type (
 
 		// Balancer events
 		OnBalancerInit           func(DriverBalancerInitStartInfo) func(DriverBalancerInitDoneInfo)
+		OnBalancerDialEntrypoint func(
+			DriverBalancerDialEntrypointStartInfo,
+		) func(
+			DriverBalancerDialEntrypointDoneInfo,
+		)
 		OnBalancerClose          func(DriverBalancerCloseStartInfo) func(DriverBalancerCloseDoneInfo)
-		OnBalancerChooseEndpoint func(DriverBalancerChooseEndpointStartInfo) func(DriverBalancerChooseEndpointDoneInfo)
-		OnBalancerUpdate         func(DriverBalancerUpdateStartInfo) func(DriverBalancerUpdateDoneInfo)
+		OnBalancerChooseEndpoint func(
+			DriverBalancerChooseEndpointStartInfo,
+		) func(
+			DriverBalancerChooseEndpointDoneInfo,
+		)
+		OnBalancerClusterDiscoveryAttempt func(
+			DriverBalancerClusterDiscoveryAttemptStartInfo,
+		) func(
+			DriverBalancerClusterDiscoveryAttemptDoneInfo,
+		)
+		OnBalancerUpdate func(DriverBalancerUpdateStartInfo) func(DriverBalancerUpdateDoneInfo)
 
 		// Credentials events
 		OnGetCredentials func(DriverGetCredentialsStartInfo) func(DriverGetCredentialsDoneInfo)
@@ -137,7 +151,19 @@ type (
 	DriverBalancerUpdateDoneInfo struct {
 		Endpoints []EndpointInfo
 		LocalDC   string
-		Error     error
+		// Deprecated: this field always nil
+		Error error
+	}
+	DriverBalancerClusterDiscoveryAttemptStartInfo struct {
+		// Context make available context in trace callback function.
+		// Pointer to context provide replacement of context in trace callback function.
+		// Warning: concurrent access to pointer on client side must be excluded.
+		// Safe replacement of context are provided only inside callback function
+		Context *context.Context
+		Address string
+	}
+	DriverBalancerClusterDiscoveryAttemptDoneInfo struct {
+		Error error
 	}
 	DriverNetReadStartInfo struct {
 		Address string
@@ -273,6 +299,17 @@ type (
 	DriverBalancerInitDoneInfo struct {
 		Error error
 	}
+	DriverBalancerDialEntrypointStartInfo struct {
+		// Context make available context in trace callback function.
+		// Pointer to context provide replacement of context in trace callback function.
+		// Warning: concurrent access to pointer on client side must be excluded.
+		// Safe replacement of context are provided only inside callback function
+		Context *context.Context
+		Address string
+	}
+	DriverBalancerDialEntrypointDoneInfo struct {
+		Error error
+	}
 	DriverBalancerCloseStartInfo struct {
 		// Context make available context in trace callback function.
 		// Pointer to context provide replacement of context in trace callback function.