-
Notifications
You must be signed in to change notification settings - Fork 809
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
* adding support for Redis cache Signed-off-by: Dmitry Shmulevich <[email protected]> * fix lint errors Signed-off-by: Dmitry Shmulevich <[email protected]> * minor bugfix Signed-off-by: Dmitry Shmulevich <[email protected]> * proper use of pool connections Signed-off-by: Dmitry Shmulevich <[email protected]> * fix lint errors Signed-off-by: Dmitry Shmulevich <[email protected]> * addressed comments added unit test Signed-off-by: Dmitry Shmulevich <[email protected]> * update unit tests Signed-off-by: Dmitry Shmulevich <[email protected]> * reuse context in request timeout Signed-off-by: Dmitry Shmulevich <[email protected]> * use correct function names in logs Signed-off-by: Dmitry Shmulevich <[email protected]> * use common config for cache storage Signed-off-by: Dmitry Shmulevich <[email protected]> * added missing module dependency Signed-off-by: Dmitry Shmulevich <[email protected]> * delete extra newline Signed-off-by: Dmitry Shmulevich <[email protected]> * optimize redis SET Signed-off-by: Dmitry Shmulevich <[email protected]> * adding support for Redis cache Signed-off-by: Dmitry Shmulevich <[email protected]> * fix lint errors Signed-off-by: Dmitry Shmulevich <[email protected]> * minor bugfix Signed-off-by: Dmitry Shmulevich <[email protected]> * proper use of pool connections Signed-off-by: Dmitry Shmulevich <[email protected]> * fix lint errors Signed-off-by: Dmitry Shmulevich <[email protected]> * addressed comments added unit test Signed-off-by: Dmitry Shmulevich <[email protected]> * update unit tests Signed-off-by: Dmitry Shmulevich <[email protected]> * reuse context in request timeout Signed-off-by: Dmitry Shmulevich <[email protected]> * use correct function names in logs Signed-off-by: Dmitry Shmulevich <[email protected]> * use common config for cache storage Signed-off-by: Dmitry Shmulevich <[email protected]> * added missing module dependency Signed-off-by: Dmitry Shmulevich <[email protected]> * delete extra newline Signed-off-by: Dmitry Shmulevich <[email protected]> * optimize redis SET Signed-off-by: Dmitry Shmulevich <[email protected]> * undo common config for storage systems Signed-off-by: Dmitry Shmulevich <[email protected]> * restore changes in CHANGELOG.md Signed-off-by: Dmitry Shmulevich <[email protected]> * fixed lint error Signed-off-by: Dmitry Shmulevich <[email protected]> * fixed mod-check error Signed-off-by: Dmitry Shmulevich <[email protected]> * fix metric labels Signed-off-by: Dmitry Shmulevich <[email protected]> * addressed comments Signed-off-by: Dmitry Shmulevich <[email protected]>
- Loading branch information
1 parent
0be2dd5
commit 857bb84
Showing
34 changed files
with
4,858 additions
and
0 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,154 @@ | ||
| package cache | ||
|
|
||
| import ( | ||
| "context" | ||
| "flag" | ||
| "time" | ||
|
|
||
| "github.com/cortexproject/cortex/pkg/util" | ||
| "github.com/go-kit/kit/log/level" | ||
| "github.com/gomodule/redigo/redis" | ||
| ) | ||
|
|
||
| // RedisCache type caches chunks in redis | ||
| type RedisCache struct { | ||
| name string | ||
| expiration int | ||
| timeout time.Duration | ||
| pool *redis.Pool | ||
| } | ||
|
|
||
| // RedisConfig defines how a RedisCache should be constructed. | ||
| type RedisConfig struct { | ||
| Endpoint string `yaml:"endpoint,omitempty"` | ||
| Timeout time.Duration `yaml:"timeout,omitempty"` | ||
| Expiration time.Duration `yaml:"expiration,omitempty"` | ||
| MaxIdleConns int `yaml:"max_idle_conns,omitempty"` | ||
| MaxActiveConns int `yaml:"max_active_conns,omitempty"` | ||
| } | ||
|
|
||
| // RegisterFlagsWithPrefix adds the flags required to config this to the given FlagSet | ||
| func (cfg *RedisConfig) RegisterFlagsWithPrefix(prefix, description string, f *flag.FlagSet) { | ||
| f.StringVar(&cfg.Endpoint, prefix+"redis.endpoint", "", description+"Redis service endpoint to use when caching chunks. If empty, no redis will be used.") | ||
| f.DurationVar(&cfg.Timeout, prefix+"redis.timeout", 100*time.Millisecond, description+"Maximum time to wait before giving up on redis requests.") | ||
| f.DurationVar(&cfg.Expiration, prefix+"redis.expiration", 0, description+"How long keys stay in the redis.") | ||
| f.IntVar(&cfg.MaxIdleConns, prefix+"redis.max-idle-conns", 80, description+"Maximum number of idle connections in pool.") | ||
| f.IntVar(&cfg.MaxActiveConns, prefix+"redis.max-active-conns", 0, description+"Maximum number of active connections in pool.") | ||
| } | ||
|
|
||
| // NewRedisCache creates a new RedisCache | ||
| func NewRedisCache(cfg RedisConfig, name string, pool *redis.Pool) *RedisCache { | ||
| // pool != nil only in unit tests | ||
| if pool == nil { | ||
| pool = &redis.Pool{ | ||
| MaxIdle: cfg.MaxIdleConns, | ||
| MaxActive: cfg.MaxActiveConns, | ||
| Dial: func() (redis.Conn, error) { | ||
| c, err := redis.Dial("tcp", cfg.Endpoint) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| return c, err | ||
| }, | ||
| } | ||
| } | ||
|
|
||
| cache := &RedisCache{ | ||
| expiration: int(cfg.Expiration.Seconds()), | ||
| timeout: cfg.Timeout, | ||
| name: name, | ||
| pool: pool, | ||
| } | ||
|
|
||
| if err := cache.ping(context.Background()); err != nil { | ||
| level.Error(util.Logger).Log("msg", "error connecting to redis", "endpoint", cfg.Endpoint, "err", err) | ||
| } | ||
|
|
||
| return cache | ||
| } | ||
|
|
||
| // Fetch gets keys from the cache. The keys that are found must be in the order of the keys requested. | ||
| func (c *RedisCache) Fetch(ctx context.Context, keys []string) (found []string, bufs [][]byte, missed []string) { | ||
| data, err := c.mget(ctx, keys) | ||
|
|
||
| if err != nil { | ||
| level.Error(util.Logger).Log("msg", "failed to get from redis", "name", c.name, "err", err) | ||
| missed = make([]string, len(keys)) | ||
| copy(missed, keys) | ||
| return | ||
| } | ||
| for i, key := range keys { | ||
| if data[i] != nil { | ||
| found = append(found, key) | ||
| bufs = append(bufs, data[i]) | ||
| } else { | ||
| missed = append(missed, key) | ||
| } | ||
| } | ||
| return | ||
| } | ||
|
|
||
| // Store stores the key in the cache. | ||
| func (c *RedisCache) Store(ctx context.Context, keys []string, bufs [][]byte) { | ||
| err := c.mset(ctx, keys, bufs, c.expiration) | ||
| if err != nil { | ||
| level.Error(util.Logger).Log("msg", "failed to put to redis", "name", c.name, "err", err) | ||
| } | ||
| } | ||
|
|
||
| // Stop stops the redis client. | ||
| func (c *RedisCache) Stop() error { | ||
| return c.pool.Close() | ||
| } | ||
|
|
||
| // mset adds key-value pairs to the cache. | ||
| func (c *RedisCache) mset(ctx context.Context, keys []string, bufs [][]byte, ttl int) error { | ||
| conn := c.pool.Get() | ||
| defer conn.Close() | ||
|
|
||
| if err := conn.Send("MULTI"); err != nil { | ||
| return err | ||
| } | ||
| for i := range keys { | ||
| if err := conn.Send("SETEX", keys[i], ttl, bufs[i]); err != nil { | ||
| return err | ||
| } | ||
| } | ||
| _, err := redis.DoWithTimeout(conn, c.timeout, "EXEC") | ||
| return err | ||
| } | ||
|
|
||
| // mget retrieves values from the cache. | ||
| func (c *RedisCache) mget(ctx context.Context, keys []string) ([][]byte, error) { | ||
| intf := make([]interface{}, len(keys)) | ||
| for i, key := range keys { | ||
| intf[i] = key | ||
| } | ||
|
|
||
| conn := c.pool.Get() | ||
| defer conn.Close() | ||
|
|
||
| return redis.ByteSlices(redis.DoWithTimeout(conn, c.timeout, "MGET", intf...)) | ||
| } | ||
|
|
||
| func (c *RedisCache) ping(ctx context.Context) error { | ||
| conn := c.pool.Get() | ||
| defer conn.Close() | ||
|
|
||
| pong, err := redis.DoWithTimeout(conn, c.timeout, "PING") | ||
| if err == nil { | ||
| _, err = redis.String(pong, err) | ||
| } | ||
| return err | ||
| } | ||
|
|
||
| func redisStatusCode(err error) string { | ||
| switch err { | ||
| case nil: | ||
| return "200" | ||
| case redis.ErrNil: | ||
| return "404" | ||
| default: | ||
| return "500" | ||
| } | ||
| } |
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,88 @@ | ||
| package cache_test | ||
|
|
||
| import ( | ||
| "context" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/cortexproject/cortex/pkg/chunk/cache" | ||
| "github.com/gomodule/redigo/redis" | ||
| "github.com/rafaeljusto/redigomock" | ||
| "github.com/stretchr/testify/require" | ||
| ) | ||
|
|
||
| func TestRedisCache(t *testing.T) { | ||
| cfg := cache.RedisConfig{ | ||
| Timeout: 10 * time.Millisecond, | ||
| } | ||
|
|
||
| conn := redigomock.NewConn() | ||
| conn.Clear() | ||
| pool := redis.NewPool(func() (redis.Conn, error) { | ||
| return conn, nil | ||
| }, 10) | ||
|
|
||
| keys := []string{"key1", "key2", "key3"} | ||
| bufs := [][]byte{[]byte("data1"), []byte("data2"), []byte("data3")} | ||
| miss := []string{"miss1", "miss2"} | ||
|
|
||
| // ensure input correctness | ||
| nHit := len(keys) | ||
| require.Len(t, bufs, nHit) | ||
|
|
||
| // mock Redis Store | ||
| mockRedisStore(conn, keys, bufs) | ||
|
|
||
| //mock cache hit | ||
| keyIntf := make([]interface{}, nHit) | ||
| bufIntf := make([]interface{}, nHit) | ||
|
|
||
| for i := 0; i < nHit; i++ { | ||
| keyIntf[i] = keys[i] | ||
| bufIntf[i] = bufs[i] | ||
| } | ||
| conn.Command("MGET", keyIntf...).Expect(bufIntf) | ||
|
|
||
| // mock cache miss | ||
| nMiss := len(miss) | ||
| missIntf := make([]interface{}, nMiss) | ||
| for i, s := range miss { | ||
| missIntf[i] = s | ||
| } | ||
| conn.Command("MGET", missIntf...).ExpectError(nil) | ||
|
|
||
| // mock the cache | ||
| c := cache.NewRedisCache(cfg, "mock", pool) | ||
| ctx := context.Background() | ||
|
|
||
| c.Store(ctx, keys, bufs) | ||
|
|
||
| // test hits | ||
| found, data, missed := c.Fetch(ctx, keys) | ||
|
|
||
| require.Len(t, found, nHit) | ||
| require.Len(t, missed, 0) | ||
| for i := 0; i < nHit; i++ { | ||
| require.Equal(t, keys[i], found[i]) | ||
| require.Equal(t, bufs[i], data[i]) | ||
| } | ||
|
|
||
| // test misses | ||
| found, _, missed = c.Fetch(ctx, miss) | ||
|
|
||
| require.Len(t, found, 0) | ||
| require.Len(t, missed, nMiss) | ||
| for i := 0; i < nMiss; i++ { | ||
| require.Equal(t, miss[i], missed[i]) | ||
| } | ||
| } | ||
|
|
||
| func mockRedisStore(conn *redigomock.Conn, keys []string, bufs [][]byte) { | ||
| conn.Command("MULTI") | ||
| ret := []interface{}{} | ||
| for i := range keys { | ||
| conn.Command("SETEX", keys[i], 0, bufs[i]) | ||
| ret = append(ret, "OK") | ||
| } | ||
| conn.Command("EXEC").Expect(ret) | ||
| } |
Oops, something went wrong.