Merge pull request #747 from ydb-platform/calcattemptsinslo

test(slo): calc attempts for requests

Author: asmyasnikov

.github/workflows/slo.yml

@@ -8,11 +8,9 @@ on:
   workflow_dispatch:
 jobs:
   native:
-    needs: [gorm]
     concurrency:
       group: slo-native-${{ github.ref }}
-    if: github.event.pull_request.head.repo.full_name == 'ydb-platform/ydb-go-sdk' &&
-      !contains(github.event.pull_request.labels.*.name, 'no slo')
+    if: (!contains(github.event.pull_request.labels.*.name, 'no slo'))
     uses: ydb-platform/slo-tests/.github/workflows/slo.yml@main
     secrets: inherit
     with:
@@ -25,8 +23,7 @@ jobs:
     needs: [native]
     concurrency:
       group: slo-database-sql-${{ github.ref }}
-    if: github.event.pull_request.head.repo.full_name == 'ydb-platform/ydb-go-sdk' &&
-      !contains(github.event.pull_request.labels.*.name, 'no slo')
+    if: always() && !contains(github.event.pull_request.labels.*.name, 'no slo')
     uses: ydb-platform/slo-tests/.github/workflows/slo.yml@main
     secrets: inherit
     with:
@@ -36,10 +33,10 @@ jobs:
       workload_build_options: -f Dockerfile --build-arg SRC_PATH=database/sql
       workload_build_context: ../..
   gorm:
+    needs: [database_sql]
     concurrency:
       group: slo-gorm-${{ github.ref }}
-    if: github.event.pull_request.head.repo.full_name == 'ydb-platform/ydb-go-sdk' &&
-      !contains(github.event.pull_request.labels.*.name, 'no slo')
+    if: always() && !contains(github.event.pull_request.labels.*.name, 'no slo')
     uses: ydb-platform/slo-tests/.github/workflows/slo.yml@main
     secrets: inherit
     with:

tests/slo/Dockerfile

@@ -2,7 +2,7 @@ FROM golang:1.20 as build
 ARG SRC_PATH
 COPY . /src
 WORKDIR /src/tests/slo/${SRC_PATH}
-RUN CGO_ENABLED=0 go build -o /build/slo-go-workload .
+RUN CGO_ENABLED=0 go build -o /build/slo-go-workload -ldflags "-X \"main.label=${SRC_PATH}\"" .
 
 FROM scratch
 COPY --from=build /build /

tests/slo/README.md

@@ -123,6 +123,7 @@ Primary key: `("hash", "id")`
 - `not_oks`  - amount of not OK requests
 - `inflight` - amount of requests in flight
 - `latency`  - summary of latencies in ms
+- `attempts` - summary of amount for request
 
 > You must reset metrics to keep them `0` in prometheus and grafana before beginning and after ending of jobs
 
