feat(cache): redis cache (#8822)

# Which Problems Are Solved

Add a cache implementation using Redis single mode. This does not add
support for Redis Cluster or sentinel.

# How the Problems Are Solved

Added the `internal/cache/redis` package. All operations occur
atomically, including setting of secondary indexes, using LUA scripts
where needed.

The [`miniredis`](https://github.com/alicebob/miniredis) package is used
to run unit tests.

# Additional Changes

- Move connector code to `internal/cache/connector/...` and remove
duplicate code from `query` and `command` packages.
- Fix a missed invalidation on the restrictions projection

# Additional Context

Closes #8130
This commit is contained in:
Tim Möhlmann
2024-11-04 11:44:51 +01:00
committed by GitHub
parent 9c3e5e467b
commit 250f2344c8
50 changed files with 1767 additions and 293 deletions

View File

@@ -6,8 +6,16 @@ import (
"time"
"github.com/zitadel/logging"
)
"github.com/zitadel/zitadel/internal/database/postgres"
// Purpose describes which object types are stored by a cache.
type Purpose int
//go:generate enumer -type Purpose -transform snake -trimprefix Purpose
const (
PurposeUnspecified Purpose = iota
PurposeAuthzInstance
PurposeMilestones
)
// Cache stores objects with a value of type `V`.
@@ -72,18 +80,19 @@ type Entry[I, K comparable] interface {
Keys(index I) (key []K)
}
type CachesConfig struct {
Connectors struct {
Memory MemoryConnectorConfig
Postgres PostgresConnectorConfig
// Redis redis.Config?
}
Instance *CacheConfig
Milestones *CacheConfig
}
type Connector int
type CacheConfig struct {
Connector string
//go:generate enumer -type Connector -transform snake -trimprefix Connector -linecomment -text
const (
// Empty line comment ensures empty string for unspecified value
ConnectorUnspecified Connector = iota //
ConnectorMemory
ConnectorPostgres
ConnectorRedis
)
type Config struct {
Connector Connector
// Age since an object was added to the cache,
// after which the object is considered invalid.
@@ -99,14 +108,3 @@ type CacheConfig struct {
// By default only errors are logged to stdout.
Log *logging.Config
}
type MemoryConnectorConfig struct {
Enabled bool
AutoPrune AutoPruneConfig
}
type PostgresConnectorConfig struct {
Enabled bool
AutoPrune AutoPruneConfig
Connection postgres.Config
}

69
internal/cache/connector/connector.go vendored Normal file
View File

@@ -0,0 +1,69 @@
// Package connector provides glue between the [cache.Cache] interface and implementations from the connector sub-packages.
package connector
import (
"context"
"fmt"
"github.com/zitadel/zitadel/internal/cache"
"github.com/zitadel/zitadel/internal/cache/connector/gomap"
"github.com/zitadel/zitadel/internal/cache/connector/noop"
"github.com/zitadel/zitadel/internal/cache/connector/pg"
"github.com/zitadel/zitadel/internal/cache/connector/redis"
"github.com/zitadel/zitadel/internal/database"
)
type CachesConfig struct {
Connectors struct {
Memory gomap.Config
Postgres pg.Config
Redis redis.Config
}
Instance *cache.Config
Milestones *cache.Config
}
type Connectors struct {
Config CachesConfig
Memory *gomap.Connector
Postgres *pg.Connector
Redis *redis.Connector
}
func StartConnectors(conf *CachesConfig, client *database.DB) (Connectors, error) {
if conf == nil {
return Connectors{}, nil
}
return Connectors{
Config: *conf,
Memory: gomap.NewConnector(conf.Connectors.Memory),
Postgres: pg.NewConnector(conf.Connectors.Postgres, client),
Redis: redis.NewConnector(conf.Connectors.Redis),
}, nil
}
func StartCache[I ~int, K ~string, V cache.Entry[I, K]](background context.Context, indices []I, purpose cache.Purpose, conf *cache.Config, connectors Connectors) (cache.Cache[I, K, V], error) {
if conf == nil || conf.Connector == cache.ConnectorUnspecified {
return noop.NewCache[I, K, V](), nil
}
if conf.Connector == cache.ConnectorMemory && connectors.Memory != nil {
c := gomap.NewCache[I, K, V](background, indices, *conf)
connectors.Memory.Config.StartAutoPrune(background, c, purpose)
return c, nil
}
if conf.Connector == cache.ConnectorPostgres && connectors.Postgres != nil {
c, err := pg.NewCache[I, K, V](background, purpose, *conf, indices, connectors.Postgres)
if err != nil {
return nil, fmt.Errorf("start cache: %w", err)
}
connectors.Postgres.Config.AutoPrune.StartAutoPrune(background, c, purpose)
return c, nil
}
if conf.Connector == cache.ConnectorRedis && connectors.Redis != nil {
db := connectors.Redis.Config.DBOffset + int(purpose)
c := redis.NewCache[I, K, V](*conf, connectors.Redis, db, indices)
return c, nil
}
return nil, fmt.Errorf("cache connector %q not enabled", conf.Connector)
}

View File

@@ -0,0 +1,23 @@
package gomap
import (
"github.com/zitadel/zitadel/internal/cache"
)
type Config struct {
Enabled bool
AutoPrune cache.AutoPruneConfig
}
type Connector struct {
Config cache.AutoPruneConfig
}
func NewConnector(config Config) *Connector {
if !config.Enabled {
return nil
}
return &Connector{
Config: config.AutoPrune,
}
}

View File

@@ -14,14 +14,14 @@ import (
)
type mapCache[I, K comparable, V cache.Entry[I, K]] struct {
config *cache.CacheConfig
config *cache.Config
indexMap map[I]*index[K, V]
logger *slog.Logger
}
// NewCache returns an in-memory Cache implementation based on the builtin go map type.
// Object values are stored as-is and there is no encoding or decoding involved.
func NewCache[I, K comparable, V cache.Entry[I, K]](background context.Context, indices []I, config cache.CacheConfig) cache.PrunerCache[I, K, V] {
func NewCache[I, K comparable, V cache.Entry[I, K]](background context.Context, indices []I, config cache.Config) cache.PrunerCache[I, K, V] {
m := &mapCache[I, K, V]{
config: &config,
indexMap: make(map[I]*index[K, V], len(indices)),
@@ -116,7 +116,7 @@ func (c *mapCache[I, K, V]) Truncate(ctx context.Context) error {
type index[K comparable, V any] struct {
mutex sync.RWMutex
config *cache.CacheConfig
config *cache.Config
entries map[K]*entry[V]
}
@@ -177,7 +177,7 @@ type entry[V any] struct {
lastUse atomic.Int64 // UnixMicro time
}
func (e *entry[V]) isValid(c *cache.CacheConfig) bool {
func (e *entry[V]) isValid(c *cache.Config) bool {
if e.invalid.Load() {
return false
}

View File

@@ -41,7 +41,7 @@ func (o *testObject) Keys(index testIndex) []string {
}
func Test_mapCache_Get(t *testing.T) {
c := NewCache[testIndex, string, *testObject](context.Background(), testIndices, cache.CacheConfig{
c := NewCache[testIndex, string, *testObject](context.Background(), testIndices, cache.Config{
MaxAge: time.Second,
LastUseAge: time.Second / 4,
Log: &logging.Config{
@@ -103,7 +103,7 @@ func Test_mapCache_Get(t *testing.T) {
}
func Test_mapCache_Invalidate(t *testing.T) {
c := NewCache[testIndex, string, *testObject](context.Background(), testIndices, cache.CacheConfig{
c := NewCache[testIndex, string, *testObject](context.Background(), testIndices, cache.Config{
MaxAge: time.Second,
LastUseAge: time.Second / 4,
Log: &logging.Config{
@@ -124,7 +124,7 @@ func Test_mapCache_Invalidate(t *testing.T) {
}
func Test_mapCache_Delete(t *testing.T) {
c := NewCache[testIndex, string, *testObject](context.Background(), testIndices, cache.CacheConfig{
c := NewCache[testIndex, string, *testObject](context.Background(), testIndices, cache.Config{
MaxAge: time.Second,
LastUseAge: time.Second / 4,
Log: &logging.Config{
@@ -157,7 +157,7 @@ func Test_mapCache_Delete(t *testing.T) {
}
func Test_mapCache_Prune(t *testing.T) {
c := NewCache[testIndex, string, *testObject](context.Background(), testIndices, cache.CacheConfig{
c := NewCache[testIndex, string, *testObject](context.Background(), testIndices, cache.Config{
MaxAge: time.Second,
LastUseAge: time.Second / 4,
Log: &logging.Config{
@@ -193,7 +193,7 @@ func Test_mapCache_Prune(t *testing.T) {
}
func Test_mapCache_Truncate(t *testing.T) {
c := NewCache[testIndex, string, *testObject](context.Background(), testIndices, cache.CacheConfig{
c := NewCache[testIndex, string, *testObject](context.Background(), testIndices, cache.Config{
MaxAge: time.Second,
LastUseAge: time.Second / 4,
Log: &logging.Config{
@@ -235,7 +235,7 @@ func Test_entry_isValid(t *testing.T) {
tests := []struct {
name string
fields fields
config *cache.CacheConfig
config *cache.Config
want bool
}{
{
@@ -245,7 +245,7 @@ func Test_entry_isValid(t *testing.T) {
invalid: true,
lastUse: time.Now(),
},
config: &cache.CacheConfig{
config: &cache.Config{
MaxAge: time.Minute,
LastUseAge: time.Second,
},
@@ -258,7 +258,7 @@ func Test_entry_isValid(t *testing.T) {
invalid: false,
lastUse: time.Now(),
},
config: &cache.CacheConfig{
config: &cache.Config{
MaxAge: time.Minute,
LastUseAge: time.Second,
},
@@ -271,7 +271,7 @@ func Test_entry_isValid(t *testing.T) {
invalid: false,
lastUse: time.Now(),
},
config: &cache.CacheConfig{
config: &cache.Config{
LastUseAge: time.Second,
},
want: true,
@@ -283,7 +283,7 @@ func Test_entry_isValid(t *testing.T) {
invalid: false,
lastUse: time.Now().Add(-(time.Second * 2)),
},
config: &cache.CacheConfig{
config: &cache.Config{
MaxAge: time.Minute,
LastUseAge: time.Second,
},
@@ -296,7 +296,7 @@ func Test_entry_isValid(t *testing.T) {
invalid: false,
lastUse: time.Now().Add(-(time.Second * 2)),
},
config: &cache.CacheConfig{
config: &cache.Config{
MaxAge: time.Minute,
},
want: true,
@@ -308,7 +308,7 @@ func Test_entry_isValid(t *testing.T) {
invalid: false,
lastUse: time.Now(),
},
config: &cache.CacheConfig{
config: &cache.Config{
MaxAge: time.Minute,
LastUseAge: time.Second,
},

View File

@@ -0,0 +1,28 @@
package pg
import (
"github.com/zitadel/zitadel/internal/cache"
"github.com/zitadel/zitadel/internal/database"
)
type Config struct {
Enabled bool
AutoPrune cache.AutoPruneConfig
}
type Connector struct {
PGXPool
Dialect string
Config Config
}
func NewConnector(config Config, client *database.DB) *Connector {
if !config.Enabled {
return nil
}
return &Connector{
PGXPool: client.Pool,
Dialect: client.Type(),
Config: config,
}
}

View File

@@ -40,25 +40,25 @@ type PGXPool interface {
}
type pgCache[I ~int, K ~string, V cache.Entry[I, K]] struct {
name string
config *cache.CacheConfig
indices []I
pool PGXPool
logger *slog.Logger
purpose cache.Purpose
config *cache.Config
indices []I
connector *Connector
logger *slog.Logger
}
// NewCache returns a cache that stores and retrieves objects using PostgreSQL unlogged tables.
func NewCache[I ~int, K ~string, V cache.Entry[I, K]](ctx context.Context, name string, config cache.CacheConfig, indices []I, pool PGXPool, dialect string) (cache.PrunerCache[I, K, V], error) {
func NewCache[I ~int, K ~string, V cache.Entry[I, K]](ctx context.Context, purpose cache.Purpose, config cache.Config, indices []I, connector *Connector) (cache.PrunerCache[I, K, V], error) {
c := &pgCache[I, K, V]{
name: name,
config: &config,
indices: indices,
pool: pool,
logger: config.Log.Slog().With("cache_name", name),
purpose: purpose,
config: &config,
indices: indices,
connector: connector,
logger: config.Log.Slog().With("cache_purpose", purpose),
}
c.logger.InfoContext(ctx, "pg cache logging enabled")
if dialect == "postgres" {
if connector.Dialect == "postgres" {
if err := c.createPartition(ctx); err != nil {
return nil, err
}
@@ -68,10 +68,10 @@ func NewCache[I ~int, K ~string, V cache.Entry[I, K]](ctx context.Context, name
func (c *pgCache[I, K, V]) createPartition(ctx context.Context) error {
var query strings.Builder
if err := createPartitionTmpl.Execute(&query, c.name); err != nil {
if err := createPartitionTmpl.Execute(&query, c.purpose.String()); err != nil {
return err
}
_, err := c.pool.Exec(ctx, query.String())
_, err := c.connector.Exec(ctx, query.String())
return err
}
@@ -87,7 +87,7 @@ func (c *pgCache[I, K, V]) set(ctx context.Context, entry V) (err error) {
keys := c.indexKeysFromEntry(entry)
c.logger.DebugContext(ctx, "pg cache set", "index_key", keys)
_, err = c.pool.Exec(ctx, setQuery, c.name, keys, entry)
_, err = c.connector.Exec(ctx, setQuery, c.purpose.String(), keys, entry)
if err != nil {
c.logger.ErrorContext(ctx, "pg cache set", "err", err)
return err
@@ -117,7 +117,7 @@ func (c *pgCache[I, K, V]) get(ctx context.Context, index I, key K) (value V, er
if !slices.Contains(c.indices, index) {
return value, cache.NewIndexUnknownErr(index)
}
err = c.pool.QueryRow(ctx, getQuery, c.name, index, key, c.config.MaxAge, c.config.LastUseAge).Scan(&value)
err = c.connector.QueryRow(ctx, getQuery, c.purpose.String(), index, key, c.config.MaxAge, c.config.LastUseAge).Scan(&value)
return value, err
}
@@ -125,7 +125,7 @@ func (c *pgCache[I, K, V]) Invalidate(ctx context.Context, index I, keys ...K) (
ctx, span := tracing.NewSpan(ctx)
defer func() { span.EndWithError(err) }()
_, err = c.pool.Exec(ctx, invalidateQuery, c.name, index, keys)
_, err = c.connector.Exec(ctx, invalidateQuery, c.purpose.String(), index, keys)
c.logger.DebugContext(ctx, "pg cache invalidate", "index", index, "keys", keys)
return err
}
@@ -134,7 +134,7 @@ func (c *pgCache[I, K, V]) Delete(ctx context.Context, index I, keys ...K) (err
ctx, span := tracing.NewSpan(ctx)
defer func() { span.EndWithError(err) }()
_, err = c.pool.Exec(ctx, deleteQuery, c.name, index, keys)
_, err = c.connector.Exec(ctx, deleteQuery, c.purpose.String(), index, keys)
c.logger.DebugContext(ctx, "pg cache delete", "index", index, "keys", keys)
return err
}
@@ -143,7 +143,7 @@ func (c *pgCache[I, K, V]) Prune(ctx context.Context) (err error) {
ctx, span := tracing.NewSpan(ctx)
defer func() { span.EndWithError(err) }()
_, err = c.pool.Exec(ctx, pruneQuery, c.name, c.config.MaxAge, c.config.LastUseAge)
_, err = c.connector.Exec(ctx, pruneQuery, c.purpose.String(), c.config.MaxAge, c.config.LastUseAge)
c.logger.DebugContext(ctx, "pg cache prune")
return err
}
@@ -152,7 +152,7 @@ func (c *pgCache[I, K, V]) Truncate(ctx context.Context) (err error) {
ctx, span := tracing.NewSpan(ctx)
defer func() { span.EndWithError(err) }()
_, err = c.pool.Exec(ctx, truncateQuery, c.name)
_, err = c.connector.Exec(ctx, truncateQuery, c.purpose.String())
c.logger.DebugContext(ctx, "pg cache truncate")
return err
}

View File

@@ -67,7 +67,7 @@ func TestNewCache(t *testing.T) {
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
conf := cache.CacheConfig{
conf := cache.Config{
Log: &logging.Config{
Level: "debug",
AddSource: true,
@@ -76,8 +76,12 @@ func TestNewCache(t *testing.T) {
pool, err := pgxmock.NewPool()
require.NoError(t, err)
tt.expect(pool)
connector := &Connector{
PGXPool: pool,
Dialect: "postgres",
}
c, err := NewCache[testIndex, string, *testObject](context.Background(), cacheName, conf, testIndices, pool, "postgres")
c, err := NewCache[testIndex, string, *testObject](context.Background(), cachePurpose, conf, testIndices, connector)
require.ErrorIs(t, err, tt.wantErr)
if tt.wantErr == nil {
assert.NotNil(t, c)
@@ -111,7 +115,7 @@ func Test_pgCache_Set(t *testing.T) {
},
expect: func(ppi pgxmock.PgxCommonIface) {
ppi.ExpectExec(queryExpect).
WithArgs("test",
WithArgs(cachePurpose.String(),
[]indexKey[testIndex, string]{
{IndexID: testIndexID, IndexKey: "id1"},
{IndexID: testIndexName, IndexKey: "foo"},
@@ -135,7 +139,7 @@ func Test_pgCache_Set(t *testing.T) {
},
expect: func(ppi pgxmock.PgxCommonIface) {
ppi.ExpectExec(queryExpect).
WithArgs("test",
WithArgs(cachePurpose.String(),
[]indexKey[testIndex, string]{
{IndexID: testIndexID, IndexKey: "id1"},
{IndexID: testIndexName, IndexKey: "foo"},
@@ -151,7 +155,7 @@ func Test_pgCache_Set(t *testing.T) {
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
c, pool := prepareCache(t, cache.CacheConfig{})
c, pool := prepareCache(t, cache.Config{})
defer pool.Close()
tt.expect(pool)
@@ -173,7 +177,7 @@ func Test_pgCache_Get(t *testing.T) {
}
tests := []struct {
name string
config cache.CacheConfig
config cache.Config
args args
expect func(pgxmock.PgxCommonIface)
want *testObject
@@ -181,7 +185,7 @@ func Test_pgCache_Get(t *testing.T) {
}{
{
name: "invalid index",
config: cache.CacheConfig{
config: cache.Config{
MaxAge: time.Minute,
LastUseAge: time.Second,
},
@@ -194,7 +198,7 @@ func Test_pgCache_Get(t *testing.T) {
},
{
name: "no rows",
config: cache.CacheConfig{
config: cache.Config{
MaxAge: 0,
LastUseAge: 0,
},
@@ -204,14 +208,14 @@ func Test_pgCache_Get(t *testing.T) {
},
expect: func(pci pgxmock.PgxCommonIface) {
pci.ExpectQuery(queryExpect).
WithArgs("test", testIndexID, "id1", time.Duration(0), time.Duration(0)).
WithArgs(cachePurpose.String(), testIndexID, "id1", time.Duration(0), time.Duration(0)).
WillReturnRows(pgxmock.NewRows([]string{"payload"}))
},
wantOk: false,
},
{
name: "error",
config: cache.CacheConfig{
config: cache.Config{
MaxAge: 0,
LastUseAge: 0,
},
@@ -221,14 +225,14 @@ func Test_pgCache_Get(t *testing.T) {
},
expect: func(pci pgxmock.PgxCommonIface) {
pci.ExpectQuery(queryExpect).
WithArgs("test", testIndexID, "id1", time.Duration(0), time.Duration(0)).
WithArgs(cachePurpose.String(), testIndexID, "id1", time.Duration(0), time.Duration(0)).
WillReturnError(pgx.ErrTxClosed)
},
wantOk: false,
},
{
name: "ok",
config: cache.CacheConfig{
config: cache.Config{
MaxAge: time.Minute,
LastUseAge: time.Second,
},
@@ -238,7 +242,7 @@ func Test_pgCache_Get(t *testing.T) {
},
expect: func(pci pgxmock.PgxCommonIface) {
pci.ExpectQuery(queryExpect).
WithArgs("test", testIndexID, "id1", time.Minute, time.Second).
WithArgs(cachePurpose.String(), testIndexID, "id1", time.Minute, time.Second).
WillReturnRows(
pgxmock.NewRows([]string{"payload"}).AddRow(&testObject{
ID: "id1",
@@ -276,14 +280,14 @@ func Test_pgCache_Invalidate(t *testing.T) {
}
tests := []struct {
name string
config cache.CacheConfig
config cache.Config
args args
expect func(pgxmock.PgxCommonIface)
wantErr error
}{
{
name: "error",
config: cache.CacheConfig{
config: cache.Config{
MaxAge: 0,
LastUseAge: 0,
},
@@ -293,14 +297,14 @@ func Test_pgCache_Invalidate(t *testing.T) {
},
expect: func(pci pgxmock.PgxCommonIface) {
pci.ExpectExec(queryExpect).
WithArgs("test", testIndexID, []string{"id1", "id2"}).
WithArgs(cachePurpose.String(), testIndexID, []string{"id1", "id2"}).
WillReturnError(pgx.ErrTxClosed)
},
wantErr: pgx.ErrTxClosed,
},
{
name: "ok",
config: cache.CacheConfig{
config: cache.Config{
MaxAge: time.Minute,
LastUseAge: time.Second,
},
@@ -310,7 +314,7 @@ func Test_pgCache_Invalidate(t *testing.T) {
},
expect: func(pci pgxmock.PgxCommonIface) {
pci.ExpectExec(queryExpect).
WithArgs("test", testIndexID, []string{"id1", "id2"}).
WithArgs(cachePurpose.String(), testIndexID, []string{"id1", "id2"}).
WillReturnResult(pgxmock.NewResult("DELETE", 1))
},
},
@@ -338,14 +342,14 @@ func Test_pgCache_Delete(t *testing.T) {
}
tests := []struct {
name string
config cache.CacheConfig
config cache.Config
args args
expect func(pgxmock.PgxCommonIface)
wantErr error
}{
{
name: "error",
config: cache.CacheConfig{
config: cache.Config{
MaxAge: 0,
LastUseAge: 0,
},
@@ -355,14 +359,14 @@ func Test_pgCache_Delete(t *testing.T) {
},
expect: func(pci pgxmock.PgxCommonIface) {
pci.ExpectExec(queryExpect).
WithArgs("test", testIndexID, []string{"id1", "id2"}).
WithArgs(cachePurpose.String(), testIndexID, []string{"id1", "id2"}).
WillReturnError(pgx.ErrTxClosed)
},
wantErr: pgx.ErrTxClosed,
},
{
name: "ok",
config: cache.CacheConfig{
config: cache.Config{
MaxAge: time.Minute,
LastUseAge: time.Second,
},
@@ -372,7 +376,7 @@ func Test_pgCache_Delete(t *testing.T) {
},
expect: func(pci pgxmock.PgxCommonIface) {
pci.ExpectExec(queryExpect).
WithArgs("test", testIndexID, []string{"id1", "id2"}).
WithArgs(cachePurpose.String(), testIndexID, []string{"id1", "id2"}).
WillReturnResult(pgxmock.NewResult("DELETE", 1))
},
},
@@ -396,32 +400,32 @@ func Test_pgCache_Prune(t *testing.T) {
queryExpect := regexp.QuoteMeta(pruneQuery)
tests := []struct {
name string
config cache.CacheConfig
config cache.Config
expect func(pgxmock.PgxCommonIface)
wantErr error
}{
{
name: "error",
config: cache.CacheConfig{
config: cache.Config{
MaxAge: 0,
LastUseAge: 0,
},
expect: func(pci pgxmock.PgxCommonIface) {
pci.ExpectExec(queryExpect).
WithArgs("test", time.Duration(0), time.Duration(0)).
WithArgs(cachePurpose.String(), time.Duration(0), time.Duration(0)).
WillReturnError(pgx.ErrTxClosed)
},
wantErr: pgx.ErrTxClosed,
},
{
name: "ok",
config: cache.CacheConfig{
config: cache.Config{
MaxAge: time.Minute,
LastUseAge: time.Second,
},
expect: func(pci pgxmock.PgxCommonIface) {
pci.ExpectExec(queryExpect).
WithArgs("test", time.Minute, time.Second).
WithArgs(cachePurpose.String(), time.Minute, time.Second).
WillReturnResult(pgxmock.NewResult("DELETE", 1))
},
},
@@ -445,32 +449,32 @@ func Test_pgCache_Truncate(t *testing.T) {
queryExpect := regexp.QuoteMeta(truncateQuery)
tests := []struct {
name string
config cache.CacheConfig
config cache.Config
expect func(pgxmock.PgxCommonIface)
wantErr error
}{
{
name: "error",
config: cache.CacheConfig{
config: cache.Config{
MaxAge: 0,
LastUseAge: 0,
},
expect: func(pci pgxmock.PgxCommonIface) {
pci.ExpectExec(queryExpect).
WithArgs("test").
WithArgs(cachePurpose.String()).
WillReturnError(pgx.ErrTxClosed)
},
wantErr: pgx.ErrTxClosed,
},
{
name: "ok",
config: cache.CacheConfig{
config: cache.Config{
MaxAge: time.Minute,
LastUseAge: time.Second,
},
expect: func(pci pgxmock.PgxCommonIface) {
pci.ExpectExec(queryExpect).
WithArgs("test").
WithArgs(cachePurpose.String()).
WillReturnResult(pgxmock.NewResult("DELETE", 1))
},
},
@@ -491,18 +495,18 @@ func Test_pgCache_Truncate(t *testing.T) {
}
const (
cacheName = "test"
expectedCreatePartitionQuery = `create unlogged table if not exists cache.objects_test
cachePurpose = cache.PurposeAuthzInstance
expectedCreatePartitionQuery = `create unlogged table if not exists cache.objects_authz_instance
partition of cache.objects
for values in ('test');
for values in ('authz_instance');
create unlogged table if not exists cache.string_keys_test
create unlogged table if not exists cache.string_keys_authz_instance
partition of cache.string_keys
for values in ('test');
for values in ('authz_instance');
`
)
func prepareCache(t *testing.T, conf cache.CacheConfig) (cache.PrunerCache[testIndex, string, *testObject], pgxmock.PgxPoolIface) {
func prepareCache(t *testing.T, conf cache.Config) (cache.PrunerCache[testIndex, string, *testObject], pgxmock.PgxPoolIface) {
conf.Log = &logging.Config{
Level: "debug",
AddSource: true,
@@ -512,8 +516,11 @@ func prepareCache(t *testing.T, conf cache.CacheConfig) (cache.PrunerCache[testI
pool.ExpectExec(regexp.QuoteMeta(expectedCreatePartitionQuery)).
WillReturnResult(pgxmock.NewResult("CREATE TABLE", 0))
c, err := NewCache[testIndex, string, *testObject](context.Background(), cacheName, conf, testIndices, pool, "postgres")
connector := &Connector{
PGXPool: pool,
Dialect: "postgres",
}
c, err := NewCache[testIndex, string, *testObject](context.Background(), cachePurpose, conf, testIndices, connector)
require.NoError(t, err)
return c, pool
}

View File

@@ -0,0 +1,10 @@
local function remove(object_id)
local setKey = keySetKey(object_id)
local keys = redis.call("SMEMBERS", setKey)
local n = #keys
for i = 1, n do
redis.call("DEL", keys[i])
end
redis.call("DEL", setKey)
redis.call("DEL", object_id)
end

View File

@@ -0,0 +1,3 @@
-- SELECT ensures the DB namespace for each script.
-- When used, it consumes the first ARGV entry.
redis.call("SELECT", ARGV[1])

View File

@@ -0,0 +1,17 @@
-- keySetKey returns the redis key of the set containing all keys to the object.
local function keySetKey (object_id)
return object_id .. "-keys"
end
local function getTime()
return tonumber(redis.call('TIME')[1])
end
-- getCall wrapts redis.call so a nil is returned instead of false.
local function getCall (...)
local result = redis.call(...)
if result == false then
return nil
end
return result
end

View File

@@ -0,0 +1,154 @@
package redis
import (
"crypto/tls"
"time"
"github.com/redis/go-redis/v9"
)
type Config struct {
Enabled bool
// The network type, either tcp or unix.
// Default is tcp.
Network string
// host:port address.
Addr string
// ClientName will execute the `CLIENT SETNAME ClientName` command for each conn.
ClientName string
// Use the specified Username to authenticate the current connection
// with one of the connections defined in the ACL list when connecting
// to a Redis 6.0 instance, or greater, that is using the Redis ACL system.
Username string
// Optional password. Must match the password specified in the
// requirepass server configuration option (if connecting to a Redis 5.0 instance, or lower),
// or the User Password when connecting to a Redis 6.0 instance, or greater,
// that is using the Redis ACL system.
Password string
// Each ZITADEL cache uses an incremental DB namespace.
// This option offsets the first DB so it doesn't conflict with other databases on the same server.
// Note that ZITADEL uses FLUSHDB command to truncate a cache.
// This can have destructive consequences when overlapping DB namespaces are used.
DBOffset int
// Maximum number of retries before giving up.
// Default is 3 retries; -1 (not 0) disables retries.
MaxRetries int
// Minimum backoff between each retry.
// Default is 8 milliseconds; -1 disables backoff.
MinRetryBackoff time.Duration
// Maximum backoff between each retry.
// Default is 512 milliseconds; -1 disables backoff.
MaxRetryBackoff time.Duration
// Dial timeout for establishing new connections.
// Default is 5 seconds.
DialTimeout time.Duration
// Timeout for socket reads. If reached, commands will fail
// with a timeout instead of blocking. Supported values:
// - `0` - default timeout (3 seconds).
// - `-1` - no timeout (block indefinitely).
// - `-2` - disables SetReadDeadline calls completely.
ReadTimeout time.Duration
// Timeout for socket writes. If reached, commands will fail
// with a timeout instead of blocking. Supported values:
// - `0` - default timeout (3 seconds).
// - `-1` - no timeout (block indefinitely).
// - `-2` - disables SetWriteDeadline calls completely.
WriteTimeout time.Duration
// Type of connection pool.
// true for FIFO pool, false for LIFO pool.
// Note that FIFO has slightly higher overhead compared to LIFO,
// but it helps closing idle connections faster reducing the pool size.
PoolFIFO bool
// Base number of socket connections.
// Default is 10 connections per every available CPU as reported by runtime.GOMAXPROCS.
// If there is not enough connections in the pool, new connections will be allocated in excess of PoolSize,
// you can limit it through MaxActiveConns
PoolSize int
// Amount of time client waits for connection if all connections
// are busy before returning an error.
// Default is ReadTimeout + 1 second.
PoolTimeout time.Duration
// Minimum number of idle connections which is useful when establishing
// new connection is slow.
// Default is 0. the idle connections are not closed by default.
MinIdleConns int
// Maximum number of idle connections.
// Default is 0. the idle connections are not closed by default.
MaxIdleConns int
// Maximum number of connections allocated by the pool at a given time.
// When zero, there is no limit on the number of connections in the pool.
MaxActiveConns int
// ConnMaxIdleTime is the maximum amount of time a connection may be idle.
// Should be less than server's timeout.
//
// Expired connections may be closed lazily before reuse.
// If d <= 0, connections are not closed due to a connection's idle time.
//
// Default is 30 minutes. -1 disables idle timeout check.
ConnMaxIdleTime time.Duration
// ConnMaxLifetime is the maximum amount of time a connection may be reused.
//
// Expired connections may be closed lazily before reuse.
// If <= 0, connections are not closed due to a connection's age.
//
// Default is to not close idle connections.
ConnMaxLifetime time.Duration
EnableTLS bool
// Disable set-lib on connect. Default is false.
DisableIndentity bool
// Add suffix to client name. Default is empty.
IdentitySuffix string
}
type Connector struct {
*redis.Client
Config Config
}
func NewConnector(config Config) *Connector {
if !config.Enabled {
return nil
}
return &Connector{
Client: redis.NewClient(optionsFromConfig(config)),
Config: config,
}
}
func optionsFromConfig(c Config) *redis.Options {
opts := &redis.Options{
Network: c.Network,
Addr: c.Addr,
ClientName: c.ClientName,
Protocol: 3,
Username: c.Username,
Password: c.Password,
MaxRetries: c.MaxRetries,
MinRetryBackoff: c.MinRetryBackoff,
MaxRetryBackoff: c.MaxRetryBackoff,
DialTimeout: c.DialTimeout,
ReadTimeout: c.ReadTimeout,
WriteTimeout: c.WriteTimeout,
ContextTimeoutEnabled: true,
PoolFIFO: c.PoolFIFO,
PoolTimeout: c.PoolTimeout,
MinIdleConns: c.MinIdleConns,
MaxIdleConns: c.MaxIdleConns,
MaxActiveConns: c.MaxActiveConns,
ConnMaxIdleTime: c.ConnMaxIdleTime,
ConnMaxLifetime: c.ConnMaxLifetime,
DisableIndentity: c.DisableIndentity,
IdentitySuffix: c.IdentitySuffix,
}
if c.EnableTLS {
opts.TLSConfig = new(tls.Config)
}
return opts
}

29
internal/cache/connector/redis/get.lua vendored Normal file
View File

@@ -0,0 +1,29 @@
local result = redis.call("GET", KEYS[1])
if result == false then
return nil
end
local object_id = tostring(result)
local object = getCall("HGET", object_id, "object")
if object == nil then
-- object expired, but there are keys that need to be cleaned up
remove(object_id)
return nil
end
-- max-age must be checked manually
local expiry = getCall("HGET", object_id, "expiry")
if not (expiry == nil) and expiry > 0 then
if getTime() > expiry then
remove(object_id)
return nil
end
end
local usage_lifetime = getCall("HGET", object_id, "usage_lifetime")
-- reset usage based TTL
if not (usage_lifetime == nil) and tonumber(usage_lifetime) > 0 then
redis.call('EXPIRE', object_id, usage_lifetime)
end
return object

View File

@@ -0,0 +1,9 @@
local n = #KEYS
for i = 1, n do
local result = redis.call("GET", KEYS[i])
if result == false then
return nil
end
local object_id = tostring(result)
remove(object_id)
end

172
internal/cache/connector/redis/redis.go vendored Normal file
View File

@@ -0,0 +1,172 @@
package redis
import (
"context"
_ "embed"
"encoding/json"
"errors"
"fmt"
"log/slog"
"strings"
"time"
"github.com/google/uuid"
"github.com/redis/go-redis/v9"
"github.com/zitadel/zitadel/internal/cache"
"github.com/zitadel/zitadel/internal/telemetry/tracing"
)
var (
//go:embed _select.lua
selectComponent string
//go:embed _util.lua
utilComponent string
//go:embed _remove.lua
removeComponent string
//go:embed set.lua
setScript string
//go:embed get.lua
getScript string
//go:embed invalidate.lua
invalidateScript string
// Don't mind the creative "import"
setParsed = redis.NewScript(strings.Join([]string{selectComponent, utilComponent, setScript}, "\n"))
getParsed = redis.NewScript(strings.Join([]string{selectComponent, utilComponent, removeComponent, getScript}, "\n"))
invalidateParsed = redis.NewScript(strings.Join([]string{selectComponent, utilComponent, removeComponent, invalidateScript}, "\n"))
)
type redisCache[I, K comparable, V cache.Entry[I, K]] struct {
db int
config *cache.Config
indices []I
connector *Connector
logger *slog.Logger
}
// NewCache returns a cache that stores and retrieves object using single Redis.
func NewCache[I, K comparable, V cache.Entry[I, K]](config cache.Config, client *Connector, db int, indices []I) cache.Cache[I, K, V] {
return &redisCache[I, K, V]{
config: &config,
db: db,
indices: indices,
connector: client,
logger: config.Log.Slog(),
}
}
func (c *redisCache[I, K, V]) Set(ctx context.Context, value V) {
if _, err := c.set(ctx, value); err != nil {
c.logger.ErrorContext(ctx, "redis cache set", "err", err)
}
}
func (c *redisCache[I, K, V]) set(ctx context.Context, value V) (objectID string, err error) {
ctx, span := tracing.NewSpan(ctx)
defer func() { span.EndWithError(err) }()
// Internal ID used for the object
objectID = uuid.NewString()
keys := []string{objectID}
// flatten the secondary keys
for _, index := range c.indices {
keys = append(keys, c.redisIndexKeys(index, value.Keys(index)...)...)
}
var buf strings.Builder
err = json.NewEncoder(&buf).Encode(value)
if err != nil {
return "", err
}
err = setParsed.Run(ctx, c.connector, keys,
c.db, // DB namespace
buf.String(), // object
int64(c.config.LastUseAge/time.Second), // usage_lifetime
int64(c.config.MaxAge/time.Second), // max_age,
).Err()
// redis.Nil is always returned because the script doesn't have a return value.
if err != nil && !errors.Is(err, redis.Nil) {
return "", err
}
return objectID, nil
}
func (c *redisCache[I, K, V]) Get(ctx context.Context, index I, key K) (value V, ok bool) {
var (
obj any
err error
)
ctx, span := tracing.NewSpan(ctx)
defer func() {
if errors.Is(err, redis.Nil) {
err = nil
}
span.EndWithError(err)
}()
logger := c.logger.With("index", index, "key", key)
obj, err = getParsed.Run(ctx, c.connector, c.redisIndexKeys(index, key), c.db).Result()
if err != nil && !errors.Is(err, redis.Nil) {
logger.ErrorContext(ctx, "redis cache get", "err", err)
return value, false
}
data, ok := obj.(string)
if !ok {
logger.With("err", err).InfoContext(ctx, "redis cache miss")
return value, false
}
err = json.NewDecoder(strings.NewReader(data)).Decode(&value)
if err != nil {
logger.ErrorContext(ctx, "redis cache get", "err", fmt.Errorf("decode: %w", err))
return value, false
}
return value, true
}
func (c *redisCache[I, K, V]) Invalidate(ctx context.Context, index I, key ...K) (err error) {
ctx, span := tracing.NewSpan(ctx)
defer func() { span.EndWithError(err) }()
if len(key) == 0 {
return nil
}
err = invalidateParsed.Run(ctx, c.connector, c.redisIndexKeys(index, key...), c.db).Err()
// redis.Nil is always returned because the script doesn't have a return value.
if err != nil && !errors.Is(err, redis.Nil) {
return err
}
return nil
}
func (c *redisCache[I, K, V]) Delete(ctx context.Context, index I, key ...K) (err error) {
ctx, span := tracing.NewSpan(ctx)
defer func() { span.EndWithError(err) }()
if len(key) == 0 {
return nil
}
pipe := c.connector.Pipeline()
pipe.Select(ctx, c.db)
pipe.Del(ctx, c.redisIndexKeys(index, key...)...)
_, err = pipe.Exec(ctx)
return err
}
func (c *redisCache[I, K, V]) Truncate(ctx context.Context) (err error) {
ctx, span := tracing.NewSpan(ctx)
defer func() { span.EndWithError(err) }()
pipe := c.connector.Pipeline()
pipe.Select(ctx, c.db)
pipe.FlushDB(ctx)
_, err = pipe.Exec(ctx)
return err
}
func (c *redisCache[I, K, V]) redisIndexKeys(index I, keys ...K) []string {
out := make([]string, len(keys))
for i, k := range keys {
out[i] = fmt.Sprintf("%v:%v", index, k)
}
return out
}

View File

@@ -0,0 +1,714 @@
package redis
import (
"context"
"testing"
"time"
"github.com/alicebob/miniredis/v2"
"github.com/redis/go-redis/v9"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/zitadel/logging"
"github.com/zitadel/zitadel/internal/cache"
)
type testIndex int
const (
testIndexID testIndex = iota
testIndexName
)
const (
testDB = 99
)
var testIndices = []testIndex{
testIndexID,
testIndexName,
}
type testObject struct {
ID string
Name []string
}
func (o *testObject) Keys(index testIndex) []string {
switch index {
case testIndexID:
return []string{o.ID}
case testIndexName:
return o.Name
default:
return nil
}
}
func Test_redisCache_set(t *testing.T) {
type args struct {
ctx context.Context
value *testObject
}
tests := []struct {
name string
config cache.Config
args args
assertions func(t *testing.T, s *miniredis.Miniredis, objectID string)
wantErr error
}{
{
name: "ok",
config: cache.Config{},
args: args{
ctx: context.Background(),
value: &testObject{
ID: "one",
Name: []string{"foo", "bar"},
},
},
assertions: func(t *testing.T, s *miniredis.Miniredis, objectID string) {
s.CheckGet(t, "0:one", objectID)
s.CheckGet(t, "1:foo", objectID)
s.CheckGet(t, "1:bar", objectID)
assert.Empty(t, s.HGet(objectID, "expiry"))
assert.JSONEq(t, `{"ID":"one","Name":["foo","bar"]}`, s.HGet(objectID, "object"))
},
},
{
name: "with last use TTL",
config: cache.Config{
LastUseAge: time.Second,
},
args: args{
ctx: context.Background(),
value: &testObject{
ID: "one",
Name: []string{"foo", "bar"},
},
},
assertions: func(t *testing.T, s *miniredis.Miniredis, objectID string) {
s.CheckGet(t, "0:one", objectID)
s.CheckGet(t, "1:foo", objectID)
s.CheckGet(t, "1:bar", objectID)
assert.Empty(t, s.HGet(objectID, "expiry"))
assert.JSONEq(t, `{"ID":"one","Name":["foo","bar"]}`, s.HGet(objectID, "object"))
assert.Positive(t, s.TTL(objectID))
s.FastForward(2 * time.Second)
v, err := s.Get(objectID)
require.Error(t, err)
assert.Empty(t, v)
},
},
{
name: "with last use TTL and max age",
config: cache.Config{
MaxAge: time.Minute,
LastUseAge: time.Second,
},
args: args{
ctx: context.Background(),
value: &testObject{
ID: "one",
Name: []string{"foo", "bar"},
},
},
assertions: func(t *testing.T, s *miniredis.Miniredis, objectID string) {
s.CheckGet(t, "0:one", objectID)
s.CheckGet(t, "1:foo", objectID)
s.CheckGet(t, "1:bar", objectID)
assert.NotEmpty(t, s.HGet(objectID, "expiry"))
assert.JSONEq(t, `{"ID":"one","Name":["foo","bar"]}`, s.HGet(objectID, "object"))
assert.Positive(t, s.TTL(objectID))
s.FastForward(2 * time.Second)
v, err := s.Get(objectID)
require.Error(t, err)
assert.Empty(t, v)
},
},
{
name: "with max age TTL",
config: cache.Config{
MaxAge: time.Minute,
},
args: args{
ctx: context.Background(),
value: &testObject{
ID: "one",
Name: []string{"foo", "bar"},
},
},
assertions: func(t *testing.T, s *miniredis.Miniredis, objectID string) {
s.CheckGet(t, "0:one", objectID)
s.CheckGet(t, "1:foo", objectID)
s.CheckGet(t, "1:bar", objectID)
assert.Empty(t, s.HGet(objectID, "expiry"))
assert.JSONEq(t, `{"ID":"one","Name":["foo","bar"]}`, s.HGet(objectID, "object"))
assert.Positive(t, s.TTL(objectID))
s.FastForward(2 * time.Minute)
v, err := s.Get(objectID)
require.Error(t, err)
assert.Empty(t, v)
},
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
c, server := prepareCache(t, tt.config)
rc := c.(*redisCache[testIndex, string, *testObject])
objectID, err := rc.set(tt.args.ctx, tt.args.value)
require.ErrorIs(t, err, tt.wantErr)
t.Log(rc.connector.HGetAll(context.Background(), objectID))
tt.assertions(t, server, objectID)
})
}
}
func Test_redisCache_Get(t *testing.T) {
type args struct {
ctx context.Context
index testIndex
key string
}
tests := []struct {
name string
config cache.Config
preparation func(t *testing.T, c cache.Cache[testIndex, string, *testObject], s *miniredis.Miniredis)
args args
want *testObject
wantOK bool
}{
{
name: "connection error",
config: cache.Config{},
preparation: func(_ *testing.T, _ cache.Cache[testIndex, string, *testObject], s *miniredis.Miniredis) {
s.RequireAuth("foobar")
},
args: args{
ctx: context.Background(),
index: testIndexName,
key: "foo",
},
wantOK: false,
},
{
name: "get by ID",
config: cache.Config{},
preparation: func(t *testing.T, c cache.Cache[testIndex, string, *testObject], s *miniredis.Miniredis) {
c.Set(context.Background(), &testObject{
ID: "one",
Name: []string{"foo", "bar"},
})
},
args: args{
ctx: context.Background(),
index: testIndexID,
key: "one",
},
want: &testObject{
ID: "one",
Name: []string{"foo", "bar"},
},
wantOK: true,
},
{
name: "get by name",
config: cache.Config{},
preparation: func(t *testing.T, c cache.Cache[testIndex, string, *testObject], s *miniredis.Miniredis) {
c.Set(context.Background(), &testObject{
ID: "one",
Name: []string{"foo", "bar"},
})
},
args: args{
ctx: context.Background(),
index: testIndexName,
key: "foo",
},
want: &testObject{
ID: "one",
Name: []string{"foo", "bar"},
},
wantOK: true,
},
{
name: "usage timeout",
config: cache.Config{
LastUseAge: time.Minute,
},
preparation: func(t *testing.T, c cache.Cache[testIndex, string, *testObject], s *miniredis.Miniredis) {
c.Set(context.Background(), &testObject{
ID: "one",
Name: []string{"foo", "bar"},
})
_, ok := c.Get(context.Background(), testIndexID, "one")
require.True(t, ok)
s.FastForward(2 * time.Minute)
},
args: args{
ctx: context.Background(),
index: testIndexName,
key: "foo",
},
want: nil,
wantOK: false,
},
{
name: "max age timeout",
config: cache.Config{
MaxAge: time.Minute,
},
preparation: func(t *testing.T, c cache.Cache[testIndex, string, *testObject], s *miniredis.Miniredis) {
c.Set(context.Background(), &testObject{
ID: "one",
Name: []string{"foo", "bar"},
})
_, ok := c.Get(context.Background(), testIndexID, "one")
require.True(t, ok)
s.FastForward(2 * time.Minute)
},
args: args{
ctx: context.Background(),
index: testIndexName,
key: "foo",
},
want: nil,
wantOK: false,
},
{
name: "not found",
config: cache.Config{},
preparation: func(t *testing.T, c cache.Cache[testIndex, string, *testObject], s *miniredis.Miniredis) {
c.Set(context.Background(), &testObject{
ID: "one",
Name: []string{"foo", "bar"},
})
},
args: args{
ctx: context.Background(),
index: testIndexName,
key: "spanac",
},
wantOK: false,
},
{
name: "json decode error",
config: cache.Config{},
preparation: func(t *testing.T, c cache.Cache[testIndex, string, *testObject], s *miniredis.Miniredis) {
c.Set(context.Background(), &testObject{
ID: "one",
Name: []string{"foo", "bar"},
})
objectID, err := s.Get(c.(*redisCache[testIndex, string, *testObject]).redisIndexKeys(testIndexID, "one")[0])
require.NoError(t, err)
s.HSet(objectID, "object", "~~~")
},
args: args{
ctx: context.Background(),
index: testIndexID,
key: "one",
},
wantOK: false,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
c, server := prepareCache(t, tt.config)
tt.preparation(t, c, server)
t.Log(server.Keys())
got, ok := c.Get(tt.args.ctx, tt.args.index, tt.args.key)
require.Equal(t, tt.wantOK, ok)
assert.Equal(t, tt.want, got)
})
}
}
func Test_redisCache_Invalidate(t *testing.T) {
type args struct {
ctx context.Context
index testIndex
key []string
}
tests := []struct {
name string
config cache.Config
preparation func(t *testing.T, c cache.Cache[testIndex, string, *testObject], s *miniredis.Miniredis)
assertions func(t *testing.T, c cache.Cache[testIndex, string, *testObject])
args args
wantErr bool
}{
{
name: "connection error",
config: cache.Config{},
preparation: func(_ *testing.T, _ cache.Cache[testIndex, string, *testObject], s *miniredis.Miniredis) {
s.RequireAuth("foobar")
},
args: args{
ctx: context.Background(),
index: testIndexName,
key: []string{"foo"},
},
wantErr: true,
},
{
name: "no keys, noop",
config: cache.Config{},
preparation: func(t *testing.T, c cache.Cache[testIndex, string, *testObject], s *miniredis.Miniredis) {
c.Set(context.Background(), &testObject{
ID: "one",
Name: []string{"foo", "bar"},
})
},
args: args{
ctx: context.Background(),
index: testIndexID,
key: []string{},
},
wantErr: false,
},
{
name: "invalidate by ID",
config: cache.Config{},
preparation: func(t *testing.T, c cache.Cache[testIndex, string, *testObject], s *miniredis.Miniredis) {
c.Set(context.Background(), &testObject{
ID: "one",
Name: []string{"foo", "bar"},
})
},
assertions: func(t *testing.T, c cache.Cache[testIndex, string, *testObject]) {
obj, ok := c.Get(context.Background(), testIndexID, "one")
assert.False(t, ok)
assert.Nil(t, obj)
obj, ok = c.Get(context.Background(), testIndexName, "foo")
assert.False(t, ok)
assert.Nil(t, obj)
},
args: args{
ctx: context.Background(),
index: testIndexID,
key: []string{"one"},
},
wantErr: false,
},
{
name: "invalidate by name",
config: cache.Config{},
preparation: func(t *testing.T, c cache.Cache[testIndex, string, *testObject], s *miniredis.Miniredis) {
c.Set(context.Background(), &testObject{
ID: "one",
Name: []string{"foo", "bar"},
})
},
assertions: func(t *testing.T, c cache.Cache[testIndex, string, *testObject]) {
obj, ok := c.Get(context.Background(), testIndexID, "one")
assert.False(t, ok)
assert.Nil(t, obj)
obj, ok = c.Get(context.Background(), testIndexName, "foo")
assert.False(t, ok)
assert.Nil(t, obj)
},
args: args{
ctx: context.Background(),
index: testIndexName,
key: []string{"foo"},
},
wantErr: false,
},
{
name: "invalidate after timeout",
config: cache.Config{
LastUseAge: time.Minute,
},
preparation: func(t *testing.T, c cache.Cache[testIndex, string, *testObject], s *miniredis.Miniredis) {
c.Set(context.Background(), &testObject{
ID: "one",
Name: []string{"foo", "bar"},
})
_, ok := c.Get(context.Background(), testIndexID, "one")
require.True(t, ok)
s.FastForward(2 * time.Minute)
},
assertions: func(t *testing.T, c cache.Cache[testIndex, string, *testObject]) {
obj, ok := c.Get(context.Background(), testIndexID, "one")
assert.False(t, ok)
assert.Nil(t, obj)
obj, ok = c.Get(context.Background(), testIndexName, "foo")
assert.False(t, ok)
assert.Nil(t, obj)
},
args: args{
ctx: context.Background(),
index: testIndexName,
key: []string{"foo"},
},
wantErr: false,
},
{
name: "not found",
config: cache.Config{},
preparation: func(t *testing.T, c cache.Cache[testIndex, string, *testObject], s *miniredis.Miniredis) {
c.Set(context.Background(), &testObject{
ID: "one",
Name: []string{"foo", "bar"},
})
},
assertions: func(t *testing.T, c cache.Cache[testIndex, string, *testObject]) {
obj, ok := c.Get(context.Background(), testIndexID, "one")
assert.True(t, ok)
assert.NotNil(t, obj)
obj, ok = c.Get(context.Background(), testIndexName, "foo")
assert.True(t, ok)
assert.NotNil(t, obj)
},
args: args{
ctx: context.Background(),
index: testIndexName,
key: []string{"spanac"},
},
wantErr: false,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
c, server := prepareCache(t, tt.config)
tt.preparation(t, c, server)
t.Log(server.Keys())
err := c.Invalidate(tt.args.ctx, tt.args.index, tt.args.key...)
if tt.wantErr {
require.Error(t, err)
return
}
require.NoError(t, err)
})
}
}
func Test_redisCache_Delete(t *testing.T) {
type args struct {
ctx context.Context
index testIndex
key []string
}
tests := []struct {
name string
config cache.Config
preparation func(t *testing.T, c cache.Cache[testIndex, string, *testObject], s *miniredis.Miniredis)
assertions func(t *testing.T, c cache.Cache[testIndex, string, *testObject])
args args
wantErr bool
}{
{
name: "connection error",
config: cache.Config{},
preparation: func(_ *testing.T, _ cache.Cache[testIndex, string, *testObject], s *miniredis.Miniredis) {
s.RequireAuth("foobar")
},
args: args{
ctx: context.Background(),
index: testIndexName,
key: []string{"foo"},
},
wantErr: true,
},
{
name: "no keys, noop",
config: cache.Config{},
preparation: func(t *testing.T, c cache.Cache[testIndex, string, *testObject], s *miniredis.Miniredis) {
c.Set(context.Background(), &testObject{
ID: "one",
Name: []string{"foo", "bar"},
})
},
args: args{
ctx: context.Background(),
index: testIndexID,
key: []string{},
},
wantErr: false,
},
{
name: "delete ID",
config: cache.Config{},
preparation: func(t *testing.T, c cache.Cache[testIndex, string, *testObject], s *miniredis.Miniredis) {
c.Set(context.Background(), &testObject{
ID: "one",
Name: []string{"foo", "bar"},
})
},
assertions: func(t *testing.T, c cache.Cache[testIndex, string, *testObject]) {
obj, ok := c.Get(context.Background(), testIndexID, "one")
assert.False(t, ok)
assert.Nil(t, obj)
// Get be name should still work
obj, ok = c.Get(context.Background(), testIndexName, "foo")
assert.True(t, ok)
assert.NotNil(t, obj)
},
args: args{
ctx: context.Background(),
index: testIndexID,
key: []string{"one"},
},
wantErr: false,
},
{
name: "delete name",
config: cache.Config{},
preparation: func(t *testing.T, c cache.Cache[testIndex, string, *testObject], s *miniredis.Miniredis) {
c.Set(context.Background(), &testObject{
ID: "one",
Name: []string{"foo", "bar"},
})
},
assertions: func(t *testing.T, c cache.Cache[testIndex, string, *testObject]) {
// get by ID should still work
obj, ok := c.Get(context.Background(), testIndexID, "one")
assert.True(t, ok)
assert.NotNil(t, obj)
obj, ok = c.Get(context.Background(), testIndexName, "foo")
assert.False(t, ok)
assert.Nil(t, obj)
},
args: args{
ctx: context.Background(),
index: testIndexName,
key: []string{"foo"},
},
wantErr: false,
},
{
name: "not found",
config: cache.Config{},
preparation: func(t *testing.T, c cache.Cache[testIndex, string, *testObject], s *miniredis.Miniredis) {
c.Set(context.Background(), &testObject{
ID: "one",
Name: []string{"foo", "bar"},
})
},
assertions: func(t *testing.T, c cache.Cache[testIndex, string, *testObject]) {
obj, ok := c.Get(context.Background(), testIndexID, "one")
assert.True(t, ok)
assert.NotNil(t, obj)
obj, ok = c.Get(context.Background(), testIndexName, "foo")
assert.True(t, ok)
assert.NotNil(t, obj)
},
args: args{
ctx: context.Background(),
index: testIndexName,
key: []string{"spanac"},
},
wantErr: false,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
c, server := prepareCache(t, tt.config)
tt.preparation(t, c, server)
t.Log(server.Keys())
err := c.Delete(tt.args.ctx, tt.args.index, tt.args.key...)
if tt.wantErr {
require.Error(t, err)
return
}
require.NoError(t, err)
})
}
}
func Test_redisCache_Truncate(t *testing.T) {
type args struct {
ctx context.Context
}
tests := []struct {
name string
config cache.Config
preparation func(t *testing.T, c cache.Cache[testIndex, string, *testObject], s *miniredis.Miniredis)
assertions func(t *testing.T, c cache.Cache[testIndex, string, *testObject])
args args
wantErr bool
}{
{
name: "connection error",
config: cache.Config{},
preparation: func(_ *testing.T, _ cache.Cache[testIndex, string, *testObject], s *miniredis.Miniredis) {
s.RequireAuth("foobar")
},
args: args{
ctx: context.Background(),
},
wantErr: true,
},
{
name: "ok",
config: cache.Config{},
preparation: func(t *testing.T, c cache.Cache[testIndex, string, *testObject], s *miniredis.Miniredis) {
c.Set(context.Background(), &testObject{
ID: "one",
Name: []string{"foo", "bar"},
})
c.Set(context.Background(), &testObject{
ID: "two",
Name: []string{"Hello", "World"},
})
},
assertions: func(t *testing.T, c cache.Cache[testIndex, string, *testObject]) {
obj, ok := c.Get(context.Background(), testIndexID, "one")
assert.False(t, ok)
assert.Nil(t, obj)
obj, ok = c.Get(context.Background(), testIndexName, "World")
assert.False(t, ok)
assert.Nil(t, obj)
},
args: args{
ctx: context.Background(),
},
wantErr: false,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
c, server := prepareCache(t, tt.config)
tt.preparation(t, c, server)
t.Log(server.Keys())
err := c.Truncate(tt.args.ctx)
if tt.wantErr {
require.Error(t, err)
return
}
require.NoError(t, err)
})
}
}
func prepareCache(t *testing.T, conf cache.Config) (cache.Cache[testIndex, string, *testObject], *miniredis.Miniredis) {
conf.Log = &logging.Config{
Level: "debug",
AddSource: true,
}
server := miniredis.RunT(t)
server.Select(testDB)
client := redis.NewClient(&redis.Options{
Network: "tcp",
Addr: server.Addr(),
})
t.Cleanup(func() {
client.Close()
server.Close()
})
connector := NewConnector(Config{
Enabled: true,
Network: "tcp",
Addr: server.Addr(),
})
c := NewCache[testIndex, string, *testObject](conf, connector, testDB, testIndices)
return c, server
}

27
internal/cache/connector/redis/set.lua vendored Normal file
View File

@@ -0,0 +1,27 @@
-- KEYS: [1]: object_id; [>1]: index keys.
local object_id = KEYS[1]
local object = ARGV[2]
local usage_lifetime = tonumber(ARGV[3]) -- usage based lifetime in seconds
local max_age = tonumber(ARGV[4]) -- max age liftime in seconds
redis.call("HSET", object_id,"object", object)
if usage_lifetime > 0 then
redis.call("HSET", object_id, "usage_lifetime", usage_lifetime)
-- enable usage based TTL
redis.call("EXPIRE", object_id, usage_lifetime)
if max_age > 0 then
-- set max_age to hash map for expired remove on Get
local expiry = getTime() + max_age
redis.call("HSET", object_id, "expiry", expiry)
end
elseif max_age > 0 then
-- enable max_age based TTL
redis.call("EXPIRE", object_id, max_age)
end
local n = #KEYS
local setKey = keySetKey(object_id)
for i = 2, n do -- offset to the second element to skip object_id
redis.call("SADD", setKey, KEYS[i]) -- set of all keys used for housekeeping
redis.call("SET", KEYS[i], object_id) -- key to object_id mapping
end

98
internal/cache/connector_enumer.go vendored Normal file
View File

@@ -0,0 +1,98 @@
// Code generated by "enumer -type Connector -transform snake -trimprefix Connector -linecomment -text"; DO NOT EDIT.
package cache
import (
"fmt"
"strings"
)
const _ConnectorName = "memorypostgresredis"
var _ConnectorIndex = [...]uint8{0, 0, 6, 14, 19}
const _ConnectorLowerName = "memorypostgresredis"
func (i Connector) String() string {
if i < 0 || i >= Connector(len(_ConnectorIndex)-1) {
return fmt.Sprintf("Connector(%d)", i)
}
return _ConnectorName[_ConnectorIndex[i]:_ConnectorIndex[i+1]]
}
// An "invalid array index" compiler error signifies that the constant values have changed.
// Re-run the stringer command to generate them again.
func _ConnectorNoOp() {
var x [1]struct{}
_ = x[ConnectorUnspecified-(0)]
_ = x[ConnectorMemory-(1)]
_ = x[ConnectorPostgres-(2)]
_ = x[ConnectorRedis-(3)]
}
var _ConnectorValues = []Connector{ConnectorUnspecified, ConnectorMemory, ConnectorPostgres, ConnectorRedis}
var _ConnectorNameToValueMap = map[string]Connector{
_ConnectorName[0:0]: ConnectorUnspecified,
_ConnectorLowerName[0:0]: ConnectorUnspecified,
_ConnectorName[0:6]: ConnectorMemory,
_ConnectorLowerName[0:6]: ConnectorMemory,
_ConnectorName[6:14]: ConnectorPostgres,
_ConnectorLowerName[6:14]: ConnectorPostgres,
_ConnectorName[14:19]: ConnectorRedis,
_ConnectorLowerName[14:19]: ConnectorRedis,
}
var _ConnectorNames = []string{
_ConnectorName[0:0],
_ConnectorName[0:6],
_ConnectorName[6:14],
_ConnectorName[14:19],
}
// ConnectorString retrieves an enum value from the enum constants string name.
// Throws an error if the param is not part of the enum.
func ConnectorString(s string) (Connector, error) {
if val, ok := _ConnectorNameToValueMap[s]; ok {
return val, nil
}
if val, ok := _ConnectorNameToValueMap[strings.ToLower(s)]; ok {
return val, nil
}
return 0, fmt.Errorf("%s does not belong to Connector values", s)
}
// ConnectorValues returns all values of the enum
func ConnectorValues() []Connector {
return _ConnectorValues
}
// ConnectorStrings returns a slice of all String values of the enum
func ConnectorStrings() []string {
strs := make([]string, len(_ConnectorNames))
copy(strs, _ConnectorNames)
return strs
}
// IsAConnector returns "true" if the value is listed in the enum definition. "false" otherwise
func (i Connector) IsAConnector() bool {
for _, v := range _ConnectorValues {
if i == v {
return true
}
}
return false
}
// MarshalText implements the encoding.TextMarshaler interface for Connector
func (i Connector) MarshalText() ([]byte, error) {
return []byte(i.String()), nil
}
// UnmarshalText implements the encoding.TextUnmarshaler interface for Connector
func (i *Connector) UnmarshalText(text []byte) error {
var err error
*i, err = ConnectorString(string(text))
return err
}

View File

@@ -31,22 +31,22 @@ type AutoPruneConfig struct {
Timeout time.Duration
}
func (c AutoPruneConfig) StartAutoPrune(background context.Context, pruner Pruner, name string) (close func()) {
return c.startAutoPrune(background, pruner, name, clockwork.NewRealClock())
func (c AutoPruneConfig) StartAutoPrune(background context.Context, pruner Pruner, purpose Purpose) (close func()) {
return c.startAutoPrune(background, pruner, purpose, clockwork.NewRealClock())
}
func (c *AutoPruneConfig) startAutoPrune(background context.Context, pruner Pruner, name string, clock clockwork.Clock) (close func()) {
func (c *AutoPruneConfig) startAutoPrune(background context.Context, pruner Pruner, purpose Purpose, clock clockwork.Clock) (close func()) {
if c.Interval <= 0 {
return func() {}
}
background, cancel := context.WithCancel(background)
// randomize the first interval
timer := clock.NewTimer(time.Duration(rand.Int63n(int64(c.Interval))))
go c.pruneTimer(background, pruner, name, timer)
go c.pruneTimer(background, pruner, purpose, timer)
return cancel
}
func (c *AutoPruneConfig) pruneTimer(background context.Context, pruner Pruner, name string, timer clockwork.Timer) {
func (c *AutoPruneConfig) pruneTimer(background context.Context, pruner Pruner, purpose Purpose, timer clockwork.Timer) {
defer func() {
if !timer.Stop() {
<-timer.Chan()
@@ -58,9 +58,9 @@ func (c *AutoPruneConfig) pruneTimer(background context.Context, pruner Pruner,
case <-background.Done():
return
case <-timer.Chan():
timer.Reset(c.Interval)
err := c.doPrune(background, pruner)
logging.OnError(err).WithField("name", name).Error("cache auto prune")
logging.OnError(err).WithField("purpose", purpose).Error("cache auto prune")
timer.Reset(c.Interval)
}
}
}

View File

@@ -30,7 +30,7 @@ func TestAutoPruneConfig_startAutoPrune(t *testing.T) {
called: make(chan struct{}),
}
clock := clockwork.NewFakeClock()
close := c.startAutoPrune(ctx, &pruner, "foo", clock)
close := c.startAutoPrune(ctx, &pruner, PurposeAuthzInstance, clock)
defer close()
clock.Advance(time.Second)

82
internal/cache/purpose_enumer.go vendored Normal file
View File

@@ -0,0 +1,82 @@
// Code generated by "enumer -type Purpose -transform snake -trimprefix Purpose"; DO NOT EDIT.
package cache
import (
"fmt"
"strings"
)
const _PurposeName = "unspecifiedauthz_instancemilestones"
var _PurposeIndex = [...]uint8{0, 11, 25, 35}
const _PurposeLowerName = "unspecifiedauthz_instancemilestones"
func (i Purpose) String() string {
if i < 0 || i >= Purpose(len(_PurposeIndex)-1) {
return fmt.Sprintf("Purpose(%d)", i)
}
return _PurposeName[_PurposeIndex[i]:_PurposeIndex[i+1]]
}
// An "invalid array index" compiler error signifies that the constant values have changed.
// Re-run the stringer command to generate them again.
func _PurposeNoOp() {
var x [1]struct{}
_ = x[PurposeUnspecified-(0)]
_ = x[PurposeAuthzInstance-(1)]
_ = x[PurposeMilestones-(2)]
}
var _PurposeValues = []Purpose{PurposeUnspecified, PurposeAuthzInstance, PurposeMilestones}
var _PurposeNameToValueMap = map[string]Purpose{
_PurposeName[0:11]: PurposeUnspecified,
_PurposeLowerName[0:11]: PurposeUnspecified,
_PurposeName[11:25]: PurposeAuthzInstance,
_PurposeLowerName[11:25]: PurposeAuthzInstance,
_PurposeName[25:35]: PurposeMilestones,
_PurposeLowerName[25:35]: PurposeMilestones,
}
var _PurposeNames = []string{
_PurposeName[0:11],
_PurposeName[11:25],
_PurposeName[25:35],
}
// PurposeString retrieves an enum value from the enum constants string name.
// Throws an error if the param is not part of the enum.
func PurposeString(s string) (Purpose, error) {
if val, ok := _PurposeNameToValueMap[s]; ok {
return val, nil
}
if val, ok := _PurposeNameToValueMap[strings.ToLower(s)]; ok {
return val, nil
}
return 0, fmt.Errorf("%s does not belong to Purpose values", s)
}
// PurposeValues returns all values of the enum
func PurposeValues() []Purpose {
return _PurposeValues
}
// PurposeStrings returns a slice of all String values of the enum
func PurposeStrings() []string {
strs := make([]string, len(_PurposeNames))
copy(strs, _PurposeNames)
return strs
}
// IsAPurpose returns "true" if the value is listed in the enum definition. "false" otherwise
func (i Purpose) IsAPurpose() bool {
for _, v := range _PurposeValues {
if i == v {
return true
}
}
return false
}