forked from gomods/athens
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
pkg/stash: implement a Redis lock for stashing (gomods#1116)
* pkg/stash: implement a Redis lock for stashing * fix tests * fix op
- Loading branch information
1 parent
c3d8a05
commit fb9437d
Showing
59 changed files
with
11,929 additions
and
3 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,73 @@ | ||
| package stash | ||
|
|
||
| import ( | ||
| "context" | ||
| "time" | ||
|
|
||
| lock "github.com/bsm/redis-lock" | ||
| "github.com/go-redis/redis" | ||
| "github.com/gomods/athens/pkg/config" | ||
| "github.com/gomods/athens/pkg/errors" | ||
| "github.com/gomods/athens/pkg/observ" | ||
| "github.com/gomods/athens/pkg/storage" | ||
| ) | ||
|
|
||
| // WithRedisLock returns a distributed singleflight | ||
| // using an redis cluster. If it cannot connect, it will return an error. | ||
| func WithRedisLock(endpoint string, checker storage.Checker) (Wrapper, error) { | ||
| const op errors.Op = "stash.WithRedisLock" | ||
| client := redis.NewClient(&redis.Options{ | ||
| Network: "tcp", | ||
| Addr: endpoint, | ||
| }) | ||
| _, err := client.Ping().Result() | ||
| if err != nil { | ||
| return nil, errors.E(op, err) | ||
| } | ||
|
|
||
| return func(s Stasher) Stasher { | ||
| return &redisLock{client, s, checker} | ||
| }, nil | ||
| } | ||
|
|
||
| type redisLock struct { | ||
| client *redis.Client | ||
| stasher Stasher | ||
| checker storage.Checker | ||
| } | ||
|
|
||
| func (s *redisLock) Stash(ctx context.Context, mod, ver string) (newVer string, err error) { | ||
| const op errors.Op = "redis.Stash" | ||
| ctx, span := observ.StartSpan(ctx, op.String()) | ||
| defer span.End() | ||
| mv := config.FmtModVer(mod, ver) | ||
|
|
||
| // Obtain a new lock with default settings | ||
| lock, err := lock.Obtain(s.client, mv, &lock.Options{ | ||
| LockTimeout: time.Minute * 5, | ||
| RetryCount: 60 * 5, | ||
| RetryDelay: time.Second, | ||
| }) | ||
| if err != nil { | ||
| return ver, errors.E(op, err) | ||
| } | ||
| defer func() { | ||
| const op errors.Op = "redis.Unlock" | ||
| lockErr := lock.Unlock() | ||
| if err == nil && lockErr != nil { | ||
| err = errors.E(op, lockErr) | ||
| } | ||
| }() | ||
| ok, err := s.checker.Exists(ctx, mod, ver) | ||
| if err != nil { | ||
| return ver, errors.E(op, err) | ||
| } | ||
| if ok { | ||
| return ver, nil | ||
| } | ||
| newVer, err = s.stasher.Stash(ctx, mod, ver) | ||
| if err != nil { | ||
| return ver, errors.E(op, err) | ||
| } | ||
| return newVer, nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,82 @@ | ||
| package stash | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "os" | ||
| "strings" | ||
| "sync" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/gomods/athens/pkg/storage" | ||
| "github.com/gomods/athens/pkg/storage/mem" | ||
| "golang.org/x/sync/errgroup" | ||
| ) | ||
|
|
||
| // WithRedisLock will ensure that 5 concurrent requests will all get the first request's | ||
| // response. We can ensure that because only the first response does not return an error | ||
| // and therefore all 5 responses should have no error. | ||
| func TestWithRedisLock(t *testing.T) { | ||
| endpoint := os.Getenv("REDIS_TEST_ENDPOINT") | ||
| if len(endpoint) == 0 { | ||
| t.SkipNow() | ||
| } | ||
| strg, err := mem.NewStorage() | ||
| if err != nil { | ||
| t.Fatal(err) | ||
| } | ||
| ms := &mockRedisStasher{strg: strg} | ||
| wrapper, err := WithRedisLock(endpoint, strg) | ||
| if err != nil { | ||
| t.Fatal(err) | ||
| } | ||
| s := wrapper(ms) | ||
|
|
||
| var eg errgroup.Group | ||
| for i := 0; i < 5; i++ { | ||
| eg.Go(func() error { | ||
| ctx, cancel := context.WithTimeout(context.Background(), time.Second*10) | ||
| defer cancel() | ||
| _, err := s.Stash(ctx, "mod", "ver") | ||
| return err | ||
| }) | ||
| } | ||
|
|
||
| err = eg.Wait() | ||
| if err != nil { | ||
| t.Fatal(err) | ||
| } | ||
| } | ||
|
|
||
| // mockRedisStasher is like mockStasher | ||
| // but leverages in memory storage | ||
| // so that redis can determine | ||
| // whether to call the underlying stasher or not. | ||
| type mockRedisStasher struct { | ||
| strg storage.Backend | ||
| mu sync.Mutex | ||
| num int | ||
| } | ||
|
|
||
| func (ms *mockRedisStasher) Stash(ctx context.Context, mod, ver string) (string, error) { | ||
| time.Sleep(time.Millisecond * 100) // allow for second requests to come in. | ||
| ms.mu.Lock() | ||
| defer ms.mu.Unlock() | ||
| if ms.num == 0 { | ||
| err := ms.strg.Save( | ||
| ctx, | ||
| mod, | ||
| ver, | ||
| []byte("mod file"), | ||
| strings.NewReader("zip file"), | ||
| []byte("info file"), | ||
| ) | ||
| if err != nil { | ||
| return "", err | ||
| } | ||
| ms.num++ | ||
| return "", nil | ||
| } | ||
| return "", fmt.Errorf("second time error") | ||
| } |
Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.
Oops, something went wrong.
Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.
Oops, something went wrong.
Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.
Oops, something went wrong.
Oops, something went wrong.