Merge branch 'master' into allow-skip-sql-fields

Author: asmyasnikov

CHANGELOG.md

@@ -1,5 +1,14 @@
 * Allowed skip column for `ScanStruct` by tag `-` 
 
+## v3.81.4
+* Returned `topicwriter.ErrQueueLimitExceed`, accidental removed at `v3.81.0`
+
+## v3.81.3
+* Fixed tracing details check for some metrics
+
+## v3.81.2
+* Removed `experimantal` comment for query service client
+
 ## v3.81.1
 * Fixed nil pointer dereference panic on failed `ydb.Open` 
 * Added ip discovery. Server can show own ip address and target hostname in the ListEndpoint message. These fields are used to bypass DNS resolving.

driver.go

@@ -194,8 +194,6 @@ func (d *Driver) Table() table.Client {
 }
 
 // Query returns query client
-//
-// Experimental: https://github.com/ydb-platform/ydb-go-sdk/blob/master/VERSIONING.md#experimental
 func (d *Driver) Query() *internalQuery.Client {
 	return d.query.Must()
 }

internal/table/client.go

@@ -100,51 +100,58 @@ func (c *Client) CreateSession(ctx context.Context, opts ...table.Option) (_ tab
 	if c.isClosed() {
 		return nil, xerrors.WithStackTrace(errClosedClient)
 	}
-	var s *session
 	createSession := func(ctx context.Context) (*session, error) {
-		s, err = c.build(ctx)
+		s, err := c.build(ctx)
 		if err != nil {
 			return nil, xerrors.WithStackTrace(err)
 		}
 
 		return s, nil
 	}
 	if !c.config.AutoRetry() {
-		s, err = createSession(ctx)
+		s, err := createSession(ctx)
 		if err != nil {
 			return nil, xerrors.WithStackTrace(err)
 		}
 
 		return s, nil
 	}
-	err = retry.Retry(ctx,
-		func(ctx context.Context) (err error) {
-			s, err = createSession(ctx)
-			if err != nil {
-				return xerrors.WithStackTrace(err)
-			}
 
-			return nil
-		},
+	var (
+		onDone = trace.TableOnCreateSession(c.config.Trace(), &ctx,
+			stack.FunctionID(
+				"github.com/ydb-platform/ydb-go-sdk/v3/internal/table.(*Client).CreateSession"),
+		)
+		attempts = 0
+		s        *session
+	)
+	defer func() {
+		if s != nil {
+			onDone(s, attempts, err)
+		} else {
+			onDone(nil, attempts, err)
+		}
+	}()
+
+	s, err = retry.RetryWithResult(ctx, createSession,
 		append(
 			[]retry.Option{
 				retry.WithIdempotent(true),
 				retry.WithTrace(&trace.Retry{
 					OnRetry: func(info trace.RetryLoopStartInfo) func(trace.RetryLoopDoneInfo) {
-						onDone := trace.TableOnCreateSession(c.config.Trace(), info.Context,
-							stack.FunctionID(
-								"github.com/ydb-platform/ydb-go-sdk/v3/internal/table.(*Client).CreateSession"))
-
 						return func(info trace.RetryLoopDoneInfo) {
-							onDone(s, info.Attempts, info.Error)
+							attempts = info.Attempts
 						}
 					},
 				}),
 			}, c.retryOptions(opts...).RetryOptions...,
 		)...,
 	)
+	if err != nil {
+		return nil, xerrors.WithStackTrace(err)
+	}
 
-	return s, xerrors.WithStackTrace(err)
+	return s, nil
 }
 
 func (c *Client) isClosed() bool {

internal/topic/topicwriterinternal/writer_reconnector.go

@@ -36,7 +36,8 @@ var (
 	errNonZeroSeqNo                                = xerrors.Wrap(errors.New("ydb: non zero seqno for auto set seqno mode"))                         //nolint:lll
 	errNonZeroCreatedAt                            = xerrors.Wrap(errors.New("ydb: non zero Message.CreatedAt and set auto fill created at option")) //nolint:lll
 	errNoAllowedCodecs                             = xerrors.Wrap(errors.New("ydb: no allowed codecs for write to topic"))
-	errLargeMessage                                = xerrors.Wrap(errors.New("ydb: message uncompressed size more, then limit"))                                                                                                                                                                                             //nolint:lll
+	errLargeMessage                                = xerrors.Wrap(errors.New("ydb: message uncompressed size more, then limit")) //nolint:lll
+	PublicErrQueueIsFull                           = xerrors.Wrap(errors.New("ydb: queue is full"))
 	PublicErrMessagesPutToInternalQueueBeforeError = xerrors.Wrap(errors.New("ydb: the messages was put to internal buffer before the error happened. It mean about the messages can be delivered to the server"))                                                                                                           //nolint:lll
 	errDiffetentTransactions                       = xerrors.Wrap(errors.New("ydb: internal writer has messages from different trasactions. It is internal logic error, write issue please: https://github.com/ydb-platform/ydb-go-sdk/issues/new?assignees=&labels=bug&projects=&template=01_BUG_REPORT.md&title=bug%3A+")) //nolint:lll
 
@@ -223,16 +224,18 @@ func (w *WriterReconnector) Write(ctx context.Context, messages []PublicMessage)
 	semaphoreWeight := int64(len(messages))
 	if semaphoreWeight > int64(w.cfg.MaxQueueLen) {
 		return xerrors.WithStackTrace(fmt.Errorf(
-			"ydb: add more messages, then max queue limit. max queue: %v, try to add: %v",
+			"ydb: add more messages, then max queue limit. max queue: %v, try to add: %v: %w",
 			w.cfg.MaxQueueLen,
 			semaphoreWeight,
+			PublicErrQueueIsFull,
 		))
 	}
 	if err := w.semaphore.Acquire(ctx, semaphoreWeight); err != nil {
 		return xerrors.WithStackTrace(
-			fmt.Errorf("ydb: add new messages exceed max queue size limit. Add count: %v, max size: %v",
+			fmt.Errorf("ydb: add new messages exceed max queue size limit. Add count: %v, max size: %v: %w",
 				semaphoreWeight,
 				w.cfg.MaxQueueLen,
+				PublicErrQueueIsFull,
 			))
 	}
 	defer func() {

internal/version/version.go

@@ -3,7 +3,7 @@ package version
 const (
 	Major = "3"
 	Minor = "81"
-	Patch = "1"
+	Patch = "4"
 
 	Package = "ydb-go-sdk"
 )

metrics/driver.go

@@ -95,10 +95,12 @@ func driver(config Config) (t trace.Driver) {
 		eventType := repeater.EventType(*info.Context)
 
 		return func(info trace.DriverBalancerClusterDiscoveryAttemptDoneInfo) {
-			balancersDiscoveries.With(map[string]string{
-				"status": errorBrief(info.Error),
-				"cause":  eventType,
-			}).Inc()
+			if config.Details()&trace.DriverBalancerEvents != 0 {
+				balancersDiscoveries.With(map[string]string{
+					"status": errorBrief(info.Error),
+					"cause":  eventType,
+				}).Inc()
+			}
 		}
 	}
 	t.OnBalancerUpdate = func(info trace.DriverBalancerUpdateStartInfo) func(trace.DriverBalancerUpdateDoneInfo) {

metrics/table.go

@@ -22,7 +22,9 @@ func table(config Config) (t trace.Table) {
 	with := config.GaugeVec("with")
 	t.OnInit = func(info trace.TableInitStartInfo) func(trace.TableInitDoneInfo) {
 		return func(info trace.TableInitDoneInfo) {
-			limit.With(nil).Set(float64(info.Limit))
+			if config.Details()&trace.TableEvents != 0 {
+				limit.With(nil).Set(float64(info.Limit))
+			}
 		}
 	}
 	t.OnSessionNew = func(info trace.TableSessionNewStartInfo) func(trace.TableSessionNewDoneInfo) {
@@ -69,11 +71,13 @@ func table(config Config) (t trace.Table) {
 		return nil
 	}
 	t.OnPoolStateChange = func(info trace.TablePoolStateChangeInfo) {
-		limit.With(nil).Set(float64(info.Limit))
-		index.With(nil).Set(float64(info.Index))
-		idle.With(nil).Set(float64(info.Idle))
-		wait.With(nil).Set(float64(info.Wait))
-		createInProgress.With(nil).Set(float64(info.CreateInProgress))
+		if config.Details()&trace.TablePoolEvents != 0 {
+			limit.With(nil).Set(float64(info.Limit))
+			index.With(nil).Set(float64(info.Index))
+			idle.With(nil).Set(float64(info.Idle))
+			wait.With(nil).Set(float64(info.Wait))
+			createInProgress.With(nil).Set(float64(info.CreateInProgress))
+		}
 	}
 	{
 		latency := session.WithSystem("query").TimerVec("latency")

query/client.go

@@ -12,8 +12,6 @@ import (
 
 type (
 	// Executor is an interface for execute queries
-	//
-	// Experimental: https://github.com/ydb-platform/ydb-go-sdk/blob/master/VERSIONING.md#experimental
 	Executor interface {
 		// Exec execute query without result
 		//
@@ -40,8 +38,6 @@ type (
 		QueryRow(ctx context.Context, query string, opts ...options.Execute) (Row, error)
 	}
 	// Client defines API of query client
-	//
-	// Experimental: https://github.com/ydb-platform/ydb-go-sdk/blob/master/VERSIONING.md#experimental
 	Client interface {
 		Executor
 
@@ -85,15 +81,11 @@ type (
 		// QueryResultSet is a helper which read all rows from first result set in result
 		//
 		// Warning: the large result set from query will be materialized and can happened to "OOM killed" problem
-		//
-		// Experimental: https://github.com/ydb-platform/ydb-go-sdk/blob/master/VERSIONING.md#experimental
 		QueryResultSet(ctx context.Context, query string, opts ...options.Execute) (ClosableResultSet, error)
 
 		// QueryRow is a helper which read only one row from first result set in result
 		//
 		// ReadRow returns error if result contains more than one result set or more than one row
-		//
-		// Experimental: https://github.com/ydb-platform/ydb-go-sdk/blob/master/VERSIONING.md#experimental
 		QueryRow(ctx context.Context, query string, opts ...options.Execute) (Row, error)
 
 		// ExecuteScript starts long executing script with polling results later
@@ -157,8 +149,6 @@ func WithLabel(lbl string) options.RetryOptionsOption {
 }
 
 // WithRetryBudget creates option with external budget
-//
-// Experimental: https://github.com/ydb-platform/ydb-go-sdk/blob/master/VERSIONING.md#experimental
 func WithRetryBudget(b budget.Budget) options.RetryOptionsOption {
 	return options.WithRetryBudget(b)
 }

tests/slo/internal/workers/read.go

@@ -14,12 +14,24 @@ import (
 func (w *Workers) Read(ctx context.Context, wg *sync.WaitGroup, rl *rate.Limiter) {
 	defer wg.Done()
 	for {
-		err := rl.Wait(ctx)
-		if err != nil {
+		select {
+		case <-ctx.Done():
 			return
+		default:
+			err := rl.Wait(ctx)
+			if err != nil {
+				return
+			}
+
+			err = w.read(ctx)
+			if err != nil {
+				if ctxErr := ctx.Err(); ctxErr == nil {
+					log.Printf("read failed: %v", err)
+				}
+
+				return
+			}
 		}
-
-		_ = w.read(ctx)
 	}
 }
 
@@ -29,9 +41,6 @@ func (w *Workers) read(ctx context.Context) error {
 	m := w.m.Start(metrics.JobRead)
 
 	_, attempts, err := w.s.Read(ctx, id)
-	if err != nil {
-		log.Printf("read failed with %d attempts: %v", attempts, err)
-	}
 
 	m.Finish(err, attempts)
 

tests/slo/internal/workers/write.go

@@ -14,12 +14,22 @@ import (
 func (w *Workers) Write(ctx context.Context, wg *sync.WaitGroup, rl *rate.Limiter, gen *generator.Generator) {
 	defer wg.Done()
 	for {
-		err := rl.Wait(ctx)
-		if err != nil {
+		select {
+		case <-ctx.Done():
 			return
+		default:
+			err := rl.Wait(ctx)
+			if err != nil {
+				return
+			}
+
+			err = w.write(ctx, gen)
+			if err != nil {
+				if ctxErr := ctx.Err(); ctxErr == nil {
+					log.Printf("write failed: %v", err)
+				}
+			}
 		}
-
-		_ = w.write(ctx, gen)
 	}
 }
 
@@ -34,9 +44,6 @@ func (w *Workers) write(ctx context.Context, gen *generator.Generator) error {
 	m := w.m.Start(metrics.JobWrite)
 
 	attempts, err := w.s.Write(ctx, row)
-	if err != nil {
-		log.Printf("write failed with %d attempts: %v", attempts, err)
-	}
 
 	m.Finish(err, attempts)