From 31123fd42a4941007ed399ccd0fcae81c92b9872 Mon Sep 17 00:00:00 2001 From: Ken Hibino Date: Fri, 24 Jan 2020 07:19:58 -0800 Subject: [PATCH] Paginate tasks with asynqmon ls command Changes: * Added --page and --size flags to ls command * By default, the command will show first 30 tasks from the specified queue --- CHANGELOG.md | 5 + internal/rdb/inspect.go | 112 +++++------ internal/rdb/inspect_test.go | 351 +++++++++++++++++++++++++++++++---- tools/asynqmon/cmd/ls.go | 58 +++--- 4 files changed, 397 insertions(+), 129 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 27706ef..606db78 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Changed + +- `asynqmon ls` command is now paginated (default 30 tasks from first page) +- `asynqmon ls enqueued:[queue name]` requires queue name to be specified + ## [0.2.1] - 2020-01-22 ### Fixed diff --git a/internal/rdb/inspect.go b/internal/rdb/inspect.go index 155535c..d47dafb 100644 --- a/internal/rdb/inspect.go +++ b/internal/rdb/inspect.go @@ -235,67 +235,46 @@ func (r *RDB) RedisInfo() (map[string]string, error) { return info, nil } +func reverse(x []string) { + for i := len(x)/2 - 1; i >= 0; i-- { + opp := len(x) - 1 - i + x[i], x[opp] = x[opp], x[i] + } +} + +// Pagination specifies the page size and page number +// for the list operation. +type Pagination struct { + // Number of items in the page. + Size uint + + // Page number starting from zero. + Page uint +} + +func (p Pagination) start() int64 { + return int64(p.Size * p.Page) +} + +func (p Pagination) stop() int64 { + return int64(p.Size*p.Page + p.Size - 1) +} + // ListEnqueued returns enqueued tasks that are ready to be processed. -// -// Queue names can be optionally passed to query only the specified queues. -// If none are passed, it will query all queues. -func (r *RDB) ListEnqueued(qnames ...string) ([]*EnqueuedTask, error) { - if len(qnames) == 0 { - return r.listAllEnqueued() +func (r *RDB) ListEnqueued(qname string, pgn Pagination) ([]*EnqueuedTask, error) { + qkey := base.QueueKey(qname) + if !r.client.SIsMember(base.AllQueues, qkey).Val() { + return nil, fmt.Errorf("queue %q does not exist", qname) } - return r.listEnqueued(qnames...) -} - -func (r *RDB) listAllEnqueued() ([]*EnqueuedTask, error) { - script := redis.NewScript(` - local res = {} - local queues = redis.call("SMEMBERS", KEYS[1]) - for _, qkey in ipairs(queues) do - local msgs = redis.call("LRANGE", qkey, 0, -1) - for _, msg in ipairs(msgs) do - table.insert(res, msg) - end - end - return res - `) - res, err := script.Run(r.client, []string{base.AllQueues}).Result() + // Note: Because we use LPUSH to redis list, we need to calculate the + // correct range and reverse the list to get the tasks with pagination. + stop := -pgn.start() - 1 + start := -pgn.stop() - 1 + data, err := r.client.LRange(qkey, start, stop).Result() if err != nil { return nil, err } - data, err := cast.ToStringSliceE(res) - if err != nil { - return nil, err - } - return toEnqueuedTasks(data) -} - -func (r *RDB) listEnqueued(qnames ...string) ([]*EnqueuedTask, error) { - script := redis.NewScript(` - local res = {} - for _, qkey in ipairs(KEYS) do - local msgs = redis.call("LRANGE", qkey, 0, -1) - for _, msg in ipairs(msgs) do - table.insert(res, msg) - end - end - return res - `) - var keys []string - for _, q := range qnames { - keys = append(keys, base.QueueKey(q)) - } - res, err := script.Run(r.client, keys).Result() - if err != nil { - return nil, err - } - data, err := cast.ToStringSliceE(res) - if err != nil { - return nil, err - } - return toEnqueuedTasks(data) -} - -func toEnqueuedTasks(data []string) ([]*EnqueuedTask, error) { + reverse(data) var tasks []*EnqueuedTask for _, s := range data { var msg base.TaskMessage @@ -314,11 +293,16 @@ func toEnqueuedTasks(data []string) ([]*EnqueuedTask, error) { } // ListInProgress returns all tasks that are currently being processed. -func (r *RDB) ListInProgress() ([]*InProgressTask, error) { - data, err := r.client.LRange(base.InProgressQueue, 0, -1).Result() +func (r *RDB) ListInProgress(pgn Pagination) ([]*InProgressTask, error) { + // Note: Because we use LPUSH to redis list, we need to calculate the + // correct range and reverse the list to get the tasks with pagination. + stop := -pgn.start() - 1 + start := -pgn.stop() - 1 + data, err := r.client.LRange(base.InProgressQueue, start, stop).Result() if err != nil { return nil, err } + reverse(data) var tasks []*InProgressTask for _, s := range data { var msg base.TaskMessage @@ -337,8 +321,8 @@ func (r *RDB) ListInProgress() ([]*InProgressTask, error) { // ListScheduled returns all tasks that are scheduled to be processed // in the future. -func (r *RDB) ListScheduled() ([]*ScheduledTask, error) { - data, err := r.client.ZRangeWithScores(base.ScheduledQueue, 0, -1).Result() +func (r *RDB) ListScheduled(pgn Pagination) ([]*ScheduledTask, error) { + data, err := r.client.ZRangeWithScores(base.ScheduledQueue, pgn.start(), pgn.stop()).Result() if err != nil { return nil, err } @@ -368,8 +352,8 @@ func (r *RDB) ListScheduled() ([]*ScheduledTask, error) { // ListRetry returns all tasks that have failed before and willl be retried // in the future. -func (r *RDB) ListRetry() ([]*RetryTask, error) { - data, err := r.client.ZRangeWithScores(base.RetryQueue, 0, -1).Result() +func (r *RDB) ListRetry(pgn Pagination) ([]*RetryTask, error) { + data, err := r.client.ZRangeWithScores(base.RetryQueue, pgn.start(), pgn.stop()).Result() if err != nil { return nil, err } @@ -401,8 +385,8 @@ func (r *RDB) ListRetry() ([]*RetryTask, error) { } // ListDead returns all tasks that have exhausted its retry limit. -func (r *RDB) ListDead() ([]*DeadTask, error) { - data, err := r.client.ZRangeWithScores(base.DeadQueue, 0, -1).Result() +func (r *RDB) ListDead(pgn Pagination) ([]*DeadTask, error) { + data, err := r.client.ZRangeWithScores(base.DeadQueue, pgn.start(), pgn.stop()).Result() if err != nil { return nil, err } diff --git a/internal/rdb/inspect_test.go b/internal/rdb/inspect_test.go index 82cc52f..33cd0db 100644 --- a/internal/rdb/inspect_test.go +++ b/internal/rdb/inspect_test.go @@ -5,6 +5,7 @@ package rdb import ( + "fmt" "sort" "testing" "time" @@ -231,25 +232,24 @@ func TestListEnqueued(t *testing.T) { t1 := &EnqueuedTask{ID: m1.ID, Type: m1.Type, Payload: m1.Payload, Queue: m1.Queue} t2 := &EnqueuedTask{ID: m2.ID, Type: m2.Type, Payload: m2.Payload, Queue: m2.Queue} t3 := &EnqueuedTask{ID: m3.ID, Type: m3.Type, Payload: m3.Payload, Queue: m3.Queue} - t4 := &EnqueuedTask{ID: m4.ID, Type: m4.Type, Payload: m4.Payload, Queue: m4.Queue} tests := []struct { enqueued map[string][]*base.TaskMessage - qnames []string + qname string want []*EnqueuedTask }{ { enqueued: map[string][]*base.TaskMessage{ base.DefaultQueueName: {m1, m2}, }, - qnames: []string{}, - want: []*EnqueuedTask{t1, t2}, + qname: base.DefaultQueueName, + want: []*EnqueuedTask{t1, t2}, }, { enqueued: map[string][]*base.TaskMessage{ base.DefaultQueueName: {}, }, - qnames: []string{}, - want: []*EnqueuedTask{}, + qname: base.DefaultQueueName, + want: []*EnqueuedTask{}, }, { enqueued: map[string][]*base.TaskMessage{ @@ -257,8 +257,8 @@ func TestListEnqueued(t *testing.T) { "critical": {m3}, "low": {m4}, }, - qnames: []string{}, - want: []*EnqueuedTask{t1, t2, t3, t4}, + qname: base.DefaultQueueName, + want: []*EnqueuedTask{t1, t2}, }, { enqueued: map[string][]*base.TaskMessage{ @@ -266,17 +266,8 @@ func TestListEnqueued(t *testing.T) { "critical": {m3}, "low": {m4}, }, - qnames: []string{"critical"}, - want: []*EnqueuedTask{t3}, - }, - { - enqueued: map[string][]*base.TaskMessage{ - base.DefaultQueueName: {m1, m2}, - "critical": {m3}, - "low": {m4}, - }, - qnames: []string{"critical", "low"}, - want: []*EnqueuedTask{t3, t4}, + qname: "critical", + want: []*EnqueuedTask{t3}, }, } @@ -286,9 +277,10 @@ func TestListEnqueued(t *testing.T) { h.SeedEnqueuedQueue(t, r.client, msgs, qname) } - got, err := r.ListEnqueued(tc.qnames...) + got, err := r.ListEnqueued(tc.qname, Pagination{Size: 20, Page: 0}) + op := fmt.Sprintf("r.ListEnqueued(%q, Pagination{Size: 20, Page: 0})", tc.qname) if err != nil { - t.Errorf("r.ListEnqueued() = %v, %v, want %v, nil", got, err, tc.want) + t.Errorf("%s = %v, %v, want %v, nil", op, got, err, tc.want) continue } sortOpt := cmp.Transformer("SortMsg", func(in []*EnqueuedTask) []*EnqueuedTask { @@ -299,11 +291,76 @@ func TestListEnqueued(t *testing.T) { return out }) if diff := cmp.Diff(tc.want, got, sortOpt); diff != "" { - t.Errorf("r.ListEnqueued() = %v, %v, want %v, nil; (-want, +got)\n%s", got, err, tc.want, diff) + t.Errorf("%s = %v, %v, want %v, nil; (-want, +got)\n%s", op, got, err, tc.want, diff) continue } } } +func TestListEnqueuedPagination(t *testing.T) { + r := setup(t) + var msgs []*base.TaskMessage + for i := 0; i < 100; i++ { + msg := h.NewTaskMessage(fmt.Sprintf("task %d", i), nil) + msgs = append(msgs, msg) + } + // create 100 tasks in default queue + h.SeedEnqueuedQueue(t, r.client, msgs) + + msgs = []*base.TaskMessage(nil) // empty list + for i := 0; i < 100; i++ { + msg := h.NewTaskMessage(fmt.Sprintf("custom %d", i), nil) + msgs = append(msgs, msg) + } + // create 100 tasks in custom queue + h.SeedEnqueuedQueue(t, r.client, msgs, "custom") + + tests := []struct { + desc string + qname string + page uint + size uint + wantSize int + wantFirst string + wantLast string + }{ + {"first page", "default", 0, 20, 20, "task 0", "task 19"}, + {"second page", "default", 1, 20, 20, "task 20", "task 39"}, + {"different page size", "default", 2, 30, 30, "task 60", "task 89"}, + {"last page", "default", 3, 30, 10, "task 90", "task 99"}, + {"out of range", "default", 4, 30, 0, "", ""}, + {"second page with custom queue", "custom", 1, 20, 20, "custom 20", "custom 39"}, + } + + for _, tc := range tests { + got, err := r.ListEnqueued(tc.qname, Pagination{Size: tc.size, Page: tc.page}) + op := fmt.Sprintf("r.ListEnqueued(%q, Pagination{Size: %d, Page: %d})", tc.qname, tc.size, tc.page) + if err != nil { + t.Errorf("%s; %s returned error %v", tc.desc, op, err) + continue + } + + if len(got) != tc.wantSize { + t.Errorf("%s; %s returned a list of size %d, want %d", tc.desc, op, len(got), tc.wantSize) + continue + } + + if tc.wantSize == 0 { + continue + } + + first := got[0] + if first.Type != tc.wantFirst { + t.Errorf("%s; %s returned a list with first message %q, want %q", + tc.desc, op, first.Type, tc.wantFirst) + } + + last := got[len(got)-1] + if last.Type != tc.wantLast { + t.Errorf("%s; %s returned a list with the last message %q, want %q", + tc.desc, op, last.Type, tc.wantLast) + } + } +} func TestListInProgress(t *testing.T) { r := setup(t) @@ -330,9 +387,10 @@ func TestListInProgress(t *testing.T) { h.FlushDB(t, r.client) // clean up db before each test case h.SeedInProgressQueue(t, r.client, tc.inProgress) - got, err := r.ListInProgress() + got, err := r.ListInProgress(Pagination{Size: 20, Page: 0}) + op := "r.ListInProgress(Pagination{Size: 20, Page: 0})" if err != nil { - t.Errorf("r.ListInProgress() = %v, %v, want %v, nil", got, err, tc.want) + t.Errorf("%s = %v, %v, want %v, nil", op, got, err, tc.want) continue } sortOpt := cmp.Transformer("SortMsg", func(in []*InProgressTask) []*InProgressTask { @@ -343,12 +401,67 @@ func TestListInProgress(t *testing.T) { return out }) if diff := cmp.Diff(tc.want, got, sortOpt); diff != "" { - t.Errorf("r.ListInProgress() = %v, %v, want %v, nil; (-want, +got)\n%s", got, err, tc.want, diff) + t.Errorf("%s = %v, %v, want %v, nil; (-want, +got)\n%s", op, got, err, tc.want, diff) continue } } } +func TestListInProgressPagination(t *testing.T) { + r := setup(t) + var msgs []*base.TaskMessage + for i := 0; i < 100; i++ { + msg := h.NewTaskMessage(fmt.Sprintf("task %d", i), nil) + msgs = append(msgs, msg) + } + h.SeedInProgressQueue(t, r.client, msgs) + + tests := []struct { + desc string + page uint + size uint + wantSize int + wantFirst string + wantLast string + }{ + {"first page", 0, 20, 20, "task 0", "task 19"}, + {"second page", 1, 20, 20, "task 20", "task 39"}, + {"different page size", 2, 30, 30, "task 60", "task 89"}, + {"last page", 3, 30, 10, "task 90", "task 99"}, + {"out of range", 4, 30, 0, "", ""}, + } + + for _, tc := range tests { + got, err := r.ListInProgress(Pagination{Size: tc.size, Page: tc.page}) + op := fmt.Sprintf("r.ListInProgress(Pagination{Size: %d, Page: %d})", tc.size, tc.page) + if err != nil { + t.Errorf("%s; %s returned error %v", tc.desc, op, err) + continue + } + + if len(got) != tc.wantSize { + t.Errorf("%s; %s returned list of size %d, want %d", tc.desc, op, len(got), tc.wantSize) + continue + } + + if tc.wantSize == 0 { + continue + } + + first := got[0] + if first.Type != tc.wantFirst { + t.Errorf("%s; %s returned a list with first message %q, want %q", + tc.desc, op, first.Type, tc.wantFirst) + } + + last := got[len(got)-1] + if last.Type != tc.wantLast { + t.Errorf("%s; %s returned a list with the last message %q, want %q", + tc.desc, op, last.Type, tc.wantLast) + } + } +} + func TestListScheduled(t *testing.T) { r := setup(t) m1 := h.NewTaskMessage("send_email", map[string]interface{}{"subject": "hello"}) @@ -379,9 +492,10 @@ func TestListScheduled(t *testing.T) { h.FlushDB(t, r.client) // clean up db before each test case h.SeedScheduledQueue(t, r.client, tc.scheduled) - got, err := r.ListScheduled() + got, err := r.ListScheduled(Pagination{Size: 20, Page: 0}) + op := "r.ListScheduled(Pagination{Size: 20, Page: 0})" if err != nil { - t.Errorf("r.ListScheduled() = %v, %v, want %v, nil", got, err, tc.want) + t.Errorf("%s = %v, %v, want %v, nil", op, got, err, tc.want) continue } sortOpt := cmp.Transformer("SortMsg", func(in []*ScheduledTask) []*ScheduledTask { @@ -392,12 +506,68 @@ func TestListScheduled(t *testing.T) { return out }) if diff := cmp.Diff(tc.want, got, sortOpt, timeCmpOpt); diff != "" { - t.Errorf("r.ListScheduled() = %v, %v, want %v, nil; (-want, +got)\n%s", got, err, tc.want, diff) + t.Errorf("%s = %v, %v, want %v, nil; (-want, +got)\n%s", op, got, err, tc.want, diff) continue } } } +func TestListScheduledPagination(t *testing.T) { + r := setup(t) + // create 100 tasks with an increasing number of wait time. + for i := 0; i < 100; i++ { + msg := h.NewTaskMessage(fmt.Sprintf("task %d", i), nil) + if err := r.Schedule(msg, time.Now().Add(time.Duration(i)*time.Second)); err != nil { + t.Fatal(err) + } + } + + tests := []struct { + desc string + page uint + size uint + wantSize int + wantFirst string + wantLast string + }{ + {"first page", 0, 20, 20, "task 0", "task 19"}, + {"second page", 1, 20, 20, "task 20", "task 39"}, + {"different page size", 2, 30, 30, "task 60", "task 89"}, + {"last page", 3, 30, 10, "task 90", "task 99"}, + {"out of range", 4, 30, 0, "", ""}, + } + + for _, tc := range tests { + got, err := r.ListScheduled(Pagination{Size: tc.size, Page: tc.page}) + op := fmt.Sprintf("r.ListScheduled(Pagination{Size: %d, Page: %d})", tc.size, tc.page) + if err != nil { + t.Errorf("%s; %s returned error %v", tc.desc, op, err) + continue + } + + if len(got) != tc.wantSize { + t.Errorf("%s; %s returned list of size %d, want %d", tc.desc, op, len(got), tc.wantSize) + continue + } + + if tc.wantSize == 0 { + continue + } + + first := got[0] + if first.Type != tc.wantFirst { + t.Errorf("%s; %s returned a list with first message %q, want %q", + tc.desc, op, first.Type, tc.wantFirst) + } + + last := got[len(got)-1] + if last.Type != tc.wantLast { + t.Errorf("%s; %s returned a list with the last message %q, want %q", + tc.desc, op, last.Type, tc.wantLast) + } + } +} + func TestListRetry(t *testing.T) { r := setup(t) m1 := &base.TaskMessage{ @@ -464,9 +634,10 @@ func TestListRetry(t *testing.T) { h.FlushDB(t, r.client) // clean up db before each test case h.SeedRetryQueue(t, r.client, tc.retry) - got, err := r.ListRetry() + got, err := r.ListRetry(Pagination{Size: 20, Page: 0}) + op := "r.ListRetry(Pagination{Size: 20, Page: 0})" if err != nil { - t.Errorf("r.ListRetry() = %v, %v, want %v, nil", got, err, tc.want) + t.Errorf("%s = %v, %v, want %v, nil", op, got, err, tc.want) continue } sortOpt := cmp.Transformer("SortMsg", func(in []*RetryTask) []*RetryTask { @@ -478,12 +649,68 @@ func TestListRetry(t *testing.T) { }) if diff := cmp.Diff(tc.want, got, sortOpt, timeCmpOpt); diff != "" { - t.Errorf("r.ListRetry() = %v, %v, want %v, nil; (-want, +got)\n%s", got, err, tc.want, diff) + t.Errorf("%s = %v, %v, want %v, nil; (-want, +got)\n%s", op, got, err, tc.want, diff) continue } } } +func TestListRetryPagination(t *testing.T) { + r := setup(t) + // create 100 tasks with an increasing number of wait time. + for i := 0; i < 100; i++ { + msg := h.NewTaskMessage(fmt.Sprintf("task %d", i), nil) + if err := r.Retry(msg, time.Now().Add(time.Duration(i)*time.Second), "error"); err != nil { + t.Fatal(err) + } + } + + tests := []struct { + desc string + page uint + size uint + wantSize int + wantFirst string + wantLast string + }{ + {"first page", 0, 20, 20, "task 0", "task 19"}, + {"second page", 1, 20, 20, "task 20", "task 39"}, + {"different page size", 2, 30, 30, "task 60", "task 89"}, + {"last page", 3, 30, 10, "task 90", "task 99"}, + {"out of range", 4, 30, 0, "", ""}, + } + + for _, tc := range tests { + got, err := r.ListRetry(Pagination{Size: tc.size, Page: tc.page}) + op := fmt.Sprintf("r.ListRetry(Pagination{Size: %d, Page: %d})", tc.size, tc.page) + if err != nil { + t.Errorf("%s; %s returned error %v", tc.desc, op, err) + continue + } + + if len(got) != tc.wantSize { + t.Errorf("%s; %s returned list of size %d, want %d", tc.desc, op, len(got), tc.wantSize) + continue + } + + if tc.wantSize == 0 { + continue + } + + first := got[0] + if first.Type != tc.wantFirst { + t.Errorf("%s; %s returned a list with first message %q, want %q", + tc.desc, op, first.Type, tc.wantFirst) + } + + last := got[len(got)-1] + if last.Type != tc.wantLast { + t.Errorf("%s; %s returned a list with the last message %q, want %q", + tc.desc, op, last.Type, tc.wantLast) + } + } +} + func TestListDead(t *testing.T) { r := setup(t) m1 := &base.TaskMessage{ @@ -542,9 +769,10 @@ func TestListDead(t *testing.T) { h.FlushDB(t, r.client) // clean up db before each test case h.SeedDeadQueue(t, r.client, tc.dead) - got, err := r.ListDead() + got, err := r.ListDead(Pagination{Size: 20, Page: 0}) + op := "r.ListDead(Pagination{Size: 20, Page: 0})" if err != nil { - t.Errorf("r.ListDead() = %v, %v, want %v, nil", got, err, tc.want) + t.Errorf("%s = %v, %v, want %v, nil", op, got, err, tc.want) continue } sortOpt := cmp.Transformer("SortMsg", func(in []*DeadTask) []*DeadTask { @@ -555,12 +783,67 @@ func TestListDead(t *testing.T) { return out }) if diff := cmp.Diff(tc.want, got, sortOpt, timeCmpOpt); diff != "" { - t.Errorf("r.ListDead() = %v, %v, want %v, nil; (-want, +got)\n%s", got, err, tc.want, diff) + t.Errorf("%s = %v, %v, want %v, nil; (-want, +got)\n%s", op, got, err, tc.want, diff) continue } } } +func TestListDeadPagination(t *testing.T) { + r := setup(t) + var entries []h.ZSetEntry + for i := 0; i < 100; i++ { + msg := h.NewTaskMessage(fmt.Sprintf("task %d", i), nil) + entries = append(entries, h.ZSetEntry{Msg: msg, Score: float64(i)}) + } + h.SeedDeadQueue(t, r.client, entries) + + tests := []struct { + desc string + page uint + size uint + wantSize int + wantFirst string + wantLast string + }{ + {"first page", 0, 20, 20, "task 0", "task 19"}, + {"second page", 1, 20, 20, "task 20", "task 39"}, + {"different page size", 2, 30, 30, "task 60", "task 89"}, + {"last page", 3, 30, 10, "task 90", "task 99"}, + {"out of range", 4, 30, 0, "", ""}, + } + + for _, tc := range tests { + got, err := r.ListDead(Pagination{Size: tc.size, Page: tc.page}) + op := fmt.Sprintf("r.ListDead(Pagination{Size: %d, Page: %d})", tc.size, tc.page) + if err != nil { + t.Errorf("%s; %s returned error %v", tc.desc, op, err) + continue + } + + if len(got) != tc.wantSize { + t.Errorf("%s; %s returned list of size %d, want %d", tc.desc, op, len(got), tc.wantSize) + continue + } + + if tc.wantSize == 0 { + continue + } + + first := got[0] + if first.Type != tc.wantFirst { + t.Errorf("%s; %s returned a list with first message %q, want %q", + tc.desc, op, first.Type, tc.wantFirst) + } + + last := got[len(got)-1] + if last.Type != tc.wantLast { + t.Errorf("%s; %s returned a list with the last message %q, want %q", + tc.desc, op, last.Type, tc.wantLast) + } + } +} + var timeCmpOpt = cmpopts.EquateApproxTime(time.Second) func TestEnqueueDeadTask(t *testing.T) { diff --git a/tools/asynqmon/cmd/ls.go b/tools/asynqmon/cmd/ls.go index 3250368..929f875 100644 --- a/tools/asynqmon/cmd/ls.go +++ b/tools/asynqmon/cmd/ls.go @@ -35,26 +35,23 @@ The argument value should be one of "enqueued", "inprogress", "scheduled", Example: asynqmon ls dead -> Lists all tasks in dead state -Enqueued tasks can optionally be filtered by providing queue names after ":" +Enqueued tasks requires a queue name after ":" Example: -asynqmon ls enqueued:critical -> List tasks from critical queue only +asynqmon ls enqueued:default -> List tasks from default queue +asynqmon ls enqueued:critical -> List tasks from critical queue `, Args: cobra.ExactValidArgs(1), Run: ls, } +// Flags +var pageSize uint +var pageNum uint + func init() { rootCmd.AddCommand(lsCmd) - - // Here you will define your flags and configuration settings. - - // Cobra supports Persistent Flags which will work for this command - // and all subcommands, e.g.: - // lsCmd.PersistentFlags().String("foo", "", "A help for foo") - - // Cobra supports local flags which will only run when this command - // is called directly, e.g.: - // lsCmd.Flags().BoolP("toggle", "t", false, "Help message for toggle") + lsCmd.Flags().UintVar(&pageSize, "size", 30, "page size") + lsCmd.Flags().UintVar(&pageNum, "page", 0, "page number - zero indexed (default 0)") } func ls(cmd *cobra.Command, args []string) { @@ -67,7 +64,11 @@ func ls(cmd *cobra.Command, args []string) { parts := strings.Split(args[0], ":") switch parts[0] { case "enqueued": - listEnqueued(r, parts[1:]...) + if len(parts) != 2 { + fmt.Printf("error: Missing queue name\n`asynqmon ls enqueued:[queue name]`\n") + os.Exit(1) + } + listEnqueued(r, parts[1]) case "inprogress": listInProgress(r) case "scheduled": @@ -77,7 +78,7 @@ func ls(cmd *cobra.Command, args []string) { case "dead": listDead(r) default: - fmt.Printf("error: `asynqmon ls [state]` only accepts %v as the argument.\n", lsValidArgs) + fmt.Printf("error: `asynqmon ls [state]`\nonly accepts %v as the argument.\n", lsValidArgs) os.Exit(1) } } @@ -113,24 +114,14 @@ func parseQueryID(queryID string) (id xid.ID, score int64, qtype string, err err return id, score, qtype, nil } -func listEnqueued(r *rdb.RDB, qnames ...string) { - tasks, err := r.ListEnqueued(qnames...) +func listEnqueued(r *rdb.RDB, qname string) { + tasks, err := r.ListEnqueued(qname, rdb.Pagination{Size: pageSize, Page: pageNum}) if err != nil { fmt.Println(err) os.Exit(1) } if len(tasks) == 0 { - msg := "No enqueued tasks" - if len(qnames) > 0 { - msg += " in" - for i, q := range qnames { - msg += fmt.Sprintf(" %q queue", q) - if i != len(qnames)-1 { - msg += "," - } - } - } - fmt.Println(msg) + fmt.Printf("No enqueued tasks in %q queue\n", qname) return } cols := []string{"ID", "Type", "Payload", "Queue"} @@ -140,10 +131,11 @@ func listEnqueued(r *rdb.RDB, qnames ...string) { } } printTable(cols, printRows) + fmt.Printf("\nShowing %d tasks from page %d\n", len(tasks), pageNum) } func listInProgress(r *rdb.RDB) { - tasks, err := r.ListInProgress() + tasks, err := r.ListInProgress(rdb.Pagination{Size: pageSize, Page: pageNum}) if err != nil { fmt.Println(err) os.Exit(1) @@ -159,10 +151,11 @@ func listInProgress(r *rdb.RDB) { } } printTable(cols, printRows) + fmt.Printf("\nShowing %d tasks from page %d\n", len(tasks), pageNum) } func listScheduled(r *rdb.RDB) { - tasks, err := r.ListScheduled() + tasks, err := r.ListScheduled(rdb.Pagination{Size: pageSize, Page: pageNum}) if err != nil { fmt.Println(err) os.Exit(1) @@ -179,10 +172,11 @@ func listScheduled(r *rdb.RDB) { } } printTable(cols, printRows) + fmt.Printf("\nShowing %d tasks from page %d\n", len(tasks), pageNum) } func listRetry(r *rdb.RDB) { - tasks, err := r.ListRetry() + tasks, err := r.ListRetry(rdb.Pagination{Size: pageSize, Page: pageNum}) if err != nil { fmt.Println(err) os.Exit(1) @@ -199,10 +193,11 @@ func listRetry(r *rdb.RDB) { } } printTable(cols, printRows) + fmt.Printf("\nShowing %d tasks from page %d\n", len(tasks), pageNum) } func listDead(r *rdb.RDB) { - tasks, err := r.ListDead() + tasks, err := r.ListDead(rdb.Pagination{Size: pageSize, Page: pageNum}) if err != nil { fmt.Println(err) os.Exit(1) @@ -218,6 +213,7 @@ func listDead(r *rdb.RDB) { } } printTable(cols, printRows) + fmt.Printf("\nShowing %d tasks from page %d\n", len(tasks), pageNum) } func printTable(cols []string, printRows func(w io.Writer, tmpl string)) {