Skip to content

Commit c2f3d9e

Browse files
ImpressionableRaccoonImpressionableRaccoon
committed
test(slo): calc attempts for requests
1 parent de39d21 commit c2f3d9e

File tree

13 files changed

+272
-80
lines changed

13 files changed

+272
-80
lines changed

‎.github/workflows/slo.yml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,6 @@ on:
88
workflow_dispatch:
99
jobs:
1010
native:
11-
needs: [gorm]
1211
concurrency:
1312
group: slo-native-${{ github.ref }}
1413
if: github.event.pull_request.head.repo.full_name == 'ydb-platform/ydb-go-sdk' &&
@@ -36,6 +35,7 @@ jobs:
3635
workload_build_options: -f Dockerfile --build-arg SRC_PATH=database/sql
3736
workload_build_context: ../..
3837
gorm:
38+
needs: [database_sql]
3939
concurrency:
4040
group: slo-gorm-${{ github.ref }}
4141
if: github.event.pull_request.head.repo.full_name == 'ydb-platform/ydb-go-sdk' &&

‎tests/slo/Dockerfile‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@ FROM golang:1.20 as build
22
ARG SRC_PATH
33
COPY . /src
44
WORKDIR /src/tests/slo/${SRC_PATH}
5-
RUN CGO_ENABLED=0 go build -o /build/slo-go-workload .
5+
RUN CGO_ENABLED=0 go build -o /build/slo-go-workload -ldflags "-X \"main.label=${SRC_PATH}\"" .
66

77
FROM scratch
88
COPY --from=build /build /

