Skip to content

fix: use configurable timezone for daily stats keys #1099

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Xio-Shark wants to merge 1 commit into
base: master
Choose a base branch
from
+45 −12
Open

fix: use configurable timezone for daily stats keys #1099

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 4 additions & 2 deletions internal/base/base.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -167,13 +167,15 @@ func FailedTotalKey(qname string) string {
}

// ProcessedKey returns a redis key for processed count for the given day for the queue.
// The caller is responsible for converting t to the desired timezone before calling this function.
func ProcessedKey(qname string, t time.Time) string {
return QueueKeyPrefix(qname) + "processed:" + t.UTC().Format("2006-01-02")
return QueueKeyPrefix(qname) + "processed:" + t.Format("2006-01-02")
}

// FailedKey returns a redis key for failure count for the given day for the queue.
// The caller is responsible for converting t to the desired timezone before calling this function.
func FailedKey(qname string, t time.Time) string {
return QueueKeyPrefix(qname) + "failed:" + t.UTC().Format("2006-01-02")
return QueueKeyPrefix(qname) + "failed:" + t.Format("2006-01-02")
}

// ServerInfoKey returns a redis key for process info.
Expand Down
16 changes: 16 additions & 0 deletions internal/base/base_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -209,6 +209,10 @@ func TestFailedTotalKey(t *testing.T) {
}

