mirror of
https://github.com/project-zot/zot.git
synced 2026-06-18 05:28:07 +08:00
Andrei Aaron
05823cd74f
* feat: add redis cache support https://github.com/project-zot/zot/pull/2005 Fixes https://github.com/project-zot/zot/issues/2004 * feat: add redis cache support Currently, we have dynamoDB as the remote shared cache but ideal only for the cloud use case. For on-prem use case, add support for redis. Signed-off-by: Ramkumar Chinchani <rchincha@cisco.com> * feat(redis): added blackbox tests for redis Signed-off-by: Petu Eusebiu <peusebiu@cisco.com> * feat(redis): dummy implementation of MetaDB interface for redis cache Signed-off-by: Alexei Dodon <adodon@cisco.com> * feat: check validity of driver configuration on metadb instantiation Signed-off-by: Andrei Aaron <aaaron@luxoft.com> * feat: multiple fixes for redis cache driver implementation - add missing method GetAllBlobs - add redis cache tests, with and without mocking Signed-off-by: Andrei Aaron <aaaron@luxoft.com> * feat(redis): redis implementation for MetaDB Signed-off-by: Andrei Aaron <aaaron@luxoft.com> * feat(redis): use redsync to block concurrent write access to the redis DB Signed-off-by: Andrei Aaron <aaaron@luxoft.com> * feat(redis): update .github/workflows/cluster.yaml to also test redis Signed-off-by: Andrei Aaron <aaaron@luxoft.com> * feat(metadb): add keyPrefix parameter for redis and remove unneeded method meta.Crate() Signed-off-by: Andrei Aaron <aaaron@luxoft.com> * feat(redis): support RedisCluster configuration and add unit tests Signed-off-by: Andrei Aaron <aaaron@luxoft.com> * feat(redis): more tests for redis metadb implementation Signed-off-by: Andrei Aaron <aaaron@luxoft.com> * feat(redis): add more examples and update examples/README.md Signed-off-by: Andrei Aaron <aaaron@luxoft.com> * feat(redis): move option parsing and redis client initialization under pkg/api/config/redis Signed-off-by: Andrei Aaron <aaaron@luxoft.com> * chore(cachedb): move Cache interface to pkg/storage/types Signed-off-by: Andrei Aaron <aaaron@luxoft.com> * feat(redis): reorganize code in pkg/storage/cache.go Signed-off-by: Andrei Aaron <aaaron@luxoft.com> * feat(redis): call redis.SetLogger() with the zot logger as parameter Signed-off-by: Andrei Aaron <aaaron@luxoft.com> * feat(redis): rename pkg/meta/redisdb to pkg/meta/redis Signed-off-by: Andrei Aaron <aaaron@luxoft.com> --------- Signed-off-by: Ramkumar Chinchani <rchincha@cisco.com> Signed-off-by: Petu Eusebiu <peusebiu@cisco.com> Signed-off-by: Alexei Dodon <adodon@cisco.com> Signed-off-by: Andrei Aaron <aaaron@luxoft.com> Co-authored-by: a <a@tuxpa.in> Co-authored-by: Ramkumar Chinchani <rchincha@cisco.com> Co-authored-by: Petu Eusebiu <peusebiu@cisco.com> Co-authored-by: Alexei Dodon <adodon@cisco.com>
713 lines
24 KiB
Go
713 lines
24 KiB
Go
package cache_test
|
|
|
|
import (
|
|
"errors"
|
|
"fmt"
|
|
"path"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/alicebob/miniredis/v2"
|
|
"github.com/go-redis/redismock/v9"
|
|
"github.com/redis/go-redis/v9"
|
|
. "github.com/smartystreets/goconvey/convey"
|
|
|
|
zerr "zotregistry.dev/zot/errors"
|
|
"zotregistry.dev/zot/pkg/log"
|
|
"zotregistry.dev/zot/pkg/storage"
|
|
"zotregistry.dev/zot/pkg/storage/cache"
|
|
"zotregistry.dev/zot/pkg/storage/constants"
|
|
test "zotregistry.dev/zot/pkg/test/common"
|
|
)
|
|
|
|
var ErrTestError = errors.New("TestError")
|
|
|
|
func TestRedisCache(t *testing.T) {
|
|
miniRedis := miniredis.RunT(t)
|
|
|
|
Convey("Make a new cache", t, func() {
|
|
dir := t.TempDir()
|
|
|
|
log := log.NewLogger("debug", "")
|
|
So(log, ShouldNotBeNil)
|
|
|
|
cacheDriver, err := storage.Create("redis", "failTypeAssertion", log)
|
|
So(cacheDriver, ShouldBeNil)
|
|
So(err, ShouldNotBeNil)
|
|
|
|
connOpts, _ := redis.ParseURL("redis://" + miniRedis.Addr())
|
|
client := redis.NewClient(connOpts)
|
|
|
|
cacheDriver, err = storage.Create("redis",
|
|
cache.RedisDriverParameters{client, dir, true, "zot"}, log)
|
|
So(cacheDriver, ShouldNotBeNil)
|
|
So(err, ShouldBeNil)
|
|
|
|
name := cacheDriver.Name()
|
|
So(name, ShouldEqual, "redis")
|
|
|
|
val, err := cacheDriver.GetBlob("key")
|
|
So(err, ShouldEqual, zerr.ErrCacheMiss)
|
|
So(val, ShouldBeEmpty)
|
|
|
|
exists := cacheDriver.HasBlob("key", path.Join(dir, "value"))
|
|
So(exists, ShouldBeFalse)
|
|
|
|
exists = cacheDriver.HasBlob("key", "value")
|
|
So(exists, ShouldBeFalse)
|
|
|
|
err = cacheDriver.PutBlob("key", path.Join(dir, "value"))
|
|
So(err, ShouldBeNil)
|
|
|
|
err = cacheDriver.PutBlob("key", "value")
|
|
So(err, ShouldNotBeNil)
|
|
|
|
exists = cacheDriver.HasBlob("key", path.Join(dir, "value"))
|
|
So(exists, ShouldBeTrue)
|
|
|
|
val, err = cacheDriver.GetBlob("key")
|
|
So(err, ShouldBeNil)
|
|
So(val, ShouldNotBeEmpty)
|
|
|
|
err = cacheDriver.DeleteBlob("bogusKey", "bogusValue")
|
|
So(err, ShouldEqual, zerr.ErrCacheMiss)
|
|
|
|
err = cacheDriver.DeleteBlob("key", "bogusValue")
|
|
So(err, ShouldBeNil)
|
|
|
|
// try to insert empty path
|
|
err = cacheDriver.PutBlob("key", "")
|
|
So(err, ShouldNotBeNil)
|
|
So(err, ShouldEqual, zerr.ErrEmptyValue)
|
|
|
|
connOpts, _ = redis.ParseURL("redis://" + miniRedis.Addr() + "/5")
|
|
client = redis.NewClient(connOpts)
|
|
|
|
cacheDriver, err = storage.Create("redis",
|
|
cache.RedisDriverParameters{client, t.TempDir(), false, "zot"}, log)
|
|
So(cacheDriver, ShouldNotBeNil)
|
|
So(err, ShouldBeNil)
|
|
|
|
err = cacheDriver.PutBlob("key1", "originalBlobPath")
|
|
So(err, ShouldBeNil)
|
|
|
|
err = cacheDriver.PutBlob("key1", "duplicateBlobPath")
|
|
So(err, ShouldBeNil)
|
|
|
|
val, err = cacheDriver.GetBlob("key1")
|
|
So(val, ShouldEqual, "originalBlobPath")
|
|
So(err, ShouldBeNil)
|
|
|
|
err = cacheDriver.DeleteBlob("key1", "duplicateBlobPath")
|
|
So(err, ShouldBeNil)
|
|
|
|
val, err = cacheDriver.GetBlob("key1")
|
|
So(val, ShouldEqual, "originalBlobPath")
|
|
So(err, ShouldBeNil)
|
|
|
|
err = cacheDriver.PutBlob("key1", "duplicateBlobPath")
|
|
So(err, ShouldBeNil)
|
|
|
|
err = cacheDriver.DeleteBlob("key1", "originalBlobPath")
|
|
So(err, ShouldBeNil)
|
|
|
|
val, err = cacheDriver.GetBlob("key1")
|
|
So(val, ShouldEqual, "duplicateBlobPath")
|
|
So(err, ShouldBeNil)
|
|
|
|
err = cacheDriver.DeleteBlob("key1", "duplicateBlobPath")
|
|
So(err, ShouldBeNil)
|
|
|
|
// should be empty
|
|
val, err = cacheDriver.GetBlob("key1")
|
|
So(err, ShouldNotBeNil)
|
|
So(val, ShouldBeEmpty)
|
|
|
|
// try to add three same values
|
|
err = cacheDriver.PutBlob("key2", "duplicate")
|
|
So(err, ShouldBeNil)
|
|
|
|
err = cacheDriver.PutBlob("key2", "duplicate")
|
|
So(err, ShouldBeNil)
|
|
|
|
err = cacheDriver.PutBlob("key2", "duplicate")
|
|
So(err, ShouldBeNil)
|
|
|
|
val, err = cacheDriver.GetBlob("key2")
|
|
So(val, ShouldEqual, "duplicate")
|
|
So(err, ShouldBeNil)
|
|
|
|
err = cacheDriver.DeleteBlob("key2", "duplicate")
|
|
So(err, ShouldBeNil)
|
|
|
|
// should be empty
|
|
val, err = cacheDriver.GetBlob("key2")
|
|
So(err, ShouldNotBeNil)
|
|
So(val, ShouldBeEmpty)
|
|
})
|
|
|
|
Convey("Test cache.GetAllBlos()", t, func() {
|
|
dir := t.TempDir()
|
|
|
|
log := log.NewLogger("debug", "")
|
|
So(log, ShouldNotBeNil)
|
|
|
|
connOpts, _ := redis.ParseURL("redis://" + miniRedis.Addr())
|
|
client := redis.NewClient(connOpts)
|
|
|
|
cacheDriver, err := storage.Create("redis",
|
|
cache.RedisDriverParameters{client, dir, true, "zot"}, log)
|
|
So(cacheDriver, ShouldNotBeNil)
|
|
So(err, ShouldBeNil)
|
|
|
|
name := cacheDriver.Name()
|
|
So(name, ShouldEqual, "redis")
|
|
|
|
blobs, err := cacheDriver.GetAllBlobs("digest")
|
|
So(err, ShouldEqual, zerr.ErrCacheMiss)
|
|
So(blobs, ShouldBeNil)
|
|
|
|
err = cacheDriver.PutBlob("digest", path.Join(dir, "first"))
|
|
So(err, ShouldBeNil)
|
|
|
|
err = cacheDriver.PutBlob("digest", path.Join(dir, "second"))
|
|
So(err, ShouldBeNil)
|
|
|
|
err = cacheDriver.PutBlob("digest", path.Join(dir, "third"))
|
|
So(err, ShouldBeNil)
|
|
|
|
blobs, err = cacheDriver.GetAllBlobs("digest")
|
|
So(err, ShouldBeNil)
|
|
|
|
So(blobs, ShouldResemble, []string{"first", "second", "third"})
|
|
|
|
err = cacheDriver.DeleteBlob("digest", path.Join(dir, "first"))
|
|
So(err, ShouldBeNil)
|
|
|
|
blobs, err = cacheDriver.GetAllBlobs("digest")
|
|
So(err, ShouldBeNil)
|
|
So(len(blobs), ShouldEqual, 2)
|
|
So(blobs, ShouldContain, "second")
|
|
So(blobs, ShouldContain, "third")
|
|
|
|
err = cacheDriver.DeleteBlob("digest", path.Join(dir, "third"))
|
|
So(err, ShouldBeNil)
|
|
|
|
blobs, err = cacheDriver.GetAllBlobs("digest")
|
|
So(err, ShouldBeNil)
|
|
|
|
So(blobs, ShouldResemble, []string{"second"})
|
|
})
|
|
}
|
|
|
|
func TestRedisCacheError(t *testing.T) {
|
|
Convey("Make a new cache", t, func() {
|
|
dir := t.TempDir()
|
|
redisURL := "redis://127.0.0.1:" + test.GetFreePort()
|
|
connOpts, _ := redis.ParseURL(redisURL)
|
|
brokenClient := redis.NewClient(connOpts)
|
|
|
|
log := log.NewLogger("debug", "")
|
|
So(log, ShouldNotBeNil)
|
|
|
|
// redis server is not running
|
|
cacheDriver, err := storage.Create("redis",
|
|
cache.RedisDriverParameters{brokenClient, dir, true, "zot"}, log)
|
|
So(err, ShouldNotBeNil)
|
|
So(cacheDriver, ShouldBeNil)
|
|
})
|
|
|
|
Convey("Redis unreachable", t, func() {
|
|
miniRedis := miniredis.RunT(t)
|
|
dir := t.TempDir()
|
|
|
|
log := log.NewLogger("debug", "")
|
|
So(log, ShouldNotBeNil)
|
|
|
|
connOpts, _ := redis.ParseURL("redis://" + miniRedis.Addr())
|
|
workingClient := redis.NewClient(connOpts)
|
|
|
|
redisURL := "redis://127.0.0.1:" + test.GetFreePort() // must not match miniRedis.Addr()
|
|
connOpts, _ = redis.ParseURL(redisURL)
|
|
brokenClient := redis.NewClient(connOpts)
|
|
|
|
cacheDriver, err := cache.NewRedisCache(
|
|
cache.RedisDriverParameters{workingClient, dir, false, "zot"}, log)
|
|
So(cacheDriver, ShouldNotBeNil)
|
|
So(err, ShouldBeNil)
|
|
|
|
// replace the working driver with the broken one
|
|
cacheDriver.SetClient(brokenClient)
|
|
|
|
err = cacheDriver.PutBlob("key", "val")
|
|
So(err, ShouldNotBeNil)
|
|
|
|
found := cacheDriver.HasBlob("key", "val")
|
|
So(found, ShouldEqual, false)
|
|
|
|
_, err = cacheDriver.GetBlob("key")
|
|
So(err, ShouldNotBeNil)
|
|
|
|
_, err = cacheDriver.GetAllBlobs("key")
|
|
So(err, ShouldNotBeNil)
|
|
|
|
err = cacheDriver.DeleteBlob("key", "val")
|
|
So(err, ShouldNotBeNil)
|
|
})
|
|
}
|
|
|
|
func TestRedisMocked(t *testing.T) {
|
|
Convey("Redis tests using mocks", t, func() {
|
|
dir := t.TempDir()
|
|
|
|
log := log.NewLogger("debug", "")
|
|
So(log, ShouldNotBeNil)
|
|
|
|
tests := []cache.RedisDriverParameters{
|
|
{
|
|
RootDir: dir,
|
|
UseRelPaths: true,
|
|
}, {
|
|
RootDir: dir,
|
|
UseRelPaths: false,
|
|
}, {
|
|
RootDir: dir,
|
|
UseRelPaths: true,
|
|
KeyPrefix: "someprefix",
|
|
}, {
|
|
RootDir: dir,
|
|
UseRelPaths: true,
|
|
KeyPrefix: "zot",
|
|
},
|
|
}
|
|
|
|
for i, redisDriverParams := range tests {
|
|
testID := fmt.Sprintf(" %d", i)
|
|
|
|
keyPrefix := redisDriverParams.KeyPrefix
|
|
if len(keyPrefix) == 0 {
|
|
// check default
|
|
keyPrefix = "zot"
|
|
}
|
|
keyPrefix += ":"
|
|
|
|
// depending on UseRelPaths value we check the relative or absolute value
|
|
// in results using path.Join(pathPrefix, path) in both cases
|
|
pathPrefix := ""
|
|
if !redisDriverParams.UseRelPaths {
|
|
pathPrefix = redisDriverParams.RootDir
|
|
}
|
|
|
|
Convey("PutBlob HExists error"+testID, func() {
|
|
// initialize mock client
|
|
cacheDB, mock := redismock.NewClientMock()
|
|
redisDriverParams.Client = cacheDB
|
|
|
|
mock.ExpectPing().SetVal("OK")
|
|
cacheDriver, err := cache.NewRedisCache(redisDriverParams, log)
|
|
So(cacheDriver, ShouldNotBeNil)
|
|
So(err, ShouldBeNil)
|
|
|
|
mock.Regexp().ExpectSetNX(keyPrefix+"locks:key", `.*`, 8*time.Second).SetVal(true)
|
|
mock.ExpectHExists(keyPrefix+constants.BlobsCache+":"+constants.OriginalBucket, "key").
|
|
SetErr(ErrTestError)
|
|
|
|
err = cacheDriver.PutBlob("key", path.Join(dir, "val"))
|
|
So(err, ShouldEqual, ErrTestError)
|
|
|
|
err = mock.ExpectationsWereMet()
|
|
So(err, ShouldBeNil)
|
|
})
|
|
|
|
Convey("PutBlob HSet error"+testID, func() {
|
|
// initialize mock client
|
|
cacheDB, mock := redismock.NewClientMock()
|
|
redisDriverParams.Client = cacheDB
|
|
|
|
mock.ExpectPing().SetVal("OK")
|
|
cacheDriver, err := cache.NewRedisCache(redisDriverParams, log)
|
|
So(cacheDriver, ShouldNotBeNil)
|
|
So(err, ShouldBeNil)
|
|
|
|
mock.Regexp().ExpectSetNX(keyPrefix+"locks:key", `.*`, 8*time.Second).SetVal(true)
|
|
mock.ExpectHExists(keyPrefix+constants.BlobsCache+":"+constants.OriginalBucket, "key").
|
|
SetVal(false)
|
|
mock.ExpectTxPipeline()
|
|
mock.ExpectHSet(keyPrefix+constants.BlobsCache+":"+constants.OriginalBucket, "key",
|
|
path.Join(pathPrefix, "val")).SetErr(ErrTestError)
|
|
|
|
err = cacheDriver.PutBlob("key", path.Join(dir, "val"))
|
|
So(err, ShouldEqual, ErrTestError)
|
|
|
|
err = mock.ExpectationsWereMet()
|
|
So(err, ShouldBeNil)
|
|
})
|
|
|
|
Convey("PutBlob SAdd error"+testID, func() {
|
|
// initialize mock client
|
|
cacheDB, mock := redismock.NewClientMock()
|
|
redisDriverParams.Client = cacheDB
|
|
|
|
mock.ExpectPing().SetVal("OK")
|
|
cacheDriver, err := cache.NewRedisCache(redisDriverParams, log)
|
|
So(cacheDriver, ShouldNotBeNil)
|
|
So(err, ShouldBeNil)
|
|
|
|
mock.Regexp().ExpectSetNX(keyPrefix+"locks:key", `.*`, 8*time.Second).SetVal(true)
|
|
mock.ExpectHExists(keyPrefix+constants.BlobsCache+":"+constants.OriginalBucket, "key").
|
|
SetVal(false)
|
|
mock.ExpectTxPipeline()
|
|
mock.ExpectHSet(keyPrefix+constants.BlobsCache+":"+constants.OriginalBucket, "key",
|
|
path.Join(pathPrefix, "val")).SetVal(1)
|
|
mock.ExpectSAdd(keyPrefix+constants.BlobsCache+":"+constants.DuplicatesBucket+":key",
|
|
path.Join(pathPrefix, "val")).SetErr(ErrTestError)
|
|
|
|
err = cacheDriver.PutBlob("key", path.Join(dir, "val"))
|
|
So(err, ShouldEqual, ErrTestError)
|
|
|
|
err = mock.ExpectationsWereMet()
|
|
So(err, ShouldBeNil)
|
|
})
|
|
|
|
Convey("PutBlob succeeds original bucket is created"+testID, func() {
|
|
// initialize mock client
|
|
cacheDB, mock := redismock.NewClientMock()
|
|
redisDriverParams.Client = cacheDB
|
|
|
|
mock.ExpectPing().SetVal("OK")
|
|
cacheDriver, err := cache.NewRedisCache(redisDriverParams, log)
|
|
So(cacheDriver, ShouldNotBeNil)
|
|
So(err, ShouldBeNil)
|
|
|
|
mock.Regexp().ExpectSetNX(keyPrefix+"locks:key", `.*`, 8*time.Second).SetVal(true)
|
|
mock.ExpectHExists(keyPrefix+constants.BlobsCache+":"+constants.OriginalBucket, "key").
|
|
SetVal(false)
|
|
mock.ExpectTxPipeline()
|
|
mock.ExpectHSet(keyPrefix+constants.BlobsCache+":"+constants.OriginalBucket, "key",
|
|
path.Join(pathPrefix, "val")).SetVal(1)
|
|
mock.ExpectSAdd(keyPrefix+constants.BlobsCache+":"+constants.DuplicatesBucket+":key",
|
|
path.Join(pathPrefix, "val")).SetVal(1)
|
|
mock.ExpectTxPipelineExec()
|
|
|
|
err = cacheDriver.PutBlob("key", path.Join(dir, "val"))
|
|
So(err, ShouldBeNil)
|
|
|
|
err = mock.ExpectationsWereMet()
|
|
So(err, ShouldBeNil)
|
|
})
|
|
|
|
Convey("PutBlob succeeds original bucket is reused"+testID, func() {
|
|
// initialize mock client
|
|
cacheDB, mock := redismock.NewClientMock()
|
|
redisDriverParams.Client = cacheDB
|
|
|
|
mock.ExpectPing().SetVal("OK")
|
|
cacheDriver, err := cache.NewRedisCache(redisDriverParams, log)
|
|
So(cacheDriver, ShouldNotBeNil)
|
|
So(err, ShouldBeNil)
|
|
|
|
mock.Regexp().ExpectSetNX(keyPrefix+"locks:key", `.*`, 8*time.Second).SetVal(true)
|
|
mock.ExpectHExists(keyPrefix+constants.BlobsCache+":"+constants.OriginalBucket, "key").
|
|
SetVal(true)
|
|
mock.ExpectTxPipeline()
|
|
mock.ExpectSAdd(keyPrefix+constants.BlobsCache+":"+constants.DuplicatesBucket+":key",
|
|
path.Join(pathPrefix, "val")).SetVal(1)
|
|
mock.ExpectTxPipelineExec()
|
|
|
|
err = cacheDriver.PutBlob("key", path.Join(dir, "val"))
|
|
So(err, ShouldBeNil)
|
|
|
|
err = mock.ExpectationsWereMet()
|
|
So(err, ShouldBeNil)
|
|
})
|
|
|
|
Convey("SMembers error in GetAllBlobs"+testID, func() {
|
|
// initialize mock client
|
|
cacheDB, mock := redismock.NewClientMock()
|
|
redisDriverParams.Client = cacheDB
|
|
|
|
mock.ExpectPing().SetVal("OK")
|
|
cacheDriver, err := cache.NewRedisCache(redisDriverParams, log)
|
|
So(cacheDriver, ShouldNotBeNil)
|
|
So(err, ShouldBeNil)
|
|
|
|
mock.ExpectHGet(keyPrefix+constants.BlobsCache+":"+constants.OriginalBucket, "key").
|
|
SetVal(path.Join(pathPrefix, "val"))
|
|
mock.ExpectSMembers(keyPrefix + constants.BlobsCache + ":" + constants.DuplicatesBucket + ":key").
|
|
SetErr(ErrTestError)
|
|
|
|
_, err = cacheDriver.GetAllBlobs("key")
|
|
So(err, ShouldEqual, ErrTestError)
|
|
|
|
err = mock.ExpectationsWereMet()
|
|
So(err, ShouldBeNil)
|
|
})
|
|
|
|
Convey("GetAllBlobs succeeds"+testID, func() {
|
|
// initialize mock client
|
|
cacheDB, mock := redismock.NewClientMock()
|
|
redisDriverParams.Client = cacheDB
|
|
|
|
mock.ExpectPing().SetVal("OK")
|
|
cacheDriver, err := cache.NewRedisCache(redisDriverParams, log)
|
|
So(cacheDriver, ShouldNotBeNil)
|
|
So(err, ShouldBeNil)
|
|
|
|
mock.Regexp().ExpectSetNX(keyPrefix+"locks:key", `.*`, 8*time.Second).SetVal(true)
|
|
mock.ExpectHExists(keyPrefix+constants.BlobsCache+":"+constants.OriginalBucket, "key").
|
|
SetVal(false)
|
|
mock.ExpectTxPipeline()
|
|
mock.ExpectHSet(keyPrefix+constants.BlobsCache+":"+constants.OriginalBucket, "key",
|
|
path.Join(pathPrefix, "val1")).SetVal(1)
|
|
mock.ExpectSAdd(keyPrefix+constants.BlobsCache+":"+constants.DuplicatesBucket+":key",
|
|
path.Join(pathPrefix, "val1")).SetVal(1)
|
|
mock.ExpectTxPipelineExec()
|
|
|
|
err = cacheDriver.PutBlob("key", path.Join(dir, "val1"))
|
|
So(err, ShouldBeNil)
|
|
|
|
mock.Regexp().ExpectSetNX(keyPrefix+"locks:key", `.*`, 8*time.Second).SetVal(true)
|
|
mock.ExpectHExists(keyPrefix+constants.BlobsCache+":"+constants.OriginalBucket, "key").
|
|
SetVal(false)
|
|
mock.ExpectTxPipeline()
|
|
mock.ExpectHSet(keyPrefix+constants.BlobsCache+":"+constants.OriginalBucket, "key",
|
|
path.Join(pathPrefix, "val2")).SetVal(1)
|
|
mock.ExpectSAdd(keyPrefix+constants.BlobsCache+":"+constants.DuplicatesBucket+":key",
|
|
path.Join(pathPrefix, "val2")).SetVal(1)
|
|
mock.ExpectTxPipelineExec()
|
|
|
|
err = cacheDriver.PutBlob("key", path.Join(dir, "val2"))
|
|
So(err, ShouldBeNil)
|
|
|
|
mock.ExpectHGet(keyPrefix+constants.BlobsCache+":"+constants.OriginalBucket, "key").
|
|
SetVal(path.Join(pathPrefix, "val1"))
|
|
mock.ExpectSMembers(keyPrefix + constants.BlobsCache + ":" + constants.DuplicatesBucket + ":key").
|
|
SetVal([]string{path.Join(pathPrefix, "val1"), path.Join(pathPrefix, "val2")})
|
|
|
|
allBlobs, err := cacheDriver.GetAllBlobs("key")
|
|
So(err, ShouldBeNil)
|
|
So(allBlobs, ShouldResemble, []string{path.Join(pathPrefix, "val1"), path.Join(pathPrefix, "val2")})
|
|
|
|
err = mock.ExpectationsWereMet()
|
|
So(err, ShouldBeNil)
|
|
})
|
|
|
|
Convey("HasBlob HExists returns error"+testID, func() {
|
|
// initialize mock client
|
|
cacheDB, mock := redismock.NewClientMock()
|
|
redisDriverParams.Client = cacheDB
|
|
|
|
mock.ExpectPing().SetVal("OK")
|
|
cacheDriver, err := cache.NewRedisCache(redisDriverParams, log)
|
|
So(cacheDriver, ShouldNotBeNil)
|
|
So(err, ShouldBeNil)
|
|
|
|
mock.ExpectSIsMember(keyPrefix+constants.BlobsCache+":"+constants.DuplicatesBucket+":key",
|
|
path.Join(pathPrefix, "val")).SetVal(true)
|
|
mock.ExpectHExists(keyPrefix+constants.BlobsCache+":"+constants.OriginalBucket, "key").
|
|
SetErr(ErrTestError)
|
|
|
|
ok := cacheDriver.HasBlob("key", path.Join(dir, "val"))
|
|
So(ok, ShouldBeFalse)
|
|
|
|
err = mock.ExpectationsWereMet()
|
|
So(err, ShouldBeNil)
|
|
})
|
|
|
|
Convey("HasBlob SIsMember returns error"+testID, func() {
|
|
// initialize mock client
|
|
cacheDB, mock := redismock.NewClientMock()
|
|
redisDriverParams.Client = cacheDB
|
|
|
|
mock.ExpectPing().SetVal("OK")
|
|
cacheDriver, err := cache.NewRedisCache(redisDriverParams, log)
|
|
So(cacheDriver, ShouldNotBeNil)
|
|
So(err, ShouldBeNil)
|
|
|
|
mock.ExpectSIsMember(keyPrefix+constants.BlobsCache+":"+constants.DuplicatesBucket+":key",
|
|
path.Join(pathPrefix, "val")).SetErr(ErrTestError)
|
|
|
|
ok := cacheDriver.HasBlob("key", path.Join(dir, "val"))
|
|
So(ok, ShouldBeFalse)
|
|
|
|
err = mock.ExpectationsWereMet()
|
|
So(err, ShouldBeNil)
|
|
})
|
|
|
|
Convey("HasBlob HExists returns false"+testID, func() {
|
|
// initialize mock client
|
|
cacheDB, mock := redismock.NewClientMock()
|
|
redisDriverParams.Client = cacheDB
|
|
|
|
mock.ExpectPing().SetVal("OK")
|
|
cacheDriver, err := cache.NewRedisCache(redisDriverParams, log)
|
|
So(cacheDriver, ShouldNotBeNil)
|
|
So(err, ShouldBeNil)
|
|
|
|
mock.ExpectSIsMember(keyPrefix+constants.BlobsCache+":"+constants.DuplicatesBucket+":key",
|
|
path.Join(pathPrefix, "val")).SetVal(true)
|
|
mock.ExpectHExists(keyPrefix+constants.BlobsCache+":"+constants.OriginalBucket, "key").
|
|
SetVal(false)
|
|
|
|
ok := cacheDriver.HasBlob("key", path.Join(dir, "val"))
|
|
So(ok, ShouldBeFalse)
|
|
|
|
err = mock.ExpectationsWereMet()
|
|
So(err, ShouldBeNil)
|
|
})
|
|
|
|
Convey("DeleteBlob tests"+testID, func() {
|
|
// initialize mock client
|
|
cacheDB, mock := redismock.NewClientMock()
|
|
redisDriverParams.Client = cacheDB
|
|
|
|
mock.ExpectPing().SetVal("OK")
|
|
cacheDriver, err := cache.NewRedisCache(redisDriverParams, log)
|
|
So(cacheDriver, ShouldNotBeNil)
|
|
So(err, ShouldBeNil)
|
|
|
|
// Create entry for 1st path
|
|
mock.Regexp().ExpectSetNX(keyPrefix+"locks:key", `.*`, 8*time.Second).SetVal(true)
|
|
mock.ExpectHExists(keyPrefix+constants.BlobsCache+":"+constants.OriginalBucket, "key").
|
|
SetVal(false)
|
|
mock.ExpectTxPipeline()
|
|
mock.ExpectHSet(keyPrefix+constants.BlobsCache+":"+constants.OriginalBucket, "key",
|
|
path.Join(pathPrefix, "val1")).SetVal(1)
|
|
mock.ExpectSAdd(keyPrefix+constants.BlobsCache+":"+constants.DuplicatesBucket+":key",
|
|
path.Join(pathPrefix, "val1")).SetVal(1)
|
|
mock.ExpectTxPipelineExec()
|
|
|
|
err = cacheDriver.PutBlob("key", path.Join(dir, "val1"))
|
|
So(err, ShouldBeNil)
|
|
|
|
Convey("DeleteBlob error in HDel"+testID, func() {
|
|
// If the 2nd path does not exist, HDel is callled
|
|
// Error switching to new path
|
|
mock.Regexp().ExpectSetNX(keyPrefix+"locks:key", `.*`, 8*time.Second).SetVal(true)
|
|
mock.ExpectSRem(keyPrefix+constants.BlobsCache+":"+constants.DuplicatesBucket+":key",
|
|
path.Join(pathPrefix, "val1")).SetVal(1)
|
|
mock.ExpectHGet(keyPrefix+constants.BlobsCache+":"+constants.OriginalBucket, "key").
|
|
SetVal(path.Join(pathPrefix, "val1"))
|
|
// failed to get new path
|
|
mock.ExpectSRandMember(keyPrefix + constants.BlobsCache + ":" + constants.DuplicatesBucket + ":key").
|
|
RedisNil()
|
|
mock.ExpectHDel(keyPrefix+constants.BlobsCache+":"+constants.OriginalBucket, "key").
|
|
SetErr(ErrTestError)
|
|
|
|
err = cacheDriver.DeleteBlob("key", path.Join(dir, "val1"))
|
|
So(err, ShouldEqual, ErrTestError)
|
|
})
|
|
|
|
Convey("DeleteBlob succeeds in deleting all data for original blob"+testID, func() {
|
|
// If the 2nd path does not exist, HDel is callled
|
|
// Error switching to new path
|
|
mock.Regexp().ExpectSetNX(keyPrefix+"locks:key", `.*`, 8*time.Second).SetVal(true)
|
|
mock.ExpectSRem(keyPrefix+constants.BlobsCache+":"+constants.DuplicatesBucket+":key",
|
|
path.Join(pathPrefix, "val1")).SetVal(1)
|
|
mock.ExpectHGet(keyPrefix+constants.BlobsCache+":"+constants.OriginalBucket, "key").
|
|
SetVal(path.Join(pathPrefix, "val1"))
|
|
// failed to get new path
|
|
mock.ExpectSRandMember(keyPrefix + constants.BlobsCache + ":" + constants.DuplicatesBucket + ":key").
|
|
RedisNil()
|
|
mock.ExpectHDel(keyPrefix+constants.BlobsCache+":"+constants.OriginalBucket, "key").
|
|
SetVal(1)
|
|
|
|
err = cacheDriver.DeleteBlob("key", path.Join(dir, "val1"))
|
|
So(err, ShouldBeNil)
|
|
})
|
|
|
|
Convey("DeleteBlob error in SRandMember"+testID, func() {
|
|
// Create entry for 2nd path
|
|
mock.Regexp().ExpectSetNX(keyPrefix+"locks:key", `.*`, 8*time.Second).SetVal(true)
|
|
mock.ExpectHExists(keyPrefix+constants.BlobsCache+":"+constants.OriginalBucket, "key").
|
|
SetVal(false)
|
|
mock.ExpectTxPipeline()
|
|
mock.ExpectHSet(keyPrefix+constants.BlobsCache+":"+constants.OriginalBucket, "key",
|
|
path.Join(pathPrefix, "val2")).SetVal(1)
|
|
mock.ExpectSAdd(keyPrefix+constants.BlobsCache+":"+constants.DuplicatesBucket+":key",
|
|
path.Join(pathPrefix, "val2")).SetVal(1)
|
|
mock.ExpectTxPipelineExec()
|
|
|
|
err = cacheDriver.PutBlob("key", path.Join(dir, "val2"))
|
|
So(err, ShouldBeNil)
|
|
|
|
// Error switching to new path
|
|
mock.Regexp().ExpectSetNX(keyPrefix+"locks:key", `.*`, 8*time.Second).SetVal(true)
|
|
mock.ExpectSRem(keyPrefix+constants.BlobsCache+":"+constants.DuplicatesBucket+":key",
|
|
path.Join(pathPrefix, "val1")).SetVal(1)
|
|
mock.ExpectHGet(keyPrefix+constants.BlobsCache+":"+constants.OriginalBucket, "key").
|
|
SetVal(path.Join(pathPrefix, "val1"))
|
|
// failed to get new path
|
|
mock.ExpectSRandMember(keyPrefix + constants.BlobsCache + ":" + constants.DuplicatesBucket + ":key").
|
|
SetErr(ErrTestError)
|
|
|
|
err = cacheDriver.DeleteBlob("key", path.Join(dir, "val1"))
|
|
So(err, ShouldEqual, ErrTestError)
|
|
})
|
|
|
|
Convey("DeleteBlob error in HSet"+testID, func() {
|
|
// Create entry for 2nd path
|
|
mock.Regexp().ExpectSetNX(keyPrefix+"locks:key", `.*`, 8*time.Second).SetVal(true)
|
|
mock.ExpectHExists(keyPrefix+constants.BlobsCache+":"+constants.OriginalBucket, "key").
|
|
SetVal(false)
|
|
mock.ExpectTxPipeline()
|
|
mock.ExpectHSet(keyPrefix+constants.BlobsCache+":"+constants.OriginalBucket, "key",
|
|
path.Join(pathPrefix, "val2")).SetVal(1)
|
|
mock.ExpectSAdd(keyPrefix+constants.BlobsCache+":"+constants.DuplicatesBucket+":key",
|
|
path.Join(pathPrefix, "val2")).SetVal(1)
|
|
mock.ExpectTxPipelineExec()
|
|
|
|
err = cacheDriver.PutBlob("key", path.Join(dir, "val2"))
|
|
So(err, ShouldBeNil)
|
|
|
|
// Error switching to new path
|
|
mock.Regexp().ExpectSetNX(keyPrefix+"locks:key", `.*`, 8*time.Second).SetVal(true)
|
|
mock.ExpectSRem(keyPrefix+constants.BlobsCache+":"+constants.DuplicatesBucket+":key",
|
|
path.Join(pathPrefix, "val1")).SetVal(1)
|
|
mock.ExpectHGet(keyPrefix+constants.BlobsCache+":"+constants.OriginalBucket, "key").
|
|
SetVal(path.Join(pathPrefix, "val1"))
|
|
mock.ExpectSRandMember(keyPrefix + constants.BlobsCache + ":" + constants.DuplicatesBucket + ":key").
|
|
SetVal(path.Join(pathPrefix, "val2"))
|
|
mock.ExpectHSet(keyPrefix+constants.BlobsCache+":"+constants.OriginalBucket, "key",
|
|
path.Join(pathPrefix, "val2")).SetErr(ErrTestError)
|
|
|
|
err = cacheDriver.DeleteBlob("key", path.Join(dir, "val1"))
|
|
So(err, ShouldEqual, ErrTestError)
|
|
})
|
|
|
|
Convey("DeleteBlob succeeds in switching original blob path"+testID, func() {
|
|
// Create entry for 2nd path
|
|
mock.Regexp().ExpectSetNX(keyPrefix+"locks:key", `.*`, 8*time.Second).SetVal(true)
|
|
mock.ExpectHExists(keyPrefix+constants.BlobsCache+":"+constants.OriginalBucket, "key").
|
|
SetVal(false)
|
|
mock.ExpectTxPipeline()
|
|
mock.ExpectHSet(keyPrefix+constants.BlobsCache+":"+constants.OriginalBucket, "key",
|
|
path.Join(pathPrefix, "val2")).SetVal(1)
|
|
mock.ExpectSAdd(keyPrefix+constants.BlobsCache+":"+constants.DuplicatesBucket+":key",
|
|
path.Join(pathPrefix, "val2")).SetVal(1)
|
|
mock.ExpectTxPipelineExec()
|
|
|
|
err = cacheDriver.PutBlob("key", path.Join(dir, "val2"))
|
|
So(err, ShouldBeNil)
|
|
|
|
mock.Regexp().ExpectSetNX(keyPrefix+"locks:key", `.*`, 8*time.Second).SetVal(true)
|
|
mock.ExpectSRem(keyPrefix+constants.BlobsCache+":"+constants.DuplicatesBucket+":key",
|
|
path.Join(pathPrefix, "val1")).SetVal(1)
|
|
mock.ExpectHGet(keyPrefix+constants.BlobsCache+":"+constants.OriginalBucket, "key").
|
|
SetVal(path.Join(pathPrefix, "val1"))
|
|
mock.ExpectSRandMember(keyPrefix + constants.BlobsCache + ":" + constants.DuplicatesBucket + ":key").
|
|
SetVal(path.Join(pathPrefix, "val2"))
|
|
mock.ExpectHSet(keyPrefix+constants.BlobsCache+":"+constants.OriginalBucket, "key",
|
|
path.Join(pathPrefix, "val2")).SetVal(1)
|
|
|
|
err = cacheDriver.DeleteBlob("key", path.Join(dir, "val1"))
|
|
So(err, ShouldBeNil)
|
|
})
|
|
|
|
err = mock.ExpectationsWereMet()
|
|
So(err, ShouldBeNil)
|
|
})
|
|
}
|
|
})
|
|
}
|