‎tests/slo/database/sql/main.go‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,8 @@ import (
2020

2121
var logger *zap.Logger
2222

23+
var label string
24+
2325
func init() {
2426
var err error
2527
logger, err = zap.NewProduction(zap.AddStacktrace(zapcore.PanicLevel))
@@ -78,7 +80,7 @@ func main() {
7880
return err
7981
}
8082

81-
err = s.Write(ctx, e)
83+
_, err = s.Write(ctx, e)
8284
if err != nil {
8385
return err
8486
}
@@ -103,7 +105,7 @@ func main() {
103105
case config.RunMode:
104106
gen := generator.New(cfg.InitialDataCount)
105107

106-
w, err := workers.New(cfg, s, logger)
108+
w, err := workers.New(cfg, s, logger, label)
107109
if err != nil {
108110
panic(fmt.Errorf("create workers failed: %w", err))
109111
}

‎tests/slo/database/sql/storage.go‎

Lines changed: 50 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -118,45 +118,84 @@ func NewStorage(ctx context.Context, cfg *config.Config, poolSize int) (s *Stora
118118
return s, nil
119119
}
120120

121-
func (s *Storage) Read(ctx context.Context, entryID generator.RowID) (res generator.Row, err error) {
121+
func (s *Storage) Read(ctx context.Context, entryID generator.RowID) (res generator.Row, attempts int, err error) {
122122
if err = ctx.Err(); err != nil {
123-
return generator.Row{}, err
123+
return generator.Row{}, attempts, err
124124
}
125125

126126
ctx, cancel := context.WithTimeout(ctx, time.Duration(s.cfg.ReadTimeout)*time.Millisecond)
127127
defer cancel()
128128

129129
err = retry.Do(ydb.WithTxControl(ctx, readTx), s.db,
130130
func(ctx context.Context, cc *sql.Conn) (err error) {
131+
if err = ctx.Err(); err != nil {
132+
return err
133+
}
134+
131135
row := cc.QueryRowContext(ydb.WithQueryMode(ctx, ydb.DataQueryMode), s.selectQuery,
132136
sql.Named("id", &entryID),
133137
)
138+
134139
var hash *uint64
140+
135141
return row.Scan(&res.ID, &res.PayloadStr, &res.PayloadDouble, &res.PayloadTimestamp, &hash)
136-
}, retry.WithDoRetryOptions(retry.WithIdempotent(true)),
142+
},
143+
retry.WithDoRetryOptions(
144+
retry.WithIdempotent(true),
145+
retry.WithTrace(
146+
trace.Retry{
147+
OnRetry: func(info trace.RetryLoopStartInfo) func(trace.RetryLoopIntermediateInfo) func(trace.RetryLoopDoneInfo) {
148+
return func(info trace.RetryLoopIntermediateInfo) func(trace.RetryLoopDoneInfo) {
149+
return func(info trace.RetryLoopDoneInfo) {
150+
attempts = info.Attempts
151+
}
152+
}
153+
},
154+
},
155+
),
156+
),
137157
)
138158

139-
return res, err
159+
return res, attempts, err
140160
}
141161

142-
func (s *Storage) Write(ctx context.Context, e generator.Row) error {
143-
if err := ctx.Err(); err != nil {
144-
return err
162+
func (s *Storage) Write(ctx context.Context, e generator.Row) (attempts int, err error) {
163+
if err = ctx.Err(); err != nil {
164+
return attempts, err
145165
}
146166

147167
ctx, cancel := context.WithTimeout(ctx, time.Duration(s.cfg.WriteTimeout)*time.Millisecond)
148168
defer cancel()
149169

150-
return retry.Do(ydb.WithTxControl(ctx, writeTx), s.db,
151-
func(ctx context.Context, cc *sql.Conn) error {
152-
_, err := cc.ExecContext(ydb.WithQueryMode(ctx, ydb.DataQueryMode), s.upsertQuery,
170+
return attempts, retry.Do(ydb.WithTxControl(ctx, writeTx), s.db,
171+
func(ctx context.Context, cc *sql.Conn) (err error) {
172+
if err = ctx.Err(); err != nil {
173+
return err
174+
}
175+
176+
_, err = cc.ExecContext(ydb.WithQueryMode(ctx, ydb.DataQueryMode), s.upsertQuery,
153177
sql.Named("id", e.ID),
154178
sql.Named("payload_str", *e.PayloadStr),
155179
sql.Named("payload_double", *e.PayloadDouble),
156180
sql.Named("payload_timestamp", *e.PayloadTimestamp),
157181
)
182+
158183
return err
159-
}, retry.WithDoRetryOptions(retry.WithIdempotent(true)),
184+
},
185+
retry.WithDoRetryOptions(
186+
retry.WithIdempotent(true),
187+
retry.WithTrace(
188+
trace.Retry{
189+
OnRetry: func(info trace.RetryLoopStartInfo) func(trace.RetryLoopIntermediateInfo) func(trace.RetryLoopDoneInfo) {
190+
return func(info trace.RetryLoopIntermediateInfo) func(trace.RetryLoopDoneInfo) {
191+
return func(info trace.RetryLoopDoneInfo) {
192+
attempts = info.Attempts
193+
}
194+
}
195+
},
196+
},
197+
),
198+
),
160199
)
161200
}
162201

‎tests/slo/gorm/main.go‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,8 @@ import (
2020

2121
var logger *zap.Logger
2222

23+
var label string
24+
2325
func init() {
2426
var err error
2527
logger, err = zap.NewProduction(zap.AddStacktrace(zapcore.PanicLevel))
@@ -78,7 +80,7 @@ func main() {
7880
return err
7981
}
8082

81-
err = s.Write(ctx, e)
83+
_, err = s.Write(ctx, e)
8284
if err != nil {
8385
return err
8486
}
@@ -103,7 +105,7 @@ func main() {
103105
case config.RunMode:
104106
gen := generator.New(cfg.InitialDataCount)
105107

106-
w, err := workers.New(cfg, s, logger)
108+
w, err := workers.New(cfg, s, logger, label)
107109
if err != nil {
108110
panic(fmt.Errorf("create workers failed: %w", err))
109111
}

‎tests/slo/gorm/storage.go‎

Lines changed: 99 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -2,13 +2,16 @@ package main
22

33
import (
44
"context"
5+
"database/sql"
56
"fmt"
67
"time"
78

89
ydb "github.com/ydb-platform/gorm-driver"
910
environ "github.com/ydb-platform/ydb-go-sdk-auth-environ"
1011
ydbZap "github.com/ydb-platform/ydb-go-sdk-zap"
1112
ydbSDK "github.com/ydb-platform/ydb-go-sdk/v3"
13+
"github.com/ydb-platform/ydb-go-sdk/v3/retry"
14+
"github.com/ydb-platform/ydb-go-sdk/v3/table"
1215
"github.com/ydb-platform/ydb-go-sdk/v3/trace"
1316
"gorm.io/gorm"
1417
"gorm.io/gorm/clause"
@@ -28,6 +31,19 @@ WITH (
2831
UNIFORM_PARTITIONS = %d
2932
)`
3033

34+
var (
35+
readTx = table.TxControl(
36+
table.BeginTx(
37+
table.WithOnlineReadOnly(),
38+
),
39+
table.CommitTx(),
40+
)
41+
42+
writeTx = table.SerializableReadWriteTxControl(
43+
table.CommitTx(),
44+
)
45+
)
46+
3147
type entry struct {
3248
Hash uint64 `gorm:"column:hash;primarykey;autoIncrement:false"`
3349

@@ -73,53 +89,108 @@ func NewStorage(ctx context.Context, cfg *config.Config, poolSize int) (*Storage
7389
return nil, err
7490
}
7591

76-
s.db = s.db.Debug()
77-
7892
return s, nil
7993
}
8094

81-
func (s *Storage) Read(ctx context.Context, id generator.RowID) (generator.Row, error) {
82-
if err := ctx.Err(); err != nil {
83-
return generator.Row{}, err
95+
func (s *Storage) Read(ctx context.Context, id generator.RowID) (r generator.Row, attempts int, err error) {
96+
if err = ctx.Err(); err != nil {
97+
return generator.Row{}, attempts, err
8498
}
8599

86100
ctx, cancel := context.WithTimeout(ctx, time.Duration(s.cfg.ReadTimeout)*time.Millisecond)
87101
defer cancel()
88102

89-
var e entry
90-
err := s.db.WithContext(ctx).Scopes(addTableToScope(s.cfg.Table)).Model(&entry{}).
91-
First(&e, "hash = ? AND id = ?",
92-
clause.Expr{
93-
SQL: "Digest::NumericHash(?)",
94-
Vars: []interface{}{id},
95-
},
96-
id,
97-
).Error
103+
db, err := s.db.DB()
98104
if err != nil {
99-
return generator.Row{}, err
105+
return generator.Row{}, attempts, err
100106
}
101107

102-
return e.Row, err
108+
return r, attempts, retry.Do(ydbSDK.WithTxControl(ctx, readTx), db,
109+
func(ctx context.Context, cc *sql.Conn) (err error) {
110+
if err = ctx.Err(); err != nil {
111+
return err
112+
}
113+
114+
var e entry
115+
err = s.db.WithContext(ctx).Scopes(addTableToScope(s.cfg.Table)).Model(&entry{}).
116+
First(&e, "hash = ? AND id = ?",
117+
clause.Expr{
118+
SQL: "Digest::NumericHash(?)",
119+
Vars: []interface{}{id},
120+
},
121+
id,
122+
).Error
123+
if err != nil {
124+
return err
125+
}
126+
127+
r = e.Row
128+
129+
return nil
130+
},
131+
retry.WithDoRetryOptions(
132+
retry.WithIdempotent(true),
133+
retry.WithTrace(
134+
trace.Retry{
135+
OnRetry: func(info trace.RetryLoopStartInfo) func(trace.RetryLoopIntermediateInfo) func(trace.RetryLoopDoneInfo) {
136+
return func(info trace.RetryLoopIntermediateInfo) func(trace.RetryLoopDoneInfo) {
137+
return func(info trace.RetryLoopDoneInfo) {
138+
attempts = info.Attempts
139+
}
140+
}
141+
},
142+
},
143+
),
144+
),
145+
)
103146
}
104147

105-
func (s *Storage) Write(ctx context.Context, row generator.Row) error {
106-
if err := ctx.Err(); err != nil {
107-
return err
148+
func (s *Storage) Write(ctx context.Context, row generator.Row) (attempts int, err error) {
149+
if err = ctx.Err(); err != nil {
150+
return attempts, err
108151
}
109152

110153
ctx, cancel := context.WithTimeout(ctx, time.Duration(s.cfg.WriteTimeout)*time.Millisecond)
111154
defer cancel()
112155

113-
return s.db.WithContext(ctx).Scopes(addTableToScope(s.cfg.Table)).Model(&entry{}).Create(map[string]interface{}{
114-
"Hash": clause.Expr{
115-
SQL: "Digest::NumericHash(?)",
116-
Vars: []interface{}{row.ID},
156+
db, err := s.db.DB()
157+
if err != nil {
158+
return attempts, err
159+
}
160+
161+
return attempts, retry.Do(ydbSDK.WithTxControl(ctx, writeTx), db,
162+
func(ctx context.Context, cc *sql.Conn) (err error) {
163+
if err = ctx.Err(); err != nil {
164+
return err
165+
}
166+
167+
return s.db.WithContext(ctx).Scopes(addTableToScope(s.cfg.Table)).Model(&entry{}).
168+
Create(map[string]interface{}{
169+
"Hash": clause.Expr{
170+
SQL: "Digest::NumericHash(?)",
171+
Vars: []interface{}{row.ID},
172+
},
173+
"ID": row.ID,
174+
"PayloadStr": row.PayloadStr,
175+
"PayloadDouble": row.PayloadDouble,
176+
"PayloadTimestamp": row.PayloadTimestamp,
177+
}).Error
117178
},
118-
"ID": row.ID,
119-
"PayloadStr": row.PayloadStr,
120-
"PayloadDouble": row.PayloadDouble,
121-
"PayloadTimestamp": row.PayloadTimestamp,
122-
}).Error
179+
retry.WithDoRetryOptions(
180+
retry.WithIdempotent(true),
181+
retry.WithTrace(
182+
trace.Retry{
183+
OnRetry: func(info trace.RetryLoopStartInfo) func(trace.RetryLoopIntermediateInfo) func(trace.RetryLoopDoneInfo) {
184+
return func(info trace.RetryLoopIntermediateInfo) func(trace.RetryLoopDoneInfo) {
185+
return func(info trace.RetryLoopDoneInfo) {
186+
attempts = info.Attempts
187+
}
188+
}
189+
},
190+
},
191+
),
192+
),
193+
)
123194
}
124195

125196
func (s *Storage) createTable(ctx context.Context) error {

0 commit comments

Comments
 (0)