From 4143a37056254a1a16947fee9612ece5a6da602e Mon Sep 17 00:00:00 2001 From: Martin Angers Date: Tue, 14 Dec 2021 16:30:26 -0500 Subject: [PATCH] Fix redis scan keys issue for live queries (#3107) --- changes/issue-3065-improve-live-queries | 1 + cmd/fleet/serve.go | 6 - docs/02-Deploying/03-Configuration.md | 3 +- server/config/config.go | 2 +- server/errorstore/errors.go | 29 +- server/errorstore/errors_test.go | 24 +- server/live_query/live_query_test.go | 49 ++++ server/live_query/redis_live_query.go | 299 ++++++++++++--------- server/live_query/redis_live_query_test.go | 73 +---- 9 files changed, 261 insertions(+), 225 deletions(-) create mode 100644 changes/issue-3065-improve-live-queries diff --git a/changes/issue-3065-improve-live-queries b/changes/issue-3065-improve-live-queries new file mode 100644 index 0000000000..4d50d80e99 --- /dev/null +++ b/changes/issue-3065-improve-live-queries @@ -0,0 +1 @@ +* Reduce load on Redis for live queries when many keys exist. diff --git a/cmd/fleet/serve.go b/cmd/fleet/serve.go index 74e206d477..a931f95b13 100644 --- a/cmd/fleet/serve.go +++ b/cmd/fleet/serve.go @@ -255,12 +255,6 @@ the way that the Fleet server works. ds = cached_mysql.New(ds) resultStore := pubsub.NewRedisQueryResults(redisPool, config.Redis.DuplicateResults) liveQueryStore := live_query.NewRedisLiveQuery(redisPool) - if err := liveQueryStore.MigrateKeys(); err != nil { - level.Info(logger).Log( - "err", err, - "msg", "failed to migrate live query redis keys", - ) - } ssoSessionStore := sso.NewSessionStore(redisPool) osqueryLogger, err := logging.New(config, logger) diff --git a/docs/02-Deploying/03-Configuration.md b/docs/02-Deploying/03-Configuration.md index 23112ff3e9..2fd06cc3ab 100644 --- a/docs/02-Deploying/03-Configuration.md +++ b/docs/02-Deploying/03-Configuration.md @@ -1108,7 +1108,8 @@ Whether or not to log the welcome banner. ##### logging_error_retention_period The amount of time to keep an error. Unique instances of errors are stored temporarily to help -with troubleshooting, this setting controls that duration. +with troubleshooting, this setting controls that duration. Set to 0 to keep them without expiration, +and a negative value to disable storage of errors in Redis. - Default value: 24h - Environment variable: `FLEET_LOGGING_ERROR_RETENTION_PERIOD` diff --git a/server/config/config.go b/server/config/config.go index a987ca0201..94c9ae0c40 100644 --- a/server/config/config.go +++ b/server/config/config.go @@ -438,7 +438,7 @@ func (man Manager) addConfigs() { man.addConfigBool("logging.disable_banner", false, "Disable startup banner") man.addConfigDuration("logging.error_retention_period", 24*time.Hour, - "Amount of time to keep errors") + "Amount of time to keep errors, 0 means no expiration, < 0 means disable storage of errors") // Firehose man.addConfigString("firehose.region", "", "AWS Region to use") diff --git a/server/errorstore/errors.go b/server/errorstore/errors.go index b6a13e1ca7..da3589e649 100644 --- a/server/errorstore/errors.go +++ b/server/errorstore/errors.go @@ -51,7 +51,9 @@ func NewHandler(ctx context.Context, pool fleet.RedisPool, logger kitlog.Logger, logger: logger, ttl: ttl, } - runHandler(ctx, eh) + if ttl >= 0 { + runHandler(ctx, eh) + } // Clear out any records that exist. // Temporary mitigation for #3065. @@ -72,7 +74,10 @@ func newTestHandler(ctx context.Context, pool fleet.RedisPool, logger kitlog.Log testOnStart: onStart, testOnStore: onStore, } - runHandler(ctx, eh) + + if ttl >= 0 { + runHandler(ctx, eh) + } return eh } @@ -224,12 +229,6 @@ func (h *Handler) handleErrors(ctx context.Context) { } func (h *Handler) storeError(ctx context.Context, err error) { - // Skip storing errors due to SCAN issues with Redis (see #3065). - // if true here because otherwise we get linting errors for unreachable code. - if true { - return - } - errorHash, errorJson, err := hashAndMarshalError(err) if err != nil { level.Error(h.logger).Log("err", err, "msg", "hashErr failed") @@ -243,11 +242,17 @@ func (h *Handler) storeError(ctx context.Context, err error) { conn := redis.ConfigureDoer(h.pool, h.pool.Get()) defer conn.Close() - secs := int(h.ttl.Seconds()) - if secs <= 0 { - secs = 1 // SET EX fails if ttl is <= 0 + var args redigo.Args + args = args.Add(jsonKey, errorJson) + if h.ttl > 0 { + secs := int(h.ttl.Seconds()) + if secs <= 0 { + secs = 1 // SET EX fails if ttl is <= 0 + } + args = args.Add("EX", secs) } - if _, err := conn.Do("SET", jsonKey, errorJson, "EX", secs); err != nil { + + if _, err := conn.Do("SET", args...); err != nil { level.Error(h.logger).Log("err", err, "msg", "redis SET failed") if h.testOnStore != nil { h.testOnStore(err) diff --git a/server/errorstore/errors_test.go b/server/errorstore/errors_test.go index 729212af1b..099ea9cdb2 100644 --- a/server/errorstore/errors_test.go +++ b/server/errorstore/errors_test.go @@ -152,9 +152,6 @@ func TestUnwrapAll(t *testing.T) { } func TestErrorHandler(t *testing.T) { - // Skipped until error publishing is re-enabled. - t.Skip() - t.Run("works if the error handler is down", func(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) cancel() // cancel immediately @@ -176,6 +173,24 @@ func TestErrorHandler(t *testing.T) { } }) + t.Run("works if the error storage is disabled", func(t *testing.T) { + eh := newTestHandler(context.Background(), nil, kitlog.NewNopLogger(), -1, nil, nil) + + doneCh := make(chan struct{}) + go func() { + eh.Store(pkgErrors.New("test")) + close(doneCh) + }() + + // should not even block in the call to Store as there is no handler running + ticker := time.NewTicker(1 * time.Second) + select { + case <-doneCh: + case <-ticker.C: + t.FailNow() + } + }) + wd, err := os.Getwd() require.NoError(t, err) wd = regexp.QuoteMeta(wd) @@ -306,9 +321,6 @@ func testErrorHandlerCollectsDifferentErrors(t *testing.T, pool fleet.RedisPool, } func TestHttpHandler(t *testing.T) { - // Skipped until error publishing is re-enabled. - t.Skip() - pool := redistest.SetupRedis(t, false, false, false) ctx, cancelFunc := context.WithCancel(context.Background()) defer cancelFunc() diff --git a/server/live_query/live_query_test.go b/server/live_query/live_query_test.go index 922c9d08cf..9cfd961bc8 100644 --- a/server/live_query/live_query_test.go +++ b/server/live_query/live_query_test.go @@ -3,7 +3,9 @@ package live_query import ( "testing" + "github.com/fleetdm/fleet/v4/server/datastore/redis" "github.com/fleetdm/fleet/v4/server/fleet" + redigo "github.com/gomodule/redigo/redis" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -12,6 +14,8 @@ var testFunctions = [...]func(*testing.T, fleet.LiveQueryStore){ testLiveQuery, testLiveQueryNoTargets, testLiveQueryStopQuery, + testLiveQueryExpiredQuery, + testLiveQueryOnlyExpired, } func testLiveQuery(t *testing.T, store fleet.LiveQueryStore) { @@ -86,3 +90,48 @@ func testLiveQueryStopQuery(t *testing.T, store fleet.LiveQueryStore) { require.NoError(t, err) assert.Len(t, queries, 1) } + +func testLiveQueryExpiredQuery(t *testing.T, store fleet.LiveQueryStore) { + oldModulo := cleanupExpiredQueriesModulo + cleanupExpiredQueriesModulo = 1 // run the cleanup each time + t.Cleanup(func() { cleanupExpiredQueriesModulo = oldModulo }) + + require.NoError(t, store.RunQuery("test", "select 1", []uint{1})) + + // simulate a "test2" live query that has expired but is still in the set + pool := store.(*redisLiveQuery).pool + conn := redis.ConfigureDoer(pool, pool.Get()) + defer conn.Close() + _, err := conn.Do("SADD", activeQueriesKey, "test2") + require.NoError(t, err) + + queries, err := store.QueriesForHost(1) + require.NoError(t, err) + assert.Len(t, queries, 1) + assert.Equal(t, map[string]string{"test": "select 1"}, queries) + + activeNames, err := redigo.Strings(conn.Do("SMEMBERS", activeQueriesKey)) + require.NoError(t, err) + require.Equal(t, []string{"test"}, activeNames) +} + +func testLiveQueryOnlyExpired(t *testing.T, store fleet.LiveQueryStore) { + oldModulo := cleanupExpiredQueriesModulo + cleanupExpiredQueriesModulo = 1 // run the cleanup each time + t.Cleanup(func() { cleanupExpiredQueriesModulo = oldModulo }) + + // simulate a "test" live query that has expired but is still in the set + pool := store.(*redisLiveQuery).pool + conn := redis.ConfigureDoer(pool, pool.Get()) + defer conn.Close() + _, err := conn.Do("SADD", activeQueriesKey, "test") + require.NoError(t, err) + + queries, err := store.QueriesForHost(1) + require.NoError(t, err) + assert.Len(t, queries, 0) + + activeNames, err := redigo.Strings(conn.Do("SMEMBERS", activeQueriesKey)) + require.NoError(t, err) + require.Len(t, activeNames, 0) +} diff --git a/server/live_query/redis_live_query.go b/server/live_query/redis_live_query.go index 53549b7a87..be1cb3fe8d 100644 --- a/server/live_query/redis_live_query.go +++ b/server/live_query/redis_live_query.go @@ -1,4 +1,4 @@ -// package live_query implements an interface for storing and +// Package live_query implements an interface for storing and // retrieving live queries. // // Design @@ -7,11 +7,10 @@ // targeting information. This key has a known prefix, and the data // is a bitfield representing _all_ the hosts in fleet. // -// In this model, a live query creation is a few redis writes. While a -// host checkin needs to scan the keyspace for matching key, and then -// fetch the bitfield value for their id. While this scan might be -// expensive, this model fits very well with having a lot of hosts and -// very few live queries. +// In this model, a live query creation is a few redis writes. While a host +// checkin needs to scan the keys stored in a set representing all active live +// queries, and then fetch the bitfield value for their id. This model fits +// very well with having a lot of hosts and very few live queries. // // A contrasting model, for the case of fewer hosts, but a lot of live // queries, is to have a set per host. In this case, the LQ is pushed @@ -24,18 +23,25 @@ // // Implementation // -// As mentioned in the Design section, there are two keys for each -// live query: the bitfield and the SQL of the query: +// As mentioned in the Design section, there are three keys for each +// live query: the bitfield, the SQL of the query and the set containing +// the IDs of all active live queries: // // livequery: is the bitfield that indicates the hosts // sql:livequery: is the SQL of the query. +// livequery:active is the set containing the active live query IDs // -// Both have an expiration, and is the campaign ID of the query. To make -// efficient use of Redis Cluster (without impacting standalone Redis), the -// is stored in braces (hash tags, e.g. livequery:{1} and -// sql:livequery:{1}), so that the two keys for the same are always stored -// on the same node (as they hash to the same cluster slot). See -// https://redis.io/topics/cluster-spec#keys-hash-tags for details. +// Both the bitfield and sql keys have an expiration, and is the campaign +// ID of the query. To make efficient use of Redis Cluster (without impacting +// standalone Redis), the is stored in braces (hash tags, e.g. +// livequery:{1} and sql:livequery:{1}), so that the two keys for the same +// are always stored on the same node (as they hash to the same cluster slot). +// See https://redis.io/topics/cluster-spec#keys-hash-tags for details. +// +// It is a noted downside that the active live queries set will necessarily +// live on a single node in cluster mode (a "hot key"), and that node will see +// increased activity due to that. Should that become a significant problem, an +// alternative approach will be required. // package live_query @@ -48,14 +54,14 @@ import ( "github.com/fleetdm/fleet/v4/server/datastore/redis" "github.com/fleetdm/fleet/v4/server/fleet" redigo "github.com/gomodule/redigo/redis" - "github.com/mna/redisc" ) const ( - bitsInByte = 8 - queryKeyPrefix = "livequery:" - sqlKeyPrefix = "sql:" - queryExpiration = 7 * 24 * time.Hour + bitsInByte = 8 + queryKeyPrefix = "livequery:" + sqlKeyPrefix = "sql:" + activeQueriesKey = "livequery:active" + queryExpiration = 7 * 24 * time.Hour ) type redisLiveQuery struct { @@ -63,7 +69,7 @@ type redisLiveQuery struct { pool fleet.RedisPool } -// NewRedisQueryResults creats a new Redis implementation of the +// NewRedisQueryResults creates a new Redis implementation of the // QueryResultStore interface using the provided Redis connection pool. func NewRedisLiveQuery(pool fleet.RedisPool) *redisLiveQuery { return &redisLiveQuery{pool: pool} @@ -92,138 +98,84 @@ func extractTargetKeyName(key string) string { return name } -// MigrateKeys migrates keys using a deprecated format to the new format. It -// should be called at startup and never after that, so for this reason it is -// not added to the fleet.LiveQueryStore interface. -func (r *redisLiveQuery) MigrateKeys() error { - qkeys, err := redis.ScanKeys(r.pool, queryKeyPrefix+"*", 100) - if err != nil { - return err - } - - // identify which of those keys are in a deprecated format - var oldKeys []string - for _, key := range qkeys { - name := extractTargetKeyName(key) - if !strings.Contains(key, "{"+name+"}") { - // add the corresponding sql key to the list - oldKeys = append(oldKeys, key, sqlKeyPrefix+key) - } - } - - keysBySlot := redis.SplitKeysBySlot(r.pool, oldKeys...) - for _, keys := range keysBySlot { - if err := migrateBatchKeys(r.pool, keys); err != nil { - return err - } - } - return nil -} - -func migrateBatchKeys(pool fleet.RedisPool, keys []string) error { - readConn := pool.Get() - defer readConn.Close() - - writeConn := pool.Get() - defer writeConn.Close() - - // use a retry conn so that we follow MOVED redirections in a Redis Cluster, - // as we will attempt to write new keys which may not belong to the same - // cluster slot. It returns an error if writeConn is not a redis cluster - // connection, in which case we simply continue with the standalone Redis - // writeConn. - if rc, err := redisc.RetryConn(writeConn, 3, 100*time.Millisecond); err == nil { - writeConn = rc - } - - // using a straightforward "read one, write one" approach as this is meant to - // run at startup, not on a hot path, and we expect a relatively small number - // of queries vs hosts (as documented in the design comment at the top). - for _, key := range keys { - s, err := redigo.String(readConn.Do("GET", key)) - if err != nil { - if err == redigo.ErrNil { - // key may have expired since the scan, ignore - continue - } - return err - } - - var newKey string - if strings.HasPrefix(key, sqlKeyPrefix) { - name := extractTargetKeyName(strings.TrimPrefix(key, sqlKeyPrefix)) - _, newKey = generateKeys(name) - } else { - name := extractTargetKeyName(key) - newKey, _ = generateKeys(name) - } - if _, err := writeConn.Do("SET", newKey, s, "EX", queryExpiration.Seconds()); err != nil { - return err - } - - // best-effort deletion of the old key, ignore error - readConn.Do("DEL", key) - } - return nil -} - +// RunQuery stores the live query information in ephemeral storage for the +// duration of the query or its TTL. Note that hostIDs *must* be sorted +// in ascending order. func (r *redisLiveQuery) RunQuery(name, sql string, hostIDs []uint) error { if len(hostIDs) == 0 { return errors.New("no hosts targeted") } - conn := r.pool.Get() - defer conn.Close() - - // Map the targeted host IDs to a bitfield. Store targets in one key and SQL - // in another. - targetKey, sqlKey := generateKeys(name) - targets := mapBitfield(hostIDs) - - // Ensure to set SQL first or else we can end up in a weird state in which a - // client reads that the query exists but cannot look up the SQL. - err := conn.Send("SET", sqlKey, sql, "EX", queryExpiration.Seconds()) - if err != nil { - return fmt.Errorf("set sql: %w", err) + // store the sql and targeted hosts information + if err := r.storeQueryInfo(name, sql, hostIDs); err != nil { + return fmt.Errorf("store query info: %w", err) } - _, err = conn.Do("SET", targetKey, targets, "EX", queryExpiration.Seconds()) - if err != nil { - return fmt.Errorf("set targets: %w", err) + + // store name (campaign id) into the active live queries set + if err := r.storeQueryNames(name); err != nil { + return fmt.Errorf("store query name: %w", err) } return nil } func (r *redisLiveQuery) StopQuery(name string) error { - conn := redis.ConfigureDoer(r.pool, r.pool.Get()) - defer conn.Close() + // remove the sql and targeted hosts keys + if err := r.removeQueryInfo(name); err != nil { + return fmt.Errorf("remove query info: %w", err) + } - targetKey, sqlKey := generateKeys(name) - if _, err := conn.Do("DEL", targetKey, sqlKey); err != nil { - return fmt.Errorf("del query keys: %w", err) + // remove name (campaign id) from the livequery set + if err := r.removeQueryNames(name); err != nil { + return fmt.Errorf("remove query name: %w", err) } return nil } +// this is a variable so it can be changed in tests +var cleanupExpiredQueriesModulo int64 = 10 + func (r *redisLiveQuery) QueriesForHost(hostID uint) (map[string]string, error) { // Get keys for active queries - queryKeys, err := redis.ScanKeys(r.pool, queryKeyPrefix+"*", 100) + names, err := r.loadActiveQueryNames() if err != nil { - return nil, fmt.Errorf("scan active queries: %w", err) + return nil, fmt.Errorf("load active queries: %w", err) } - keysBySlot := redis.SplitKeysBySlot(r.pool, queryKeys...) + // convert the query name (campaign id) to the key name + for i, name := range names { + tkey, _ := generateKeys(name) + names[i] = tkey + } + + keysBySlot := redis.SplitKeysBySlot(r.pool, names...) queries := make(map[string]string) + expired := make(map[string]struct{}) for _, qkeys := range keysBySlot { - if err := r.collectBatchQueriesForHost(hostID, qkeys, queries); err != nil { + if err := r.collectBatchQueriesForHost(hostID, qkeys, queries, expired); err != nil { return nil, err } } + + if len(expired) > 0 { + // a certain percentage of the time so that we don't overwhelm redis with a + // bunch of similar deletion commands at the same time, clean up the + // expired queries. + if time.Now().UnixNano()%cleanupExpiredQueriesModulo == 0 { + names := make([]string, 0, len(expired)) + for k := range expired { + names = append(names, k) + } + // ignore error, best effort removal + _ = r.removeQueryNames(names...) + } + } + return queries, nil } -func (r *redisLiveQuery) collectBatchQueriesForHost(hostID uint, queryKeys []string, queriesByHost map[string]string) error { +func (r *redisLiveQuery) collectBatchQueriesForHost(hostID uint, queryKeys []string, queriesByHost map[string]string, expiredQueries map[string]struct{}) error { conn := redis.ReadOnlyConn(r.pool, r.pool.Get()) defer conn.Close() @@ -252,6 +204,9 @@ func (r *redisLiveQuery) collectBatchQueriesForHost(hostID uint, queryKeys []str for _, key := range queryKeys { name := extractTargetKeyName(key) + // the result of GETBIT will not fail if the key does not exist, it will + // just return 0, so it can't be used to detect if the livequery still + // exists. targeted, err := redigo.Int(conn.Receive()) if err != nil { return fmt.Errorf("receive target: %w", err) @@ -262,12 +217,15 @@ func (r *redisLiveQuery) collectBatchQueriesForHost(hostID uint, queryKeys []str // the pipeline. sql, err := redigo.String(conn.Receive()) if err != nil { - // Not being able to get the sql for a matched query could mean things - // have ended up in a weird state. Or it could be that the query was - // stopped since we did the key scan. In any case, attempt to clean - // up here. - _ = r.StopQuery(name) - return fmt.Errorf("receive sql: %w", err) + if err != redigo.ErrNil { + return fmt.Errorf("receive sql: %w", err) + } + + // It is possible the livequery key has expired but was still in the set + // - handle this gracefully by collecting the keys to remove them from + // the set and keep going. + expiredQueries[name] = struct{}{} + continue } if targeted == 0 { @@ -290,9 +248,81 @@ func (r *redisLiveQuery) QueryCompletedByHost(name string, hostID uint) error { return fmt.Errorf("setbit query key: %w", err) } + // NOTE(mna): we could remove the query here if all bits are now off, meaning + // that all hosts have completed this query, but the BITCOUNT command can be + // costly on large strings and we will have quite large ones. This should not be + // needed anyway as StopQuery appears to be called every time a campaign is + // run (see svc.CompleteCampaign). + return nil } +func (r *redisLiveQuery) storeQueryInfo(name, sql string, hostIDs []uint) error { + conn := r.pool.Get() + defer conn.Close() + + // Map the targeted host IDs to a bitfield. Store targets in one key and SQL + // in another. + targetKey, sqlKey := generateKeys(name) + targets := mapBitfield(hostIDs) + + // Ensure to set SQL first or else we can end up in a weird state in which a + // client reads that the query exists but cannot look up the SQL. + err := conn.Send("SET", sqlKey, sql, "EX", queryExpiration.Seconds()) + if err != nil { + return fmt.Errorf("set sql: %w", err) + } + _, err = conn.Do("SET", targetKey, targets, "EX", queryExpiration.Seconds()) + if err != nil { + return fmt.Errorf("set targets: %w", err) + } + return nil +} + +func (r *redisLiveQuery) storeQueryNames(names ...string) error { + conn := redis.ConfigureDoer(r.pool, r.pool.Get()) + defer conn.Close() + + var args redigo.Args + args = args.Add(activeQueriesKey) + args = args.AddFlat(names) + _, err := conn.Do("SADD", args...) + return err +} + +func (r *redisLiveQuery) removeQueryInfo(name string) error { + conn := redis.ConfigureDoer(r.pool, r.pool.Get()) + defer conn.Close() + + targetKey, sqlKey := generateKeys(name) + if _, err := conn.Do("DEL", targetKey, sqlKey); err != nil { + return fmt.Errorf("del query keys: %w", err) + } + return nil +} + +func (r *redisLiveQuery) removeQueryNames(names ...string) error { + conn := redis.ConfigureDoer(r.pool, r.pool.Get()) + defer conn.Close() + + var args redigo.Args + args = args.Add(activeQueriesKey) + args = args.AddFlat(names) + _, err := conn.Do("SREM", args...) + return err +} + +func (r *redisLiveQuery) loadActiveQueryNames() ([]string, error) { + conn := redis.ConfigureDoer(r.pool, r.pool.Get()) + defer conn.Close() + + names, err := redigo.Strings(conn.Do("SMEMBERS", activeQueriesKey)) + if err != nil && err != redigo.ErrNil { + return nil, err + } + return names, nil +} + // mapBitfield takes the given host IDs and maps them into a bitfield compatible // with Redis. It is expected that the input IDs are in ascending order. func mapBitfield(hostIDs []uint) []byte { @@ -300,6 +330,21 @@ func mapBitfield(hostIDs []uint) []byte { return []byte{} } + // NOTE(mna): note that this is efficient storage if the host IDs are mostly + // sequential and starting from 1, e.g. as in a newly created database. If + // there's substantial churn in hosts (e.g. some are coming on and off) or + // for some reason the auto_increment had to be bumped (e.g. it increments with + // failed inserts, even if there's an "on duplicate" clause), then it could get + // quite inefficient. If the id gets, say, to 10M then the bitfield will take + // over 1MB, even if there are only 100K hosts - at which point it would + // likely become more efficient to store a set of host IDs (without any + // redis-internal storage optimization, that would be 100K * 4 bytes = + // ~380KB). This large bitfield usage of memory would even be true if there + // was only one host selected in the query, should that host be one of the + // high IDs. Something to keep in mind if at some point we have reports of + // unexpectedly large redis memory usage, as that storage is repeated for + // each live query. + // As the input IDs are in ascending order, we get two optimizations here: // 1. We can calculate the length of the bitfield necessary by using the // last ID in the slice. Then we allocate the slice all at once. diff --git a/server/live_query/redis_live_query_test.go b/server/live_query/redis_live_query_test.go index 0471c64492..45daf94890 100644 --- a/server/live_query/redis_live_query_test.go +++ b/server/live_query/redis_live_query_test.go @@ -2,15 +2,10 @@ package live_query import ( "testing" - "time" - "github.com/fleetdm/fleet/v4/server/datastore/redis" "github.com/fleetdm/fleet/v4/server/datastore/redis/redistest" "github.com/fleetdm/fleet/v4/server/test" - redigo "github.com/gomodule/redigo/redis" - "github.com/mna/redisc" "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" ) func TestRedisLiveQuery(t *testing.T) { @@ -29,74 +24,8 @@ func TestRedisLiveQuery(t *testing.T) { } } -func TestMigrateKeys(t *testing.T) { - startKeys := map[string]string{ - "unrelated": "u", - queryKeyPrefix + "a": "a", - sqlKeyPrefix + queryKeyPrefix + "a": "sqla", - queryKeyPrefix + "b": "b", - queryKeyPrefix + "{c}": "c", - sqlKeyPrefix + queryKeyPrefix + "{c}": "sqlc", - } - - endKeys := map[string]string{ - "unrelated": "u", - queryKeyPrefix + "{a}": "a", - sqlKeyPrefix + queryKeyPrefix + "{a}": "sqla", - queryKeyPrefix + "{b}": "b", - queryKeyPrefix + "{c}": "c", - sqlKeyPrefix + queryKeyPrefix + "{c}": "sqlc", - } - - runTest := func(t *testing.T, store *redisLiveQuery) { - conn := store.pool.Get() - defer conn.Close() - if rc, err := redisc.RetryConn(conn, 3, 100*time.Millisecond); err == nil { - conn = rc - } - - for k, v := range startKeys { - _, err := conn.Do("SET", k, v) - require.NoError(t, err) - } - - err := store.MigrateKeys() - require.NoError(t, err) - - got := make(map[string]string) - err = redis.EachNode(store.pool, false, func(conn redigo.Conn) error { - keys, err := redigo.Strings(conn.Do("KEYS", "*")) - if err != nil { - return err - } - - for _, k := range keys { - v, err := redigo.String(conn.Do("GET", k)) - if err != nil { - return err - } - got[k] = v - } - return nil - }) - require.NoError(t, err) - - require.EqualValues(t, endKeys, got) - } - - t.Run("standalone", func(t *testing.T) { - store := setupRedisLiveQuery(t, false) - runTest(t, store) - }) - - t.Run("cluster", func(t *testing.T) { - store := setupRedisLiveQuery(t, true) - runTest(t, store) - }) -} - func setupRedisLiveQuery(t *testing.T, cluster bool) *redisLiveQuery { - pool := redistest.SetupRedis(t, cluster, false, false) + pool := redistest.SetupRedis(t, cluster, true, true) return NewRedisLiveQuery(pool) }