Fix redis scan keys issue for live queries (#3107)
This commit is contained in:
+17
-12
@@ -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)
|
||||
|
||||
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user