Add ls command to asynqmon

tools/asynqmon/cmd/ls.go

@@ -0,0 +1,177 @@
+package cmd
+
+import (
+	"fmt"
+	"io"
+	"log"
+	"os"
+	"strings"
+	"text/tabwriter"
+	"time"
+
+	"github.com/go-redis/redis/v7"
+	"github.com/hibiken/asynq/internal/rdb"
+	"github.com/spf13/cobra"
+)
+
+var validArgs = []string{"enqueued", "inprogress", "scheduled", "retry", "dead"}
+
+// lsCmd represents the ls command
+var lsCmd = &cobra.Command{
+	Use:   "ls",
+	Short: "lists queue contents",
+	Long: `The ls command lists all tasks from the given queue in a table format.
+
+The command takes one argument which specifies the queue. The value
+for the argument should be one of "enqueued", "inprogress", "scheduled",
+"retry", or "dead".
+
+Example: asynqmon ls dead`,
+	ValidArgs: validArgs,
+	Args:      cobra.ExactValidArgs(1),
+	Run:       ls,
+}
+
+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")
+}
+
+func ls(cmd *cobra.Command, args []string) {
+	c := redis.NewClient(&redis.Options{
+		Addr: uri,
+		DB:   db,
+	})
+	r := rdb.NewRDB(c)
+	switch args[0] {
+	case "enqueued":
+		listEnqueued(r)
+	case "inprogress":
+		listInProgress(r)
+	case "scheduled":
+		listScheduled(r)
+	case "retry":
+		listRetry(r)
+	case "dead":
+		listDead(r)
+	default:
+		fmt.Printf("error: `asynqmon ls <queue>` only accepts %v as the argument for queue.\n", validArgs)
+		return
+	}
+}
+
+func listEnqueued(r *rdb.RDB) {
+	tasks, err := r.ListEnqueued()
+	if err != nil {
+		log.Fatal(err)
+	}
+	if len(tasks) == 0 {
+		fmt.Println("No enqueued tasks")
+		return
+	}
+	cols := []string{"ID", "Type", "Payload"}
+	printRows := func(w io.Writer, tmpl string) {
+		for _, t := range tasks {
+			fmt.Fprintf(w, tmpl, t.ID, t.Type, t.Payload)
+		}
+	}
+	printTable(cols, printRows)
+}
+
+func listInProgress(r *rdb.RDB) {
+	tasks, err := r.ListInProgress()
+	if err != nil {
+		log.Fatal(err)
+	}
+	if len(tasks) == 0 {
+		fmt.Println("No in-progress tasks")
+		return
+	}
+	cols := []string{"ID", "Type", "Payload"}
+	printRows := func(w io.Writer, tmpl string) {
+		for _, t := range tasks {
+			fmt.Fprintf(w, tmpl, t.ID, t.Type, t.Payload)
+		}
+	}
+	printTable(cols, printRows)
+}
+
+func listScheduled(r *rdb.RDB) {
+	tasks, err := r.ListScheduled()
+	if err != nil {
+		log.Fatal(err)
+	}
+	if len(tasks) == 0 {
+		fmt.Println("No scheduled tasks")
+		return
+	}
+	cols := []string{"ID", "Type", "Payload", "Process In"}
+	printRows := func(w io.Writer, tmpl string) {
+		for _, t := range tasks {
+			processIn := fmt.Sprintf("%.0f seconds", t.ProcessAt.Sub(time.Now()).Seconds())
+			fmt.Fprintf(w, tmpl, t.ID, t.Type, t.Payload, processIn)
+		}
+	}
+	printTable(cols, printRows)
+}
+
+func listRetry(r *rdb.RDB) {
+	tasks, err := r.ListRetry()
+	if err != nil {
+		log.Fatal(err)
+	}
+	if len(tasks) == 0 {
+		fmt.Println("No retry tasks")
+		return
+	}
+	cols := []string{"ID", "Type", "Payload", "Retry In", "Last Error", "Retried", "Max Retry"}
+	printRows := func(w io.Writer, tmpl string) {
+		for _, t := range tasks {
+			retryIn := fmt.Sprintf("%.0f seconds", t.ProcessAt.Sub(time.Now()).Seconds())
+			fmt.Fprintf(w, tmpl, t.ID, t.Type, t.Payload, retryIn, t.ErrorMsg, t.Retried, t.Retry)
+		}
+	}
+	printTable(cols, printRows)
+}
+
+func listDead(r *rdb.RDB) {
+	tasks, err := r.ListDead()
+	if err != nil {
+		log.Fatal(err)
+	}
+	if len(tasks) == 0 {
+		fmt.Println("No dead tasks")
+		return
+	}
+	cols := []string{"ID", "Type", "Payload", "Last Failed", "Last Error"}
+	printRows := func(w io.Writer, tmpl string) {
+		for _, t := range tasks {
+			fmt.Fprintf(w, tmpl, t.ID, t.Type, t.Payload, t.LastFailedAt, t.ErrorMsg)
+		}
+	}
+	printTable(cols, printRows)
+}
+
+func printTable(cols []string, printRows func(w io.Writer, tmpl string)) {
+	format := strings.Repeat("%v\t", len(cols)) + "\n"
+	tw := new(tabwriter.Writer).Init(os.Stdout, 0, 8, 2, ' ', 0)
+	var headers []interface{}
+	var seps []interface{}
+	for _, name := range cols {
+		headers = append(headers, name)
+		seps = append(seps, strings.Repeat("-", len(name)))
+	}
+	fmt.Fprintf(tw, format, headers...)
+	fmt.Fprintf(tw, format, seps...)
+	printRows(tw, format)
+	tw.Flush()
+}

