Merge pull request #1145 from ydb-platform/slo-logs

Performance of QueryService client over SLO + refactoring of traces

Author: asmyasnikov

CHANGELOG.md

@@ -3,6 +3,11 @@
 * Added query pool metrics
 * Fixed logic of query session pool
 * Changed initialization of internal driver clients to lazy
+* Disabled the logic of background grpc-connection parking 
+* Removed `ydb.WithSessionPoolSizeLimit()` option
+* Added async put session into pool if external context is done
+* Dropped intermediate callbacks from `trace.{Table,Retry,Query}` events
+* Wrapped errors from `internal/pool.Pool.getItem` as retryable
 * Disabled the logic of background grpc-connection parking
 * Improved stringification for postgres types
 

driver.go

@@ -300,7 +300,7 @@ func New(ctx context.Context, opts ...Option) (_ *Driver, err error) {
 
 //nolint:cyclop, nonamedreturns
 func newConnectionFromOptions(ctx context.Context, opts ...Option) (_ *Driver, err error) {
-	ctx, driverCtxCancel := xcontext.WithCancel(xcontext.WithoutDeadline(ctx))
+	ctx, driverCtxCancel := xcontext.WithCancel(xcontext.ValueOnly(ctx))
 	defer func() {
 		if err != nil {
 			driverCtxCancel()
@@ -401,7 +401,7 @@ func (d *Driver) connect(ctx context.Context) (err error) {
 	}
 
 	d.table = xsync.OnceValue(func() *internalTable.Client {
-		return internalTable.New(xcontext.WithoutDeadline(ctx),
+		return internalTable.New(xcontext.ValueOnly(ctx),
 			d.balancer,
 			tableConfig.New(
 				append(
@@ -416,7 +416,7 @@ func (d *Driver) connect(ctx context.Context) (err error) {
 	})
 
 	d.query = xsync.OnceValue(func() *internalQuery.Client {
-		return internalQuery.New(xcontext.WithoutDeadline(ctx),
+		return internalQuery.New(xcontext.ValueOnly(ctx),
 			d.balancer,
 			queryConfig.New(
 				append(
@@ -434,7 +434,7 @@ func (d *Driver) connect(ctx context.Context) (err error) {
 	}
 
 	d.scheme = xsync.OnceValue(func() *internalScheme.Client {
-		return internalScheme.New(xcontext.WithoutDeadline(ctx),
+		return internalScheme.New(xcontext.ValueOnly(ctx),
 			d.balancer,
 			schemeConfig.New(
 				append(
@@ -450,7 +450,7 @@ func (d *Driver) connect(ctx context.Context) (err error) {
 	})
 
 	d.coordination = xsync.OnceValue(func() *internalCoordination.Client {
-		return internalCoordination.New(xcontext.WithoutDeadline(ctx),
+		return internalCoordination.New(xcontext.ValueOnly(ctx),
 			d.balancer,
 			coordinationConfig.New(
 				append(
@@ -465,7 +465,7 @@ func (d *Driver) connect(ctx context.Context) (err error) {
 	})
 
 	d.ratelimiter = xsync.OnceValue(func() *internalRatelimiter.Client {
-		return internalRatelimiter.New(xcontext.WithoutDeadline(ctx),
+		return internalRatelimiter.New(xcontext.ValueOnly(ctx),
 			d.balancer,
 			ratelimiterConfig.New(
 				append(
@@ -480,7 +480,7 @@ func (d *Driver) connect(ctx context.Context) (err error) {
 	})
 
 	d.discovery = xsync.OnceValue(func() *internalDiscovery.Client {
-		return internalDiscovery.New(xcontext.WithoutDeadline(ctx),
+		return internalDiscovery.New(xcontext.ValueOnly(ctx),
 			d.pool.Get(endpoint.New(d.config.Endpoint())),
 			discoveryConfig.New(
 				append(
@@ -499,7 +499,7 @@ func (d *Driver) connect(ctx context.Context) (err error) {
 	})
 
 	d.scripting = xsync.OnceValue(func() *internalScripting.Client {
-		return internalScripting.New(xcontext.WithoutDeadline(ctx),
+		return internalScripting.New(xcontext.ValueOnly(ctx),
 			d.balancer,
 			scriptingConfig.New(
 				append(
@@ -514,7 +514,7 @@ func (d *Driver) connect(ctx context.Context) (err error) {
 	})
 
 	d.topic = xsync.OnceValue(func() *topicclientinternal.Client {
-		return topicclientinternal.New(xcontext.WithoutDeadline(ctx),
+		return topicclientinternal.New(xcontext.ValueOnly(ctx),
 			d.balancer,
 			d.config.Credentials(),
 			append(

internal/balancer/balancer.go

@@ -280,7 +280,7 @@ func New(
 		}
 		// run background discovering
 		if d := discoveryConfig.Interval(); d > 0 {
-			b.discoveryRepeater = repeater.New(xcontext.WithoutDeadline(ctx),
+			b.discoveryRepeater = repeater.New(xcontext.ValueOnly(ctx),
 				d, b.clusterDiscoveryAttempt,
 				repeater.WithName("discovery"),
 				repeater.WithTrace(b.driverConfig.Trace()),

internal/conn/conn.go

@@ -183,6 +183,10 @@ func (c *conn) realConn(ctx context.Context) (cc *grpc.ClientConn, err error) {
 		}, c.config.GrpcDialOptions()...,
 	)...)
 	if err != nil {
+		if xerrors.IsContextError(err) {
+			return nil, xerrors.WithStackTrace(err)
+		}
+
 		defer func() {
 			c.onTransportError(ctx, err)
 		}()
@@ -310,6 +314,10 @@ func (c *conn) Invoke(
 
 	err = cc.Invoke(ctx, method, req, res, append(opts, grpc.Trailer(&md))...)
 	if err != nil {
+		if xerrors.IsContextError(err) {
+			return xerrors.WithStackTrace(err)
+		}
+
 		defer func() {
 			c.onTransportError(ctx, err)
 		}()
@@ -392,6 +400,10 @@ func (c *conn) NewStream(
 
 	s, err = cc.NewStream(ctx, desc, method, opts...)
 	if err != nil {
+		if xerrors.IsContextError(err) {
+			return nil, xerrors.WithStackTrace(err)
+		}
+
 		defer func() {
 			c.onTransportError(ctx, err)
 		}()

internal/conn/grpc_client_stream.go

@@ -33,6 +33,10 @@ func (s *grpcClientStream) CloseSend() (err error) {
 	err = s.ClientStream.CloseSend()
 
 	if err != nil {
+		if xerrors.IsContextError(err) {
+			return xerrors.WithStackTrace(err)
+		}
+
 		if s.wrapping {
 			return s.wrapError(
 				xerrors.Transport(
@@ -58,6 +62,10 @@ func (s *grpcClientStream) SendMsg(m interface{}) (err error) {
 	err = s.ClientStream.SendMsg(m)
 
 	if err != nil {
+		if xerrors.IsContextError(err) {
+			return xerrors.WithStackTrace(err)
+		}
+
 		defer func() {
 			s.c.onTransportError(s.Context(), err)
 		}()
@@ -97,7 +105,11 @@ func (s *grpcClientStream) RecvMsg(m interface{}) (err error) {
 
 	err = s.ClientStream.RecvMsg(m)
 
-	if err != nil {
+	if err != nil { //nolint:nestif
+		if xerrors.IsContextError(err) {
+			return xerrors.WithStackTrace(err)
+		}
+
 		defer func() {
 			if !xerrors.Is(err, io.EOF) {
 				s.c.onTransportError(s.Context(), err)

internal/conn/pool.go

@@ -78,11 +78,24 @@ func (p *Pool) Ban(ctx context.Context, cc Conn, cause error) {
 		return
 	}
 
-	if xerrors.IsTransportError(cause,
-		grpcCodes.OK,
-		grpcCodes.Canceled,
+	if !xerrors.IsTransportError(cause,
 		grpcCodes.ResourceExhausted,
-		grpcCodes.OutOfRange,
+		grpcCodes.Unavailable,
+		// grpcCodes.OK,
+		// grpcCodes.Canceled,
+		// grpcCodes.Unknown,
+		// grpcCodes.InvalidArgument,
+		// grpcCodes.DeadlineExceeded,
+		// grpcCodes.NotFound,
+		// grpcCodes.AlreadyExists,
+		// grpcCodes.PermissionDenied,
+		// grpcCodes.FailedPrecondition,
+		// grpcCodes.Aborted,
+		// grpcCodes.OutOfRange,
+		// grpcCodes.Unimplemented,
+		// grpcCodes.Internal,
+		// grpcCodes.DataLoss,
+		// grpcCodes.Unauthenticated,
 	) {
 		return
 	}

internal/pool/errors.go

@@ -2,12 +2,9 @@ package pool
 
 import (
 	"errors"
-
-	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xerrors"
 )
 
 var (
 	errClosedPool     = errors.New("closed pool")
-	errPoolOverflow   = xerrors.Retryable(errors.New("pool overflow"))
 	errItemIsNotAlive = errors.New("item is not alive")
 )