go-pkg/asynq
2
0
Fork 0
mirror of https://github.com/hibiken/asynq.git synced 2025-10-26 11:16:12 +08:00
Code Issues Projects Releases Wiki Activity

chore(): set scheduler_history TTL

Browse Source
This commit is contained in:
youngxu
2022-07-13 14:48:04 +08:00
committed by 徐胖
parent a7e533f268
commit f27a72c2f2
1 changed files with 9 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
internal/rdb/rdb.go
View File

@@ -13,10 +13,11 @@ import (
"github.com/go-redis/redis/v8" "github.com/go-redis/redis/v8"
"github.com/google/uuid" "github.com/google/uuid"
"github.com/spf13/cast"
"github.com/hibiken/asynq/internal/base" "github.com/hibiken/asynq/internal/base"
"github.com/hibiken/asynq/internal/errors" "github.com/hibiken/asynq/internal/errors"
"github.com/hibiken/asynq/internal/timeutil" "github.com/hibiken/asynq/internal/timeutil"
"github.com/spf13/cast"
) )
const statsTTL = 90 * 24 * time.Hour // 90 days const statsTTL = 90 * 24 * time.Hour // 90 days
@@ -1471,14 +1472,18 @@ func (r *RDB) PublishCancelation(id string) error {
// ARGV[1] -> enqueued_at timestamp // ARGV[1] -> enqueued_at timestamp
// ARGV[2] -> serialized SchedulerEnqueueEvent data // ARGV[2] -> serialized SchedulerEnqueueEvent data
// ARGV[3] -> max number of events to be persisted // ARGV[3] -> max number of events to be persisted
// ARGV[4] -> history expiration timestamp
var recordSchedulerEnqueueEventCmd = redis.NewScript(` var recordSchedulerEnqueueEventCmd = redis.NewScript(`
redis.call("ZREMRANGEBYRANK", KEYS[1], 0, -ARGV[3]) redis.call("ZREMRANGEBYRANK", KEYS[1], 0, -ARGV[3])
redis.call("ZADD", KEYS[1], ARGV[1], ARGV[2]) redis.call("ZADD", KEYS[1], ARGV[1], ARGV[2])
redis.call("EXPIREAT", KEYS[1], ARGV[4])
return redis.status_reply("OK")`) return redis.status_reply("OK")`)
// Maximum number of enqueue events to store per entry. // Maximum number of enqueue events to store per entry.
const maxEvents = 1000 const maxEvents = 1000
const schedulerHistoryExpirationInDays = 90
// RecordSchedulerEnqueueEvent records the time when the given task was enqueued. // RecordSchedulerEnqueueEvent records the time when the given task was enqueued.
func (r *RDB) RecordSchedulerEnqueueEvent(entryID string, event *base.SchedulerEnqueueEvent) error { func (r *RDB) RecordSchedulerEnqueueEvent(entryID string, event *base.SchedulerEnqueueEvent) error {
var op errors.Op = "rdb.RecordSchedulerEnqueueEvent" var op errors.Op = "rdb.RecordSchedulerEnqueueEvent"
@@ -1487,6 +1492,8 @@ func (r *RDB) RecordSchedulerEnqueueEvent(entryID string, event *base.SchedulerE
if err != nil { if err != nil {
return errors.E(op, errors.Internal, fmt.Sprintf("cannot encode scheduler enqueue event: %v", err)) return errors.E(op, errors.Internal, fmt.Sprintf("cannot encode scheduler enqueue event: %v", err))
} }
now := r.clock.Now()
expireAt := now.AddDate(0, 0, schedulerHistoryExpirationInDays)
keys := []string{ keys := []string{
base.SchedulerHistoryKey(entryID), base.SchedulerHistoryKey(entryID),
} }
@@ -1494,6 +1501,7 @@ func (r *RDB) RecordSchedulerEnqueueEvent(entryID string, event *base.SchedulerE
event.EnqueuedAt.Unix(), event.EnqueuedAt.Unix(),
data, data,
maxEvents, maxEvents,
expireAt.Unix(),
} }
return r.runScript(ctx, op, recordSchedulerEnqueueEventCmd, keys, argv...) return r.runScript(ctx, op, recordSchedulerEnqueueEventCmd, keys, argv...)
} }