@@ -140,6 +141,8 @@ func (m *Metrics) Reset() error {
 
     m.latencies.Reset()
 
+    m.attempts.Reset()
+
     return m.Push()
 }
 ```

tests/slo/database/sql/main.go

@@ -20,6 +20,8 @@ import (
 
 var logger *zap.Logger
 
+var label string
+
 func init() {
 	var err error
 	logger, err = zap.NewProduction(zap.AddStacktrace(zapcore.PanicLevel))
@@ -78,7 +80,7 @@ func main() {
 					return err
 				}
 
-				err = s.Write(ctx, e)
+				_, err = s.Write(ctx, e)
 				if err != nil {
 					return err
 				}
@@ -103,7 +105,7 @@ func main() {
 	case config.RunMode:
 		gen := generator.New(cfg.InitialDataCount)
 
-		w, err := workers.New(cfg, s, logger)
+		w, err := workers.New(cfg, s, logger, label)
 		if err != nil {
 			panic(fmt.Errorf("create workers failed: %w", err))
 		}

tests/slo/database/sql/storage.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"database/sql"
 	"fmt"
+	"path"
 	"time"
 
 	env "github.com/ydb-platform/ydb-go-sdk-auth-environ"
@@ -104,6 +105,7 @@ func NewStorage(ctx context.Context, cfg *config.Config, poolSize int) (s *Stora
 
 	s.c, err = ydb.Connector(s.cc,
 		ydb.WithAutoDeclare(),
+		ydb.WithTablePathPrefix(path.Join(s.cc.Name(), label)),
 	)
 	if err != nil {
 		return nil, fmt.Errorf("ydb.Connector error: %w", err)
@@ -118,45 +120,84 @@ func NewStorage(ctx context.Context, cfg *config.Config, poolSize int) (s *Stora
 	return s, nil
 }
 
-func (s *Storage) Read(ctx context.Context, entryID generator.RowID) (res generator.Row, err error) {
+func (s *Storage) Read(ctx context.Context, entryID generator.RowID) (res generator.Row, attempts int, err error) {
 	if err = ctx.Err(); err != nil {
-		return generator.Row{}, err
+		return generator.Row{}, attempts, err
 	}
 
 	ctx, cancel := context.WithTimeout(ctx, time.Duration(s.cfg.ReadTimeout)*time.Millisecond)
 	defer cancel()
 
 	err = retry.Do(ydb.WithTxControl(ctx, readTx), s.db,
 		func(ctx context.Context, cc *sql.Conn) (err error) {
+			if err = ctx.Err(); err != nil {
+				return err
+			}
+
 			row := cc.QueryRowContext(ydb.WithQueryMode(ctx, ydb.DataQueryMode), s.selectQuery,
 				sql.Named("id", &entryID),
 			)
+
 			var hash *uint64
+
 			return row.Scan(&res.ID, &res.PayloadStr, &res.PayloadDouble, &res.PayloadTimestamp, &hash)
-		}, retry.WithDoRetryOptions(retry.WithIdempotent(true)),
+		},
+		retry.WithDoRetryOptions(
+			retry.WithIdempotent(true),
+			retry.WithTrace(
+				trace.Retry{
+					OnRetry: func(info trace.RetryLoopStartInfo) func(trace.RetryLoopIntermediateInfo) func(trace.RetryLoopDoneInfo) {
+						return func(info trace.RetryLoopIntermediateInfo) func(trace.RetryLoopDoneInfo) {
+							return func(info trace.RetryLoopDoneInfo) {
+								attempts = info.Attempts
+							}
+						}
+					},
+				},
+			),
+		),
 	)
 
-	return res, err
+	return res, attempts, err
 }
 
-func (s *Storage) Write(ctx context.Context, e generator.Row) error {
-	if err := ctx.Err(); err != nil {
-		return err
+func (s *Storage) Write(ctx context.Context, e generator.Row) (attempts int, err error) {
+	if err = ctx.Err(); err != nil {
+		return attempts, err
 	}
 
 	ctx, cancel := context.WithTimeout(ctx, time.Duration(s.cfg.WriteTimeout)*time.Millisecond)
 	defer cancel()
 
-	return retry.Do(ydb.WithTxControl(ctx, writeTx), s.db,
-		func(ctx context.Context, cc *sql.Conn) error {
-			_, err := cc.ExecContext(ydb.WithQueryMode(ctx, ydb.DataQueryMode), s.upsertQuery,
+	return attempts, retry.Do(ydb.WithTxControl(ctx, writeTx), s.db,
+		func(ctx context.Context, cc *sql.Conn) (err error) {
+			if err = ctx.Err(); err != nil {
+				return err
+			}
+
+			_, err = cc.ExecContext(ydb.WithQueryMode(ctx, ydb.DataQueryMode), s.upsertQuery,
 				sql.Named("id", e.ID),
 				sql.Named("payload_str", *e.PayloadStr),
 				sql.Named("payload_double", *e.PayloadDouble),
 				sql.Named("payload_timestamp", *e.PayloadTimestamp),
 			)
+
 			return err
-		}, retry.WithDoRetryOptions(retry.WithIdempotent(true)),
+		},
+		retry.WithDoRetryOptions(
+			retry.WithIdempotent(true),
+			retry.WithTrace(
+				trace.Retry{
+					OnRetry: func(info trace.RetryLoopStartInfo) func(trace.RetryLoopIntermediateInfo) func(trace.RetryLoopDoneInfo) {
+						return func(info trace.RetryLoopIntermediateInfo) func(trace.RetryLoopDoneInfo) {
+							return func(info trace.RetryLoopDoneInfo) {
+								attempts = info.Attempts
+							}
+						}
+					},
+				},
+			),
+		),
 	)
 }
 

tests/slo/gorm/main.go

@@ -20,6 +20,8 @@ import (
 
 var logger *zap.Logger
 
+var label string
+
 func init() {
 	var err error
 	logger, err = zap.NewProduction(zap.AddStacktrace(zapcore.PanicLevel))
@@ -78,7 +80,7 @@ func main() {
 					return err
 				}
 
-				err = s.Write(ctx, e)
+				_, err = s.Write(ctx, e)
 				if err != nil {
 					return err
 				}
@@ -103,7 +105,7 @@ func main() {
 	case config.RunMode:
 		gen := generator.New(cfg.InitialDataCount)
 
-		w, err := workers.New(cfg, s, logger)
+		w, err := workers.New(cfg, s, logger, label)
 		if err != nil {
 			panic(fmt.Errorf("create workers failed: %w", err))
 		}