func TestProcessedKey(t *testing.T) {
shanghaiLoc, err := time.LoadLocation("Asia/Shanghai")
if err != nil {
t.Skip("could not load Asia/Shanghai timezone, skipping")
}
tests := []struct {
qname string
input time.Time
Expand All @@ -217,6 +221,11 @@ func TestProcessedKey(t *testing.T) {
{"default", time.Date(2019, 11, 14, 10, 30, 1, 1, time.UTC), "asynq:{default}:processed:2019-11-14"},
{"critical", time.Date(2020, 12, 1, 1, 0, 1, 1, time.UTC), "asynq:{critical}:processed:2020-12-01"},
{"default", time.Date(2020, 1, 6, 15, 02, 1, 1, time.UTC), "asynq:{default}:processed:2020-01-06"},
// Timezone-aware test: 2024-03-22 07:30 CST = 2024-03-21 23:30 UTC
// With the fix, the key should use the local date (03-22), not UTC date (03-21).
{"default", time.Date(2024, 3, 22, 7, 30, 0, 0, shanghaiLoc), "asynq:{default}:processed:2024-03-22"},
// Edge case: exactly midnight CST = 16:00 previous day UTC
{"default", time.Date(2024, 3, 22, 0, 0, 0, 0, shanghaiLoc), "asynq:{default}:processed:2024-03-22"},
}

for _, tc := range tests {
Expand All @@ -228,6 +237,10 @@ func TestProcessedKey(t *testing.T) {
}

func TestFailedKey(t *testing.T) {
shanghaiLoc, err := time.LoadLocation("Asia/Shanghai")
if err != nil {
t.Skip("could not load Asia/Shanghai timezone, skipping")
}
tests := []struct {
qname string
input time.Time
Expand All @@ -236,6 +249,9 @@ func TestFailedKey(t *testing.T) {
{"default", time.Date(2019, 11, 14, 10, 30, 1, 1, time.UTC), "asynq:{default}:failed:2019-11-14"},
{"custom", time.Date(2020, 12, 1, 1, 0, 1, 1, time.UTC), "asynq:{custom}:failed:2020-12-01"},
{"low", time.Date(2020, 1, 6, 15, 02, 1, 1, time.UTC), "asynq:{low}:failed:2020-01-06"},
// Timezone-aware test: 2024-03-22 07:30 CST = 2024-03-21 23:30 UTC
// With the fix, the key should use the local date (03-22), not UTC date (03-21).
{"default", time.Date(2024, 3, 22, 7, 30, 0, 0, shanghaiLoc), "asynq:{default}:failed:2024-03-22"},
}

for _, tc := range tests {
Expand Down
4 changes: 2 additions & 2 deletions internal/rdb/inspect.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -146,7 +146,7 @@ func (r *RDB) CurrentStats(qname string) (*Stats, error) {
if !exists {
return nil, errors.E(op, errors.NotFound, &errors.QueueNotFoundError{Queue: qname})
}
now := r.clock.Now()
now := r.clock.Now().In(r.location)
keys := []string{
base.PendingKey(qname),
base.ActiveKey(qname),
Expand Down Expand Up @@ -373,7 +373,7 @@ func (r *RDB) HistoricalStats(qname string, n int) ([]*DailyStats, error) {
return nil, errors.E(op, errors.NotFound, &errors.QueueNotFoundError{Queue: qname})
}
const day = 24 * time.Hour
now := r.clock.Now().UTC()
now := r.clock.Now().In(r.location)
var days []time.Time
var keys []string
for i := 0; i < n; i++ {
Expand Down
31 changes: 23 additions & 8 deletions internal/rdb/rdb.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,14 +29,16 @@ const LeaseDuration = 30 * time.Second
type RDB struct {
client redis.UniversalClient
clock timeutil.Clock
location *time.Location
queuesPublished sync.Map
}

// NewRDB returns a new instance of RDB.
func NewRDB(client redis.UniversalClient) *RDB {
return &RDB{
client: client,
clock: timeutil.NewRealClock(),
client: client,
clock: timeutil.NewRealClock(),
location: time.UTC,
}
}

Expand All @@ -57,6 +59,19 @@ func (r *RDB) SetClock(c timeutil.Clock) {
r.clock = c
}

// SetLocation sets the time zone location used by RDB to determine
// the date boundary for daily processed/failed stats keys.
//
// By default, RDB uses time.UTC. Setting this to a different location
// (e.g., time.Local, or a location loaded via time.LoadLocation("Asia/Shanghai"))
// will cause daily stats to be aggregated according to that timezone's date boundary.
func (r *RDB) SetLocation(loc *time.Location) {
Copy link
Copy Markdown
Collaborator

@kamikazechaser kamikazechaser Apr 9, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm not sure how someone can use this because it is inside internal.

if loc == nil {
loc = time.UTC
}
r.location = loc
}

// Ping checks the connection with redis server.
func (r *RDB) Ping() error {
return r.client.Ping(context.Background()).Err()
Expand Down Expand Up @@ -351,7 +366,7 @@ func (r *RDB) Done(ctx context.Context, msg *base.TaskMessage) error {
base.ActiveKey(msg.Queue),
base.LeaseKey(msg.Queue),
base.TaskKey(msg.Queue, msg.ID),
base.ProcessedKey(msg.Queue, now),
base.ProcessedKey(msg.Queue, now.In(r.location)),
base.ProcessedTotalKey(msg.Queue),
}
argv := []interface{}{
Expand Down Expand Up @@ -459,7 +474,7 @@ func (r *RDB) MarkAsComplete(ctx context.Context, msg *base.TaskMessage) error {
base.LeaseKey(msg.Queue),
base.CompletedKey(msg.Queue),
base.TaskKey(msg.Queue, msg.ID),
base.ProcessedKey(msg.Queue, now),
base.ProcessedKey(msg.Queue, now.In(r.location)),
base.ProcessedTotalKey(msg.Queue),
}
argv := []interface{}{
Expand Down Expand Up @@ -820,8 +835,8 @@ func (r *RDB) Retry(ctx context.Context, msg *base.TaskMessage, processAt time.T
base.ActiveKey(msg.Queue),
base.LeaseKey(msg.Queue),
base.RetryKey(msg.Queue),
base.ProcessedKey(msg.Queue, now),
base.FailedKey(msg.Queue, now),
base.ProcessedKey(msg.Queue, now.In(r.location)),
base.FailedKey(msg.Queue, now.In(r.location)),
base.ProcessedTotalKey(msg.Queue),
base.FailedTotalKey(msg.Queue),
}
Expand Down Expand Up @@ -920,8 +935,8 @@ func (r *RDB) Archive(ctx context.Context, msg *base.TaskMessage, errMsg string)
base.ActiveKey(msg.Queue),
base.LeaseKey(msg.Queue),
base.ArchivedKey(msg.Queue),
base.ProcessedKey(msg.Queue, now),
base.FailedKey(msg.Queue, now),
base.ProcessedKey(msg.Queue, now.In(r.location)),
base.FailedKey(msg.Queue, now.In(r.location)),
base.ProcessedTotalKey(msg.Queue),
base.FailedTotalKey(msg.Queue),
base.TaskKeyPrefix(msg.Queue),
Expand Down
Loading