go-pkg/asynq
2
0
Fork 0
mirror of https://github.com/hibiken/asynq.git synced 2024-12-27 00:02:19 +08:00
Code Issues Projects Releases Wiki Activity

Update migrate command to move all dead tasks to the new archived zset

Browse Source
This commit is contained in:
Ken Hibino 2021-01-12 11:24:12 -08:00
parent 00a25ca570
commit ea23db4f6b
2 changed files with 31 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
tools/asynq/cmd/migrate.go
View File

@ -98,7 +98,9 @@ func migrate(cmd *cobra.Command, args []string) {
printError(err) printError(err)
os.Exit(1) os.Exit(1)
} }
if err := partitionZSetMembersByQueue(c, "asynq:dead", base.DeadKey); err != nil { // Note: base.DeadKey function was renamed in v0.14. We define the legacy function here since we need it for this migration script.
deadKeyFunc := func(qname string) string { return fmt.Sprintf("asynq:{%s}:dead", qname) }
if err := partitionZSetMembersByQueue(c, "asynq:dead", deadKeyFunc); err != nil {
printError(err) printError(err)
os.Exit(1) os.Exit(1)
} }
@ -113,7 +115,7 @@ func migrate(cmd *cobra.Command, args []string) {
paused, err := c.SMembers("asynq:paused").Result() paused, err := c.SMembers("asynq:paused").Result()
if err != nil { if err != nil {
printError(fmt.Errorf("command SMEMBERS asynq:paused failed: ", err)) printError(fmt.Errorf("command SMEMBERS asynq:paused failed: %v", err))
os.Exit(1) os.Exit(1)
} }
for _, qkey := range paused { for _, qkey := range paused {
@ -136,6 +138,27 @@ func migrate(cmd *cobra.Command, args []string) {
printError(err) printError(err)
os.Exit(1) os.Exit(1)
} }
/*** Migrate from 0.13 to 0.14 compatible ***/
// Move all dead tasks to archived ZSET.
for _, qname := range allQueues {
zs, err := c.ZRangeWithScores(deadKeyFunc(qname), 0, -1).Result()
if err != nil {
printError(err)
os.Exit(1)
}
for _, z := range zs {
if err := c.ZAdd(base.ArchivedKey(qname), &z).Err(); err != nil {
printError(err)
os.Exit(1)
}
}
if err := deleteKey(c, deadKeyFunc(qname)); err != nil {
printError(err)
os.Exit(1)
}
}
} }
func backupKey(key string) string { func backupKey(key string) string {

12
tools/asynq/cmd/task.go
View File

@ -125,7 +125,7 @@ var taskArchiveAllCmd = &cobra.Command{
Use: "archive-all --queue=QUEUE --state=STATE", Use: "archive-all --queue=QUEUE --state=STATE",
Short: "Archive all tasks in the given state", Short: "Archive all tasks in the given state",
Args: cobra.NoArgs, Args: cobra.NoArgs,
Run: taskKillAll, Run: taskArchiveAll,
} }
var taskDeleteAllCmd = &cobra.Command{ var taskDeleteAllCmd = &cobra.Command{
@ -275,7 +275,7 @@ func listRetryTasks(qname string, pageNum, pageSize int) {
func listArchivedTasks(qname string, pageNum, pageSize int) { func listArchivedTasks(qname string, pageNum, pageSize int) {
i := createInspector() i := createInspector()
tasks, err := i.ListDeadTasks(qname, asynq.PageSize(pageSize), asynq.Page(pageNum)) tasks, err := i.ListArchivedTasks(qname, asynq.PageSize(pageSize), asynq.Page(pageNum))
if err != nil { if err != nil {
fmt.Println(err) fmt.Println(err)
os.Exit(1) os.Exit(1)
@ -318,7 +318,7 @@ func taskKill(cmd *cobra.Command, args []string) {
} }
i := createInspector() i := createInspector()
err = i.KillTaskByKey(qname, key) err = i.ArchiveTaskByKey(qname, key)
if err != nil { if err != nil {
fmt.Printf("error: %v\n", err) fmt.Printf("error: %v\n", err)
os.Exit(1) os.Exit(1)
@ -368,7 +368,7 @@ func taskRun(cmd *cobra.Command, args []string) {
fmt.Println("task transitioned to pending state") fmt.Println("task transitioned to pending state")
} }
func taskKillAll(cmd *cobra.Command, args []string) { func taskArchiveAll(cmd *cobra.Command, args []string) {
qname, err := cmd.Flags().GetString("queue") qname, err := cmd.Flags().GetString("queue")
if err != nil { if err != nil {
fmt.Printf("error: %v\n", err) fmt.Printf("error: %v\n", err)
@ -384,9 +384,9 @@ func taskKillAll(cmd *cobra.Command, args []string) {
var n int var n int
switch state { switch state {
case "scheduled": case "scheduled":
n, err = i.KillAllScheduledTasks(qname) n, err = i.ArchiveAllScheduledTasks(qname)
case "retry": case "retry":
n, err = i.KillAllRetryTasks(qname) n, err = i.ArchiveAllRetryTasks(qname)
default: default:
fmt.Printf("error: unsupported state %q\n", state) fmt.Printf("error: unsupported state %q\n", state)
os.Exit(1) os.Exit(1)