tools/asynqmon/cmd/root.go

@@ -11,6 +11,10 @@ import (
 
 var cfgFile string
 
+// Flags
+var uri string
+var db int
+
 // rootCmd represents the base command when called without any subcommands
 var rootCmd = &cobra.Command{
 	Use:   "asynqmon",
@@ -43,10 +47,8 @@ func init() {
 	// will be global for your application.
 
 	rootCmd.PersistentFlags().StringVar(&cfgFile, "config", "", "config file (default is $HOME/.asynqmon.yaml)")
-
-	// Cobra also supports local flags, which will only run
-	// when this action is called directly.
-	rootCmd.Flags().BoolP("toggle", "t", false, "Help message for toggle")
+	rootCmd.PersistentFlags().StringVarP(&uri, "uri", "u", "127.0.0.1:6379", "Redis server URI")
+	rootCmd.PersistentFlags().IntVarP(&db, "db", "n", 0, "Redis database number (default is 0)")
 }
 
 // initConfig reads in config file and ENV variables if set.

tools/asynqmon/cmd/stats.go

@@ -12,23 +12,18 @@ import (
 	"github.com/spf13/cobra"
 )
 
-// Flags
-var uri string
-var db int
-
 // statsCmd represents the stats command
 var statsCmd = &cobra.Command{
 	Use:   "stats",
-	Short: "Shows current state of the queues",
-	Long: `Stats command shows the number of tasks in each queue at that instant.
+	Short: "shows current state of the queues",
+	Long: `The stats command shows the number of tasks in each queue at that instant.
 
 To monitor the queues continuously, it's recommended that you run this
 command in conjunction with the watch command.
 
 Example: watch -n 5 asynqmon stats`,
-	Run: func(cmd *cobra.Command, args []string) {
-		stats(cmd, args)
-	},
+	Args: cobra.NoArgs,
+	Run:  stats,
 }
 
 func init() {
@@ -43,9 +38,6 @@ func init() {
 	// Cobra supports local flags which will only run when this command
 	// is called directly, e.g.:
 	// statsCmd.Flags().BoolP("toggle", "t", false, "Help message for toggle")
-
-	statsCmd.Flags().StringVarP(&uri, "uri", "u", "127.0.0.1:6379", "Redis server URI")
-	statsCmd.Flags().IntVarP(&db, "db", "n", 0, "Redis database number")
 }
 
 func stats(cmd *cobra.Command, args []string) {