Skip to content

Commit

Permalink
partially resolve #61, #100 (#99)
Browse files Browse the repository at this point in the history 
- custom decoder for 'vshard.storage.call' response

Co-authored-by: Nurzhan Saktaganov <[email protected]>
  • Loading branch information
@KaymeKaydex @nurzhan-saktaganov
KaymeKaydex and nurzhan-saktaganov authored Dec 12, 2024
1 parent e025d0b commit 17d1ba2
Show file tree
Hide file tree
Showing 6 changed files with 208 additions and 97 deletions.
6 changes: 6 additions & 0 deletions CHANGELOG.md
View file
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,9 @@ CHANGES:
* Linter: don't capitalize error strings and capitalize log.
* Fix misspellings.
* Handle vshard error the same way as lua vshard router (resolve issue #77).
* Decode 'vshard.storage.call' response manually into struct vshardStorageCallResponseProto using DecodeMsgpack interface to reduce allocations (partially #61, #100).
* Remove `mapstructure` tag from StorageCallVShardError.
* Update benchmarks in README files.

FEATURES:

Expand All @@ -19,6 +22,9 @@ REFACTOR:

* Use constants for vshard error names and codes.

TESTS:
* Rename bootstrap_test.go -> tarantool_test.go and new test in this file.

## v1.2.0

CHANGES:
Expand Down
12 changes: 6 additions & 6 deletions README.md
View file
Original file line number Diff line number Diff line change
Expand Up @@ -191,12 +191,12 @@ Service with go-vshard-router on top of the tarantool example from the original
## Benchmarks
### Go Bench

| Benchmark | Runs | Time (ns/op) | Memory (B/op) | Allocations (allocs/op) |
|-------------------------------------|--------|---------------|----------------|--------------------------|
| BenchmarkCallSimpleInsert_GO-8 | 9844 | 114596 | 1894 | 41 |
| BenchmarkCallSimpleInsert_Lua-8 | 7587 | 156181 | 1101 | 19 |
| BenchmarkCallSimpleSelect_GO-8 | 16350 | 75770 | 2827 | 60 |
| BenchmarkCallSimpleSelect_Lua-8 | 10060 | 116768 | 1610 | 28 |
| Benchmark | Runs | Time (ns/op) | Memory (B/op) | Allocations (allocs/op) |
|---------------------------------------|--------|---------------|----------------|-------------------------|
| BenchmarkCallSimpleInsert_GO-12 | 14216 | 81118 | 1419 | 29 |
| BenchmarkCallSimpleInsert_Lua-12 | 9580 | 123307 | 1131 | 19 |
| BenchmarkCallSimpleSelect_GO-12 | 18832 | 65190 | 1879 | 38 |
| BenchmarkCallSimpleSelect_Lua-12 | 9963 | 104781 | 1617 | 28 |


### [K6](https://github.com/grafana/k6)
Expand Down
12 changes: 6 additions & 6 deletions README_ru.md
View file
Original file line number Diff line number Diff line change
Expand Up @@ -191,12 +191,12 @@ func main() {
## Бенчмарки
### Go Bench

| Бенчмарк | Запусков | Время (ns/op) | Память (B/op) | Аллокации (allocs/op) |
|----------------------------------------|----------|----------------|----------------|------------------------|
| BenchmarkCallSimpleInsert_GO-8 | 9844 | 114596 | 1894 | 41 |
| BenchmarkCallSimpleInsert_Lua-8 | 7587 | 156181 | 1101 | 19 |
| BenchmarkCallSimpleSelect_GO-8 | 16350 | 75770 | 2827 | 60 |
| BenchmarkCallSimpleSelect_Lua-8 | 10060 | 116768 | 1610 | 28 |
| Бенчмарк | Число запусков | Время (ns/op) | Память (B/op) | Аллокации (allocs/op) |
|----------------------------------|----------------|---------------|---------------|-----------------------|
| BenchmarkCallSimpleInsert_GO-12 | 14216 | 81118 | 1419 | 29 |
| BenchmarkCallSimpleInsert_Lua-12 | 9580 | 123307 | 1131 | 19 |
| BenchmarkCallSimpleSelect_GO-12 | 18832 | 65190 | 1879 | 38 |
| BenchmarkCallSimpleSelect_Lua-12 | 9963 | 104781 | 1617 | 28 |


### [K6](https://github.com/grafana/k6)
Expand Down
192 changes: 130 additions & 62 deletions api.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,8 @@ import (

"github.com/tarantool/go-tarantool/v2"
"github.com/tarantool/go-tarantool/v2/pool"
"github.com/vmihailenco/msgpack/v5"
"github.com/vmihailenco/msgpack/v5/msgpcode"
)

// --------------------------------------------------------------------------------
Expand All @@ -27,32 +29,135 @@ func (c VshardMode) String() string {
return string(c)
}

type storageCallAssertError struct {
type vshardStorageCallResponseProto struct {
AssertError *assertError // not nil if there is assert error
VshardError *StorageCallVShardError // not nil if there is vshard response
Data []interface{} // raw response data
}

func (r *vshardStorageCallResponseProto) DecodeMsgpack(d *msgpack.Decoder) error {
/* vshard.storage.call(func) response has the next 4 possbile formats:
See: https://github.com/tarantool/vshard/blob/dfa2cc8a2aff221d5f421298851a9a229b2e0434/vshard/storage/init.lua#L3130
1. vshard error has occurred:
array[nil, vshard_error]
2. User method has finished with some error:
array[false, assert error]
3. User mehod has finished successfully
a) but has not returned anything
array[true]
b) has returned 1 element
array[true, elem1]
c) has returned 2 element
array[true, elem1, elem2]
d) has returned 3 element
array[true, elem1, elem2, elem3]
*/

// Ensure it is an array and get array len for protocol violation check
respArrayLen, err := d.DecodeArrayLen()
if err != nil {
return err
}

if respArrayLen == 0 {
return fmt.Errorf("protocol violation: invalid array length: %d", respArrayLen)
}

// we need peek code to make our check faster than decode interface
// later we will check if code nil or bool
code, err := d.PeekCode()
if err != nil {
return err
}

// this is storage error
if code == msgpcode.Nil {
err = d.DecodeNil()
if err != nil {
return err
}

if respArrayLen != 2 {
return fmt.Errorf("protocol violation: length is %d on vshard error case", respArrayLen)
}

var vshardError StorageCallVShardError

err = d.Decode(&vshardError)
if err != nil {
return fmt.Errorf("failed to decode storage vshard error: %w", err)
}

r.VshardError = &vshardError

return nil
}

isVShardRespOk, err := d.DecodeBool()
if err != nil {
return err
}

if !isVShardRespOk {
// that means we have an assert errors and response is not ok
if respArrayLen != 2 {
return fmt.Errorf("protocol violation: length is %d on assert error case", respArrayLen)
}

var assertError assertError
err = d.Decode(&assertError)
if err != nil {
return fmt.Errorf("failed to decode storage assert error: %w", err)
}

r.AssertError = &assertError

return nil
}

// isVShardRespOk is true
r.Data = make([]interface{}, 0, respArrayLen-1)

for i := 1; i < respArrayLen; i++ {
elem, err := d.DecodeInterface()
if err != nil {
return fmt.Errorf("failed to decode into interface element #%d of response array", i+1)
}
r.Data = append(r.Data, elem)
}

return nil
}

type assertError struct {
Code int `msgpack:"code"`
BaseType string `msgpack:"base_type"`
Type string `msgpack:"type"`
Message string `msgpack:"message"`
Trace interface{} `msgpack:"trace"`
}

func (s storageCallAssertError) Error() string {
type alias storageCallAssertError
func (s assertError) Error() string {
// Just print struct as is, use hack with alias type to avoid recursion:
// %v attempts to call Error() method for s, which is recursion.
// This alias doesn't have method Error().
type alias assertError
return fmt.Sprintf("%+v", alias(s))
}

type StorageCallVShardError struct {
BucketID uint64 `msgpack:"bucket_id" mapstructure:"bucket_id"`
BucketID uint64 `msgpack:"bucket_id"`
Reason string `msgpack:"reason"`
Code int `msgpack:"code"`
Type string `msgpack:"type"`
Message string `msgpack:"message"`
Name string `msgpack:"name"`
// These 3 fields below are send as string by vshard storage, so we decode them into string, not uuid.UUID type
// Example: 00000000-0000-0002-0002-000000000000
MasterUUID string `msgpack:"master" mapstructure:"master"`
ReplicasetUUID string `msgpack:"replicaset" mapstructure:"replicaset"`
ReplicaUUID string `msgpack:"replica" mapstructure:"replica"`
Destination string `msgpack:"destination" mapstructure:"destination"`
MasterUUID string `msgpack:"master"`
ReplicasetUUID string `msgpack:"replicaset"`
ReplicaUUID string `msgpack:"replica"`
Destination string `msgpack:"destination"`
}

func (s StorageCallVShardError) Error() string {
Expand Down Expand Up @@ -108,7 +213,6 @@ func (r *Router) RouterCallImpl(ctx context.Context,
})

var err error
var vshardError StorageCallVShardError

for {
if since := time.Since(timeStart); since > timeout {
Expand Down Expand Up @@ -141,32 +245,20 @@ func (r *Router) RouterCallImpl(ctx context.Context,

future := rs.conn.Do(req, opts.PoolMode)

var respData []interface{}
respData, err = future.Get()
var storageCallResponse vshardStorageCallResponseProto
err = future.GetTyped(&storageCallResponse)
if err != nil {
return nil, nil, fmt.Errorf("got error on future.Get(): %w", err)
}

r.log().Debugf(ctx, "Got call result response data %v", respData)
r.log().Debugf(ctx, "Got call result response data %+v", storageCallResponse)

if len(respData) == 0 {
// vshard.storage.call(func) returns up to two values:
// - true/false/nil
// - func result, omitted if func does not return anything
return nil, nil, fmt.Errorf("protocol violation %s: got empty response", vshardStorageClientCall)
if storageCallResponse.AssertError != nil {
return nil, nil, fmt.Errorf("%s: %s failed: %+v", vshardStorageClientCall, fnc, storageCallResponse.AssertError)
}

if respData[0] == nil {
if len(respData) != 2 {
return nil, nil, fmt.Errorf("protocol violation %s: length is %d when respData[0] is nil", vshardStorageClientCall, len(respData))
}

err = mapstructure.Decode(respData[1], &vshardError)
if err != nil {
// Something unexpected happened: we couldn't decode respData[1] as a vshardError,
// so return reason why and respData[1], that is supposed to be a vshardError.
return nil, nil, fmt.Errorf("cant decode vhsard err by trarantool with err: %v (%v)", err, respData[1])
}
if storageCallResponse.VshardError != nil {
vshardError := storageCallResponse.VshardError

switch vshardError.Name {
case VShardErrNameWrongBucket, VShardErrNameBucketIsLocked:
Expand Down Expand Up @@ -201,7 +293,7 @@ func (r *Router) RouterCallImpl(ctx context.Context,
time.Sleep(defaultPoolingPause)

if time.Since(timeStart) > timeout {
return nil, nil, &vshardError
return nil, nil, vshardError
}
}
}
Expand All @@ -210,60 +302,36 @@ func (r *Router) RouterCallImpl(ctx context.Context,

r.metrics().RetryOnCall("bucket_migrate")

r.log().Debugf(ctx, "Retrying fnc '%s' cause got vshard error: %v", fnc, &vshardError)
r.log().Debugf(ctx, "Retrying fnc '%s' cause got vshard error: %v", fnc, vshardError)

// this vshardError will be returned to a caller in case of timeout
err = &vshardError
err = vshardError
continue
case VShardErrNameTransferIsInProgress:
// Since lua vshard router doesn't retry here, we don't retry too.
// There is a comment why lua vshard router doesn't retry:
// https://github.com/tarantool/vshard/blob/b6fdbe950a2e4557f05b83bd8b846b126ec3724e/vshard/router/init.lua#L697
r.BucketReset(bucketID)
return nil, nil, &vshardError
return nil, nil, vshardError
case VShardErrNameNonMaster:
// vshard.storage has returned NON_MASTER error, lua vshard router updates info about master in this case:
// See: https://github.com/tarantool/vshard/blob/b6fdbe950a2e4557f05b83bd8b846b126ec3724e/vshard/router/init.lua#L704.
// Since we use go-tarantool library, and go-tarantool library doesn't provide API to update info about current master,
// we just return this error as is.
return nil, nil, &vshardError
return nil, nil, vshardError
default:
return nil, nil, &vshardError
return nil, nil, vshardError
}
}

var isVShardRespOk bool
err = future.GetTyped(&[]interface{}{&isVShardRespOk})
if err != nil {
return nil, nil, fmt.Errorf("protocol violation %s: can't decode respData[0] as boolean: %v", vshardStorageClientCall, err)
}

if !isVShardRespOk {
// Since we got respData[0] == false, it means that an error has happened
// while executing user-defined function on vshard storage.
// In this case, vshard storage must return a pair: false, error.
if len(respData) != 2 {
return nil, nil, fmt.Errorf("protocol violation %s: response length is %d when respData[0] is false", vshardStorageClientCall, len(respData))
}

var assertError storageCallAssertError
err = mapstructure.Decode(respData[1], &assertError)
if err != nil {
// We could not decode respData[1] as assertError, so return respData[1] as is, add info why we could not decode.
return nil, nil, fmt.Errorf("%s: %s failed %v (decoding to assertError failed %v)", vshardStorageClientCall, fnc, respData[1], err)
}

return nil, nil, fmt.Errorf("%s: %s failed: %+v", vshardStorageClientCall, fnc, assertError)
}

r.metrics().RequestDuration(time.Since(timeStart), true, false)

return respData[1:], func(result interface{}) error {
if len(respData) < 2 {
return storageCallResponse.Data, func(result interface{}) error {
if len(storageCallResponse.Data) == 0 {
return nil
}

var stub interface{}
var stub bool

return future.GetTyped(&[]interface{}{&stub, result})
}, nil
Expand Down Expand Up @@ -399,7 +467,7 @@ func (r *Router) RouterMapCallRWImpl(
return nil, fmt.Errorf("protocol violation: invalid respData length when respData[0] == nil, must be = 2, current: %d", len(respData))
}

var assertError storageCallAssertError
var assertError assertError
err = mapstructure.Decode(respData[1], &assertError)
if err != nil {
// We could not decode respData[1] as assertError, so return respData[1] as is, add info why we could not decode.
Expand Down
Loading

0 comments on commit 17d1ba2

Please sign in to comment.