2
0
mirror of https://github.com/hibiken/asynq.git synced 2024-11-15 11:58:47 +08:00
asynq/inspector.go

1027 lines
29 KiB
Go
Raw Normal View History

2020-07-13 21:29:41 +08:00
// Copyright 2020 Kentaro Hibino. All rights reserved.
// Use of this source code is governed by a MIT license
// that can be found in the LICENSE file.
package asynq
2020-07-13 21:29:41 +08:00
import (
"fmt"
"strconv"
"strings"
"time"
"github.com/hibiken/asynq/internal/base"
2021-05-11 12:19:57 +08:00
"github.com/hibiken/asynq/internal/errors"
2020-07-13 21:29:41 +08:00
"github.com/hibiken/asynq/internal/rdb"
2023-09-19 16:16:51 +08:00
"github.com/redis/go-redis/v9"
2020-07-13 21:29:41 +08:00
)
// Inspector is a client interface to inspect and mutate the state of
// queues and tasks.
type Inspector struct {
rdb *rdb.RDB
2023-09-19 16:16:51 +08:00
// When an Inspector has been created with an existing Redis connection, we do
// not want to close it.
sharedConnection bool
2020-07-13 21:29:41 +08:00
}
// New returns a new instance of Inspector.
func NewInspector(r RedisConnOpt) *Inspector {
2021-01-29 22:37:35 +08:00
c, ok := r.MakeRedisClient().(redis.UniversalClient)
if !ok {
panic(fmt.Sprintf("inspeq: unsupported RedisConnOpt type %T", r))
2020-07-13 21:29:41 +08:00
}
2023-09-19 16:16:51 +08:00
inspector := NewInspectorFromRedisClient(c)
inspector.sharedConnection = false
return inspector
}
// NewInspectorFromRedisClient returns a new instance of Inspector given a redis.UniversalClient
// Warning: The underlying redis connection pool will not be closed by Asynq, you are responsible for closing it.
2023-09-19 16:16:51 +08:00
func NewInspectorFromRedisClient(c redis.UniversalClient) *Inspector {
2021-01-29 22:37:35 +08:00
return &Inspector{
2023-09-19 16:16:51 +08:00
rdb: rdb.NewRDB(c),
sharedConnection: true,
}
}
2020-09-08 21:52:34 +08:00
// Close closes the connection with redis.
func (i *Inspector) Close() error {
2023-09-19 16:16:51 +08:00
if i.sharedConnection {
return fmt.Errorf("redis connection is shared so the Inspector can't be closed through asynq")
}
2020-09-08 21:52:34 +08:00
return i.rdb.Close()
}
2020-08-21 12:17:44 +08:00
// Queues returns a list of all queue names.
func (i *Inspector) Queues() ([]string, error) {
return i.rdb.AllQueues()
}
2022-03-19 22:23:09 +08:00
// Groups returns a list of all groups within the given queue.
func (i *Inspector) Groups(queue string) ([]*GroupInfo, error) {
stats, err := i.rdb.GroupStats(queue)
2022-03-19 22:23:09 +08:00
if err != nil {
return nil, err
}
var res []*GroupInfo
for _, s := range stats {
res = append(res, &GroupInfo{
Group: s.Group,
Size: s.Size,
})
}
return res, nil
}
2022-05-17 12:14:15 +08:00
// GroupInfo represents a state of a group at a certain time.
2022-03-19 22:23:09 +08:00
type GroupInfo struct {
// Name of the group.
Group string
// Size is the total number of tasks in the group.
Size int
}
// QueueInfo represents a state of a queue at a certain time.
type QueueInfo struct {
2020-08-17 05:51:56 +08:00
// Name of the queue.
Queue string
2021-01-27 10:36:45 +08:00
// Total number of bytes that the queue and its tasks require to be stored in redis.
// It is an approximate memory usage value in bytes since the value is computed by sampling.
2021-01-27 10:36:45 +08:00
MemoryUsage int64
2021-12-11 22:27:44 +08:00
// Latency of the queue, measured by the oldest pending task in the queue.
Latency time.Duration
// Size is the total number of tasks in the queue.
// The value is the sum of Pending, Active, Scheduled, Retry, Aggregating and Archived.
Size int
// Groups is the total number of groups in the queue.
Groups int
2020-09-05 22:03:43 +08:00
// Number of pending tasks.
Pending int
2020-09-06 03:43:15 +08:00
// Number of active tasks.
Active int
2020-08-17 05:51:56 +08:00
// Number of scheduled tasks.
Scheduled int
// Number of retry tasks.
Retry int
// Number of archived tasks.
Archived int
// Number of stored completed tasks.
Completed int
// Number of aggregating tasks.
Aggregating int
2021-12-17 08:53:02 +08:00
// Total number of tasks being processed within the given date (counter resets daily).
2020-08-17 05:51:56 +08:00
// The number includes both succeeded and failed tasks.
Processed int
2021-12-17 08:53:02 +08:00
// Total number of tasks failed to be processed within the given date (counter resets daily).
2020-08-17 05:51:56 +08:00
Failed int
2021-12-17 08:53:02 +08:00
// Total number of tasks processed (cumulative).
ProcessedTotal int
// Total number of tasks failed (cumulative).
FailedTotal int
2020-07-13 21:29:41 +08:00
// Paused indicates whether the queue is paused.
2020-08-17 05:51:56 +08:00
// If true, tasks in the queue will not be processed.
2020-07-13 21:29:41 +08:00
Paused bool
// Time when this queue info snapshot was taken.
2020-08-17 05:51:56 +08:00
Timestamp time.Time
2020-07-13 21:29:41 +08:00
}
// GetQueueInfo returns current information of the given queue.
func (i *Inspector) GetQueueInfo(queue string) (*QueueInfo, error) {
if err := base.ValidateQueueName(queue); err != nil {
2020-08-31 20:53:17 +08:00
return nil, err
}
stats, err := i.rdb.CurrentStats(queue)
2020-07-13 21:29:41 +08:00
if err != nil {
return nil, err
}
return &QueueInfo{
2021-12-17 08:53:02 +08:00
Queue: stats.Queue,
MemoryUsage: stats.MemoryUsage,
Latency: stats.Latency,
Size: stats.Size,
Groups: stats.Groups,
2021-12-17 08:53:02 +08:00
Pending: stats.Pending,
Active: stats.Active,
Scheduled: stats.Scheduled,
Retry: stats.Retry,
Archived: stats.Archived,
Completed: stats.Completed,
Aggregating: stats.Aggregating,
2021-12-17 08:53:02 +08:00
Processed: stats.Processed,
Failed: stats.Failed,
ProcessedTotal: stats.ProcessedTotal,
FailedTotal: stats.FailedTotal,
Paused: stats.Paused,
Timestamp: stats.Timestamp,
2020-08-18 11:49:55 +08:00
}, nil
2020-08-17 05:51:56 +08:00
}
// DailyStats holds aggregate data for a given day for a given queue.
2020-07-13 21:29:41 +08:00
type DailyStats struct {
2020-08-17 05:51:56 +08:00
// Name of the queue.
Queue string
// Total number of tasks being processed during the given date.
// The number includes both succeeded and failed tasks.
2020-07-13 21:29:41 +08:00
Processed int
2020-08-17 05:51:56 +08:00
// Total number of tasks failed to be processed during the given date.
Failed int
// Date this stats was taken.
Date time.Time
2020-07-13 21:29:41 +08:00
}
// History returns a list of stats from the last n days.
func (i *Inspector) History(queue string, n int) ([]*DailyStats, error) {
if err := base.ValidateQueueName(queue); err != nil {
2020-08-31 20:53:17 +08:00
return nil, err
}
stats, err := i.rdb.HistoricalStats(queue, n)
2020-07-13 21:29:41 +08:00
if err != nil {
return nil, err
}
var res []*DailyStats
for _, s := range stats {
res = append(res, &DailyStats{
2020-08-17 05:51:56 +08:00
Queue: s.Queue,
2020-07-13 21:29:41 +08:00
Processed: s.Processed,
Failed: s.Failed,
Date: s.Time,
})
}
return res, nil
}
2021-05-01 21:47:49 +08:00
var (
// ErrQueueNotFound indicates that the specified queue does not exist.
ErrQueueNotFound = errors.New("queue not found")
2021-05-20 07:39:02 +08:00
2021-05-01 21:47:49 +08:00
// ErrQueueNotEmpty indicates that the specified queue is not empty.
ErrQueueNotEmpty = errors.New("queue is not empty")
2021-05-20 07:39:02 +08:00
// ErrTaskNotFound indicates that the specified task cannot be found in the queue.
ErrTaskNotFound = errors.New("task not found")
2021-05-01 21:47:49 +08:00
)
// DeleteQueue removes the specified queue.
//
// If force is set to true, DeleteQueue will remove the queue regardless of
// the queue size as long as no tasks are active in the queue.
// If force is set to false, DeleteQueue will remove the queue only if
// the queue is empty.
//
// If the specified queue does not exist, DeleteQueue returns ErrQueueNotFound.
// If force is set to false and the specified queue is not empty, DeleteQueue
// returns ErrQueueNotEmpty.
func (i *Inspector) DeleteQueue(queue string, force bool) error {
err := i.rdb.RemoveQueue(queue, force)
2021-05-11 12:19:57 +08:00
if errors.IsQueueNotFound(err) {
return fmt.Errorf("%w: queue=%q", ErrQueueNotFound, queue)
}
2021-05-11 12:19:57 +08:00
if errors.IsQueueNotEmpty(err) {
return fmt.Errorf("%w: queue=%q", ErrQueueNotEmpty, queue)
}
return err
}
2021-05-23 09:30:44 +08:00
// GetTaskInfo retrieves task information given a task id and queue name.
//
// Returns an error wrapping ErrQueueNotFound if a queue with the given name doesn't exist.
// Returns an error wrapping ErrTaskNotFound if a task with the given id doesn't exist in the queue.
func (i *Inspector) GetTaskInfo(queue, id string) (*TaskInfo, error) {
info, err := i.rdb.GetTaskInfo(queue, id)
2021-05-23 09:30:44 +08:00
switch {
case errors.IsQueueNotFound(err):
return nil, fmt.Errorf("asynq: %w", ErrQueueNotFound)
case errors.IsTaskNotFound(err):
return nil, fmt.Errorf("asynq: %w", ErrTaskNotFound)
case err != nil:
return nil, fmt.Errorf("asynq: %v", err)
}
return newTaskInfo(info.Message, info.State, info.NextProcessAt, info.Result), nil
2021-05-23 09:30:44 +08:00
}
2020-07-13 21:29:41 +08:00
// ListOption specifies behavior of list operation.
type ListOption interface{}
// Internal list option representations.
type (
pageSizeOpt int
pageNumOpt int
)
type listOption struct {
pageSize int
pageNum int
}
const (
// Page size used by default in list operation.
defaultPageSize = 30
// Page number used by default in list operation.
defaultPageNum = 1
)
func composeListOptions(opts ...ListOption) listOption {
res := listOption{
pageSize: defaultPageSize,
pageNum: defaultPageNum,
}
for _, opt := range opts {
switch opt := opt.(type) {
case pageSizeOpt:
res.pageSize = int(opt)
case pageNumOpt:
res.pageNum = int(opt)
default:
// ignore unexpected option
}
}
return res
}
// PageSize returns an option to specify the page size for list operation.
//
// Negative page size is treated as zero.
func PageSize(n int) ListOption {
if n < 0 {
n = 0
}
return pageSizeOpt(n)
}
// Page returns an option to specify the page number for list operation.
// The value 1 fetches the first page.
//
// Negative page number is treated as one.
func Page(n int) ListOption {
if n < 0 {
n = 1
}
return pageNumOpt(n)
}
2020-09-05 22:03:43 +08:00
// ListPendingTasks retrieves pending tasks from the specified queue.
2020-07-13 21:29:41 +08:00
//
// By default, it retrieves the first 30 tasks.
func (i *Inspector) ListPendingTasks(queue string, opts ...ListOption) ([]*TaskInfo, error) {
if err := base.ValidateQueueName(queue); err != nil {
return nil, fmt.Errorf("asynq: %v", err)
2020-08-31 20:53:17 +08:00
}
2020-07-13 21:29:41 +08:00
opt := composeListOptions(opts...)
pgn := rdb.Pagination{Size: opt.pageSize, Page: opt.pageNum - 1}
infos, err := i.rdb.ListPending(queue, pgn)
switch {
case errors.IsQueueNotFound(err):
return nil, fmt.Errorf("asynq: %w", ErrQueueNotFound)
case err != nil:
return nil, fmt.Errorf("asynq: %v", err)
2020-07-13 21:29:41 +08:00
}
var tasks []*TaskInfo
for _, i := range infos {
tasks = append(tasks, newTaskInfo(
i.Message,
i.State,
i.NextProcessAt,
i.Result,
))
2020-07-13 21:29:41 +08:00
}
return tasks, err
}
2020-09-06 03:43:15 +08:00
// ListActiveTasks retrieves active tasks from the specified queue.
2020-07-13 21:29:41 +08:00
//
// By default, it retrieves the first 30 tasks.
func (i *Inspector) ListActiveTasks(queue string, opts ...ListOption) ([]*TaskInfo, error) {
if err := base.ValidateQueueName(queue); err != nil {
return nil, fmt.Errorf("asynq: %v", err)
2020-08-31 20:53:17 +08:00
}
2020-07-13 21:29:41 +08:00
opt := composeListOptions(opts...)
pgn := rdb.Pagination{Size: opt.pageSize, Page: opt.pageNum - 1}
infos, err := i.rdb.ListActive(queue, pgn)
switch {
case errors.IsQueueNotFound(err):
return nil, fmt.Errorf("asynq: %w", ErrQueueNotFound)
case err != nil:
return nil, fmt.Errorf("asynq: %v", err)
2020-07-13 21:29:41 +08:00
}
expired, err := i.rdb.ListLeaseExpired(time.Now(), queue)
2022-02-19 12:50:18 +08:00
if err != nil {
return nil, fmt.Errorf("asynq: %v", err)
}
expiredSet := make(map[string]struct{}) // set of expired message IDs
for _, msg := range expired {
expiredSet[msg.ID] = struct{}{}
}
var tasks []*TaskInfo
for _, i := range infos {
2022-02-19 12:50:18 +08:00
t := newTaskInfo(
i.Message,
i.State,
i.NextProcessAt,
i.Result,
2022-02-19 12:50:18 +08:00
)
if _, ok := expiredSet[i.Message.ID]; ok {
t.IsOrphaned = true
}
tasks = append(tasks, t)
2020-07-13 21:29:41 +08:00
}
2022-02-19 12:50:18 +08:00
return tasks, nil
2020-07-13 21:29:41 +08:00
}
2022-03-22 21:23:22 +08:00
// ListAggregatingTasks retrieves scheduled tasks from the specified group.
//
// By default, it retrieves the first 30 tasks.
func (i *Inspector) ListAggregatingTasks(queue, group string, opts ...ListOption) ([]*TaskInfo, error) {
if err := base.ValidateQueueName(queue); err != nil {
2022-03-22 21:23:22 +08:00
return nil, fmt.Errorf("asynq: %v", err)
}
opt := composeListOptions(opts...)
pgn := rdb.Pagination{Size: opt.pageSize, Page: opt.pageNum - 1}
infos, err := i.rdb.ListAggregating(queue, group, pgn)
2022-03-22 21:23:22 +08:00
switch {
case errors.IsQueueNotFound(err):
return nil, fmt.Errorf("asynq: %w", ErrQueueNotFound)
case err != nil:
return nil, fmt.Errorf("asynq: %v", err)
}
var tasks []*TaskInfo
for _, i := range infos {
tasks = append(tasks, newTaskInfo(
i.Message,
i.State,
i.NextProcessAt,
i.Result,
))
}
return tasks, nil
2022-03-19 22:23:09 +08:00
}
2020-08-17 05:51:56 +08:00
// ListScheduledTasks retrieves scheduled tasks from the specified queue.
// Tasks are sorted by NextProcessAt in ascending order.
2020-07-13 21:29:41 +08:00
//
// By default, it retrieves the first 30 tasks.
func (i *Inspector) ListScheduledTasks(queue string, opts ...ListOption) ([]*TaskInfo, error) {
if err := base.ValidateQueueName(queue); err != nil {
return nil, fmt.Errorf("asynq: %v", err)
2020-08-31 20:53:17 +08:00
}
2020-07-13 21:29:41 +08:00
opt := composeListOptions(opts...)
pgn := rdb.Pagination{Size: opt.pageSize, Page: opt.pageNum - 1}
infos, err := i.rdb.ListScheduled(queue, pgn)
switch {
case errors.IsQueueNotFound(err):
return nil, fmt.Errorf("asynq: %w", ErrQueueNotFound)
case err != nil:
return nil, fmt.Errorf("asynq: %v", err)
2020-07-13 21:29:41 +08:00
}
var tasks []*TaskInfo
for _, i := range infos {
tasks = append(tasks, newTaskInfo(
i.Message,
i.State,
i.NextProcessAt,
i.Result,
))
2020-07-13 21:29:41 +08:00
}
return tasks, nil
}
2020-08-17 05:51:56 +08:00
// ListRetryTasks retrieves retry tasks from the specified queue.
// Tasks are sorted by NextProcessAt in ascending order.
2020-07-13 21:29:41 +08:00
//
// By default, it retrieves the first 30 tasks.
func (i *Inspector) ListRetryTasks(queue string, opts ...ListOption) ([]*TaskInfo, error) {
if err := base.ValidateQueueName(queue); err != nil {
return nil, fmt.Errorf("asynq: %v", err)
2020-08-31 20:53:17 +08:00
}
2020-07-13 21:29:41 +08:00
opt := composeListOptions(opts...)
pgn := rdb.Pagination{Size: opt.pageSize, Page: opt.pageNum - 1}
infos, err := i.rdb.ListRetry(queue, pgn)
switch {
case errors.IsQueueNotFound(err):
return nil, fmt.Errorf("asynq: %w", ErrQueueNotFound)
case err != nil:
return nil, fmt.Errorf("asynq: %v", err)
2020-07-13 21:29:41 +08:00
}
var tasks []*TaskInfo
for _, i := range infos {
tasks = append(tasks, newTaskInfo(
i.Message,
i.State,
i.NextProcessAt,
i.Result,
))
2020-07-13 21:29:41 +08:00
}
return tasks, nil
}
// ListArchivedTasks retrieves archived tasks from the specified queue.
// Tasks are sorted by LastFailedAt in descending order.
2020-07-13 21:29:41 +08:00
//
// By default, it retrieves the first 30 tasks.
func (i *Inspector) ListArchivedTasks(queue string, opts ...ListOption) ([]*TaskInfo, error) {
if err := base.ValidateQueueName(queue); err != nil {
return nil, fmt.Errorf("asynq: %v", err)
2020-08-31 20:53:17 +08:00
}
2020-07-13 21:29:41 +08:00
opt := composeListOptions(opts...)
pgn := rdb.Pagination{Size: opt.pageSize, Page: opt.pageNum - 1}
infos, err := i.rdb.ListArchived(queue, pgn)
switch {
case errors.IsQueueNotFound(err):
return nil, fmt.Errorf("asynq: %w", ErrQueueNotFound)
case err != nil:
return nil, fmt.Errorf("asynq: %v", err)
2020-07-13 21:29:41 +08:00
}
2021-05-19 12:06:53 +08:00
var tasks []*TaskInfo
for _, i := range infos {
tasks = append(tasks, newTaskInfo(
i.Message,
i.State,
i.NextProcessAt,
i.Result,
))
}
return tasks, nil
}
// ListCompletedTasks retrieves completed tasks from the specified queue.
// Tasks are sorted by expiration time (i.e. CompletedAt + Retention) in descending order.
//
// By default, it retrieves the first 30 tasks.
func (i *Inspector) ListCompletedTasks(queue string, opts ...ListOption) ([]*TaskInfo, error) {
if err := base.ValidateQueueName(queue); err != nil {
return nil, fmt.Errorf("asynq: %v", err)
}
opt := composeListOptions(opts...)
pgn := rdb.Pagination{Size: opt.pageSize, Page: opt.pageNum - 1}
infos, err := i.rdb.ListCompleted(queue, pgn)
switch {
case errors.IsQueueNotFound(err):
return nil, fmt.Errorf("asynq: %w", ErrQueueNotFound)
case err != nil:
return nil, fmt.Errorf("asynq: %v", err)
}
var tasks []*TaskInfo
for _, i := range infos {
tasks = append(tasks, newTaskInfo(
i.Message,
i.State,
i.NextProcessAt,
i.Result,
))
2020-07-13 21:29:41 +08:00
}
return tasks, nil
}
// DeleteAllPendingTasks deletes all pending tasks from the specified queue,
// and reports the number tasks deleted.
func (i *Inspector) DeleteAllPendingTasks(queue string) (int, error) {
if err := base.ValidateQueueName(queue); err != nil {
return 0, err
}
n, err := i.rdb.DeleteAllPendingTasks(queue)
return int(n), err
}
2020-08-17 05:51:56 +08:00
// DeleteAllScheduledTasks deletes all scheduled tasks from the specified queue,
2020-07-13 21:29:41 +08:00
// and reports the number tasks deleted.
func (i *Inspector) DeleteAllScheduledTasks(queue string) (int, error) {
if err := base.ValidateQueueName(queue); err != nil {
2020-08-31 20:53:17 +08:00
return 0, err
}
n, err := i.rdb.DeleteAllScheduledTasks(queue)
2020-07-13 21:29:41 +08:00
return int(n), err
}
2020-08-17 05:51:56 +08:00
// DeleteAllRetryTasks deletes all retry tasks from the specified queue,
2020-07-13 21:29:41 +08:00
// and reports the number tasks deleted.
func (i *Inspector) DeleteAllRetryTasks(queue string) (int, error) {
if err := base.ValidateQueueName(queue); err != nil {
2020-08-31 20:53:17 +08:00
return 0, err
}
n, err := i.rdb.DeleteAllRetryTasks(queue)
2020-07-13 21:29:41 +08:00
return int(n), err
}
// DeleteAllArchivedTasks deletes all archived tasks from the specified queue,
2020-07-13 21:29:41 +08:00
// and reports the number tasks deleted.
func (i *Inspector) DeleteAllArchivedTasks(queue string) (int, error) {
if err := base.ValidateQueueName(queue); err != nil {
2020-08-31 20:53:17 +08:00
return 0, err
}
n, err := i.rdb.DeleteAllArchivedTasks(queue)
2020-07-13 21:29:41 +08:00
return int(n), err
}
// DeleteAllCompletedTasks deletes all completed tasks from the specified queue,
// and reports the number tasks deleted.
func (i *Inspector) DeleteAllCompletedTasks(queue string) (int, error) {
if err := base.ValidateQueueName(queue); err != nil {
return 0, err
}
n, err := i.rdb.DeleteAllCompletedTasks(queue)
return int(n), err
}
// DeleteAllAggregatingTasks deletes all tasks from the specified group,
// and reports the number of tasks deleted.
func (i *Inspector) DeleteAllAggregatingTasks(queue, group string) (int, error) {
if err := base.ValidateQueueName(queue); err != nil {
return 0, err
}
n, err := i.rdb.DeleteAllAggregatingTasks(queue, group)
return int(n), err
2022-03-19 22:23:09 +08:00
}
// DeleteTask deletes a task with the given id from the given queue.
// The task needs to be in pending, scheduled, retry, or archived state,
// otherwise DeleteTask will return an error.
//
// If a queue with the given name doesn't exist, it returns an error wrapping ErrQueueNotFound.
// If a task with the given id doesn't exist in the queue, it returns an error wrapping ErrTaskNotFound.
// If the task is in active state, it returns a non-nil error.
func (i *Inspector) DeleteTask(queue, id string) error {
if err := base.ValidateQueueName(queue); err != nil {
return fmt.Errorf("asynq: %v", err)
2020-08-31 20:53:17 +08:00
}
err := i.rdb.DeleteTask(queue, id)
switch {
case errors.IsQueueNotFound(err):
return fmt.Errorf("asynq: %w", ErrQueueNotFound)
case errors.IsTaskNotFound(err):
return fmt.Errorf("asynq: %w", ErrTaskNotFound)
case err != nil:
return fmt.Errorf("asynq: %v", err)
}
return nil
2020-07-13 21:29:41 +08:00
}
// RunAllScheduledTasks schedules all scheduled tasks from the given queue to run,
// and reports the number of tasks scheduled to run.
func (i *Inspector) RunAllScheduledTasks(queue string) (int, error) {
if err := base.ValidateQueueName(queue); err != nil {
2020-08-31 20:53:17 +08:00
return 0, err
}
n, err := i.rdb.RunAllScheduledTasks(queue)
2020-07-13 21:29:41 +08:00
return int(n), err
}
// RunAllRetryTasks schedules all retry tasks from the given queue to run,
// and reports the number of tasks scheduled to run.
func (i *Inspector) RunAllRetryTasks(queue string) (int, error) {
if err := base.ValidateQueueName(queue); err != nil {
2020-08-31 20:53:17 +08:00
return 0, err
}
n, err := i.rdb.RunAllRetryTasks(queue)
2020-07-13 21:29:41 +08:00
return int(n), err
}
// RunAllArchivedTasks schedules all archived tasks from the given queue to run,
// and reports the number of tasks scheduled to run.
func (i *Inspector) RunAllArchivedTasks(queue string) (int, error) {
if err := base.ValidateQueueName(queue); err != nil {
2020-08-31 20:53:17 +08:00
return 0, err
}
n, err := i.rdb.RunAllArchivedTasks(queue)
2020-07-13 21:29:41 +08:00
return int(n), err
}
// RunAllAggregatingTasks schedules all tasks from the given grou to run.
// and reports the number of tasks scheduled to run.
func (i *Inspector) RunAllAggregatingTasks(queue, group string) (int, error) {
if err := base.ValidateQueueName(queue); err != nil {
return 0, err
}
n, err := i.rdb.RunAllAggregatingTasks(queue, group)
return int(n), err
}
// RunTask updates the task to pending state given a queue name and task id.
// The task needs to be in scheduled, retry, or archived state, otherwise RunTask
// will return an error.
//
// If a queue with the given name doesn't exist, it returns an error wrapping ErrQueueNotFound.
// If a task with the given id doesn't exist in the queue, it returns an error wrapping ErrTaskNotFound.
// If the task is in pending or active state, it returns a non-nil error.
func (i *Inspector) RunTask(queue, id string) error {
if err := base.ValidateQueueName(queue); err != nil {
return fmt.Errorf("asynq: %v", err)
2020-08-31 20:53:17 +08:00
}
err := i.rdb.RunTask(queue, id)
switch {
case errors.IsQueueNotFound(err):
return fmt.Errorf("asynq: %w", ErrQueueNotFound)
case errors.IsTaskNotFound(err):
return fmt.Errorf("asynq: %w", ErrTaskNotFound)
case err != nil:
return fmt.Errorf("asynq: %v", err)
}
return nil
2020-07-13 21:29:41 +08:00
}
// ArchiveAllPendingTasks archives all pending tasks from the given queue,
// and reports the number of tasks archived.
func (i *Inspector) ArchiveAllPendingTasks(queue string) (int, error) {
if err := base.ValidateQueueName(queue); err != nil {
return 0, err
}
n, err := i.rdb.ArchiveAllPendingTasks(queue)
return int(n), err
}
// ArchiveAllScheduledTasks archives all scheduled tasks from the given queue,
// and reports the number of tasks archiveed.
func (i *Inspector) ArchiveAllScheduledTasks(queue string) (int, error) {
if err := base.ValidateQueueName(queue); err != nil {
2020-08-31 20:53:17 +08:00
return 0, err
}
n, err := i.rdb.ArchiveAllScheduledTasks(queue)
2020-07-13 21:29:41 +08:00
return int(n), err
}
// ArchiveAllRetryTasks archives all retry tasks from the given queue,
// and reports the number of tasks archiveed.
func (i *Inspector) ArchiveAllRetryTasks(queue string) (int, error) {
if err := base.ValidateQueueName(queue); err != nil {
2020-08-31 20:53:17 +08:00
return 0, err
}
n, err := i.rdb.ArchiveAllRetryTasks(queue)
2020-07-13 21:29:41 +08:00
return int(n), err
}
// ArchiveAllAggregatingTasks archives all tasks from the given group,
// and reports the number of tasks archived.
func (i *Inspector) ArchiveAllAggregatingTasks(queue, group string) (int, error) {
if err := base.ValidateQueueName(queue); err != nil {
return 0, err
}
n, err := i.rdb.ArchiveAllAggregatingTasks(queue, group)
return int(n), err
2022-03-19 22:23:09 +08:00
}
// ArchiveTask archives a task with the given id in the given queue.
// The task needs to be in pending, scheduled, or retry state, otherwise ArchiveTask
// will return an error.
//
// If a queue with the given name doesn't exist, it returns an error wrapping ErrQueueNotFound.
// If a task with the given id doesn't exist in the queue, it returns an error wrapping ErrTaskNotFound.
// If the task is in already archived, it returns a non-nil error.
func (i *Inspector) ArchiveTask(queue, id string) error {
if err := base.ValidateQueueName(queue); err != nil {
return fmt.Errorf("asynq: err")
2020-08-31 20:53:17 +08:00
}
err := i.rdb.ArchiveTask(queue, id)
switch {
case errors.IsQueueNotFound(err):
return fmt.Errorf("asynq: %w", ErrQueueNotFound)
case errors.IsTaskNotFound(err):
return fmt.Errorf("asynq: %w", ErrTaskNotFound)
case err != nil:
return fmt.Errorf("asynq: %v", err)
}
return nil
2020-07-13 21:29:41 +08:00
}
// CancelProcessing sends a signal to cancel processing of the task
// given a task id. CancelProcessing is best-effort, which means that it does not
// guarantee that the task with the given id will be canceled. The return
// value only indicates whether the cancelation signal has been sent.
func (i *Inspector) CancelProcessing(id string) error {
return i.rdb.PublishCancelation(id)
}
// PauseQueue pauses task processing on the specified queue.
// If the queue is already paused, it will return a non-nil error.
func (i *Inspector) PauseQueue(queue string) error {
if err := base.ValidateQueueName(queue); err != nil {
2020-08-31 20:53:17 +08:00
return err
}
return i.rdb.Pause(queue)
}
// UnpauseQueue resumes task processing on the specified queue.
// If the queue is not paused, it will return a non-nil error.
func (i *Inspector) UnpauseQueue(queue string) error {
if err := base.ValidateQueueName(queue); err != nil {
2020-08-31 20:53:17 +08:00
return err
}
return i.rdb.Unpause(queue)
}
// Servers return a list of running servers' information.
func (i *Inspector) Servers() ([]*ServerInfo, error) {
servers, err := i.rdb.ListServers()
if err != nil {
return nil, err
}
workers, err := i.rdb.ListWorkers()
if err != nil {
return nil, err
}
m := make(map[string]*ServerInfo) // ServerInfo keyed by serverID
for _, s := range servers {
m[s.ServerID] = &ServerInfo{
ID: s.ServerID,
Host: s.Host,
PID: s.PID,
Concurrency: s.Concurrency,
Queues: s.Queues,
StrictPriority: s.StrictPriority,
Started: s.Started,
Status: s.Status,
ActiveWorkers: make([]*WorkerInfo, 0),
}
}
for _, w := range workers {
srvInfo, ok := m[w.ServerID]
if !ok {
continue
}
wrkInfo := &WorkerInfo{
TaskID: w.ID,
TaskType: w.Type,
TaskPayload: w.Payload,
Queue: w.Queue,
Started: w.Started,
Deadline: w.Deadline,
}
srvInfo.ActiveWorkers = append(srvInfo.ActiveWorkers, wrkInfo)
}
var out []*ServerInfo
for _, srvInfo := range m {
out = append(out, srvInfo)
}
return out, nil
}
// ServerInfo describes a running Server instance.
type ServerInfo struct {
// Unique Identifier for the server.
ID string
// Host machine on which the server is running.
Host string
// PID of the process in which the server is running.
PID int
// Server configuration details.
// See Config doc for field descriptions.
Concurrency int
Queues map[string]int
StrictPriority bool
// Time the server started.
Started time.Time
// Status indicates the status of the server.
// TODO: Update comment with more details.
Status string
// A List of active workers currently processing tasks.
ActiveWorkers []*WorkerInfo
}
// WorkerInfo describes a running worker processing a task.
type WorkerInfo struct {
// ID of the task the worker is processing.
TaskID string
// Type of the task the worker is processing.
TaskType string
// Payload of the task the worker is processing.
TaskPayload []byte
// Queue from which the worker got its task.
Queue string
// Time the worker started processing the task.
Started time.Time
2021-01-28 07:55:43 +08:00
// Time the worker needs to finish processing the task by.
Deadline time.Time
}
// ClusterKeySlot returns an integer identifying the hash slot the given queue hashes to.
func (i *Inspector) ClusterKeySlot(queue string) (int64, error) {
return i.rdb.ClusterKeySlot(queue)
}
// ClusterNode describes a node in redis cluster.
type ClusterNode struct {
// Node ID in the cluster.
2021-05-24 11:46:22 +08:00
ID string
// Address of the node.
2021-05-24 11:46:22 +08:00
Addr string
}
// ClusterNodes returns a list of nodes the given queue belongs to.
2021-05-20 07:48:38 +08:00
//
// Only relevant if task queues are stored in redis cluster.
func (i *Inspector) ClusterNodes(queue string) ([]*ClusterNode, error) {
nodes, err := i.rdb.ClusterNodes(queue)
if err != nil {
return nil, err
}
2021-05-20 07:48:38 +08:00
var res []*ClusterNode
for _, node := range nodes {
2021-05-24 11:46:22 +08:00
res = append(res, &ClusterNode{ID: node.ID, Addr: node.Addr})
}
return res, nil
}
// SchedulerEntry holds information about a periodic task registered with a scheduler.
type SchedulerEntry struct {
// Identifier of this entry.
ID string
// Spec describes the schedule of this entry.
Spec string
// Periodic Task registered for this entry.
Task *Task
// Opts is the options for the periodic task.
Opts []Option
// Next shows the next time the task will be enqueued.
Next time.Time
// Prev shows the last time the task was enqueued.
// Zero time if task was never enqueued.
Prev time.Time
}
// SchedulerEntries returns a list of all entries registered with
// currently running schedulers.
func (i *Inspector) SchedulerEntries() ([]*SchedulerEntry, error) {
var entries []*SchedulerEntry
res, err := i.rdb.ListSchedulerEntries()
if err != nil {
return nil, err
}
for _, e := range res {
task := NewTask(e.Type, e.Payload)
var opts []Option
for _, s := range e.Opts {
if o, err := parseOption(s); err == nil {
// ignore bad data
opts = append(opts, o)
}
}
entries = append(entries, &SchedulerEntry{
ID: e.ID,
Spec: e.Spec,
Task: task,
Opts: opts,
Next: e.Next,
Prev: e.Prev,
})
}
return entries, nil
}
// parseOption interprets a string s as an Option and returns the Option if parsing is successful,
// otherwise returns non-nil error.
func parseOption(s string) (Option, error) {
fn, arg := parseOptionFunc(s), parseOptionArg(s)
switch fn {
case "Queue":
queue, err := strconv.Unquote(arg)
if err != nil {
return nil, err
}
return Queue(queue), nil
case "MaxRetry":
n, err := strconv.Atoi(arg)
if err != nil {
return nil, err
}
return MaxRetry(n), nil
case "Timeout":
d, err := time.ParseDuration(arg)
if err != nil {
return nil, err
}
return Timeout(d), nil
case "Deadline":
t, err := time.Parse(time.UnixDate, arg)
if err != nil {
return nil, err
}
return Deadline(t), nil
case "Unique":
d, err := time.ParseDuration(arg)
if err != nil {
return nil, err
}
return Unique(d), nil
case "ProcessAt":
t, err := time.Parse(time.UnixDate, arg)
if err != nil {
return nil, err
}
return ProcessAt(t), nil
case "ProcessIn":
d, err := time.ParseDuration(arg)
if err != nil {
return nil, err
}
return ProcessIn(d), nil
case "Retention":
d, err := time.ParseDuration(arg)
if err != nil {
return nil, err
}
return Retention(d), nil
default:
return nil, fmt.Errorf("cannot not parse option string %q", s)
}
}
func parseOptionFunc(s string) string {
i := strings.Index(s, "(")
return s[:i]
}
func parseOptionArg(s string) string {
i := strings.Index(s, "(")
if i >= 0 {
j := strings.Index(s, ")")
if j > i {
return s[i+1 : j]
}
}
return ""
}
// SchedulerEnqueueEvent holds information about an enqueue event by a scheduler.
type SchedulerEnqueueEvent struct {
// ID of the task that was enqueued.
TaskID string
// Time the task was enqueued.
EnqueuedAt time.Time
}
// ListSchedulerEnqueueEvents retrieves a list of enqueue events from the specified scheduler entry.
//
// By default, it retrieves the first 30 tasks.
func (i *Inspector) ListSchedulerEnqueueEvents(entryID string, opts ...ListOption) ([]*SchedulerEnqueueEvent, error) {
opt := composeListOptions(opts...)
pgn := rdb.Pagination{Size: opt.pageSize, Page: opt.pageNum - 1}
data, err := i.rdb.ListSchedulerEnqueueEvents(entryID, pgn)
if err != nil {
return nil, err
}
var events []*SchedulerEnqueueEvent
for _, e := range data {
events = append(events, &SchedulerEnqueueEvent{TaskID: e.TaskID, EnqueuedAt: e.EnqueuedAt})
}
return events, nil
}