Repository: hibiken/asynq (go)

Issue:
[FEATURE REQUEST] Show elapsed time for "in-progress" tasks 
**Is your feature request related to a problem? Please describe.**
When have I tasks that takes a while to complete, I want to know how long it's taking to process the task. 

**Describe the solution you'd like**
`asynqmon ls inprogress` shows the elapsed time for each task.

**Describe alternatives you've considered**
None.



Pull request:
Add "asynqmon workers" command
Closes #39 

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 4ebfa6f..6066b2d 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- `asynqmon workers` was added to list all running workers information
+
 ## [0.4.0] - 2020-02-13
 
 ### Changed
diff --git a/go.mod b/go.mod
index 18354ea..67c3abd 100644
--- a/go.mod
+++ b/go.mod
@@ -3,13 +3,12 @@ module github.com/hibiken/asynq
 go 1.13
 
 require (
-	github.com/go-redis/redis/v7 v7.0.0-beta.4
+	github.com/go-redis/redis/v7 v7.2.0
 	github.com/google/go-cmp v0.4.0
 	github.com/rs/xid v1.2.1
 	github.com/spf13/cast v1.3.1
 	go.uber.org/goleak v0.10.0
 	golang.org/x/sys v0.0.0-20191204072324-ce4227a45e2e // indirect
-	golang.org/x/text v0.3.2 // indirect
 	golang.org/x/time v0.0.0-20190308202827-9d24e82272b4
 	gopkg.in/yaml.v2 v2.2.7 // indirect
 )
diff --git a/go.sum b/go.sum
index 752e698..913a295 100644
--- a/go.sum
+++ b/go.sum
@@ -4,9 +4,12 @@ github.com/fsnotify/fsnotify v1.4.7 h1:IXs+QLmnXW2CcXuY+8Mzv/fWEsPGWxqefPtCP5CnV
 github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
 github.com/go-redis/redis/v7 v7.0.0-beta.4 h1:p6z7Pde69EGRWvlC++y8aFcaWegyrKHzOBGo0zUACTQ=
 github.com/go-redis/redis/v7 v7.0.0-beta.4/go.mod h1:xhhSbUMTsleRPur+Vgx9sUHtyN33bdjxY+9/0n9Ig8s=
+github.com/go-redis/redis/v7 v7.2.0 h1:CrCexy/jYWZjW0AyVoHlcJUeZN19VWlbepTh1Vq6dJs=
+github.com/go-redis/redis/v7 v7.2.0/go.mod h1:JDNMw23GTyLNC4GZu9njt15ctBQVn7xjRfnwdHj/Dcg=
 github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
 github.com/golang/protobuf v1.3.1 h1:YF8+flBXS5eO826T4nzqPrxfhQThhXl0YzfuUPu4SBg=
 github.com/golang/protobuf v1.3.1/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
+github.com/golang/protobuf v1.3.2/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
 github.com/google/go-cmp v0.4.0 h1:xsAVV57WRhGj6kEIi8ReJzQlHHqcBYCElAvkovg3B/4=
 github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/hpcloud/tail v1.0.0 h1:nfCOvKYfkgYP8hkirhJocXT2+zOD8yUNjXaWfTlyFKI=
@@ -19,8 +22,10 @@ github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
 github.com/onsi/ginkgo v1.8.0 h1:VkHVNpR4iVnU8XQR6DBm8BqYjN7CRzw+xKUbVVbbW9w=
 github.com/onsi/ginkgo v1.8.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
+github.com/onsi/ginkgo v1.10.1/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
 github.com/onsi/gomega v1.5.0 h1:izbySO9zDPmjJ8rDjLvkA2zJHIo+HkYXHnf7eN7SSyo=
 github.com/onsi/gomega v1.5.0/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
+github.com/onsi/gomega v1.7.0/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/rs/xid v1.2.1 h1:mhH9Nq+C1fY2l1XIpgxIiUOfNpRBYH1kKcr+qfKgjRc=
@@ -36,11 +41,13 @@ golang.org/x/net v0.0.0-20180906233101-161cd47e91fd h1:nTDtHvHSdCn1m6ITfMRqtOd/9
 golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20190522155817-f3200d17e092 h1:4QSRKanuywn15aTZvI/mIDEgPQpswuFndXpOj3rKEco=
 golang.org/x/net v0.0.0-20190522155817-f3200d17e092/go.mod h1:HSz+uSET+XFnRR8LxR5pz3Of3rY3CfYBVs4xY44aLks=
+golang.org/x/net v0.0.0-20190923162816-aa69164e4478/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20200202094626-16171245cfb2 h1:CCH4IOTTfewWjGOlSp+zGcjutRKlBEZQ6wTn8ozI/nI=
 golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e h1:o3PsSEY8E4eXWkXrIP9YJALUkVZqzHJT5DOasTyn8Vs=
 golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20191010194322-b09406accb47/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20191204072324-ce4227a45e2e h1:9vRrk9YW2BTzLP0VCB9ZDjU4cPqkg+IDWL7XgxA1yxQ=
 golang.org/x/sys v0.0.0-20191204072324-ce4227a45e2e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/text v0.3.0 h1:g61tztE5qeGQ89tm6NTjjM9VPIm088od1l6aSorWRWg=
@@ -55,11 +62,13 @@ golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8T
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 h1:qIbj1fsPNlZgppZ+VLlY7N33q108Sa+fhmuc+sWQYwY=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/fsnotify.v1 v1.4.7 h1:xOHLXZwVvI9hhs+cLKq5+I5onOuwQLhQwiu63xxlHs4=
 gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys=
 gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 h1:uRGJdciOHaEIrze2W8Q3AKkepLTh2hOroT7a+7czfdQ=
 gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWDmTeBkI65Dw0HsyUHuEVlX15mw=
 gopkg.in/yaml.v2 v2.2.1 h1:mUhvW9EsL+naU5Q3cakzfE91YhliOondGd6ZrsDBHQE=
 gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v2 v2.2.4/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.7 h1:VUgggvou5XRW9mHwD/yXxIYSMtY0zoKQf/v226p2nyo=
 gopkg.in/yaml.v2 v2.2.7/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
diff --git a/internal/rdb/inspect.go b/internal/rdb/inspect.go
index d0f4be7..1678136 100644
--- a/internal/rdb/inspect.go
+++ b/internal/rdb/inspect.go
@@ -794,3 +794,39 @@ func (r *RDB) ListProcesses() ([]*base.ProcessInfo, error) {
 	}
 	return processes, nil
 }
+
+// Note: Script also removes stale keys.
+var listWorkersCmd = redis.NewScript(`
+local res = {}
+local now = tonumber(ARGV[1])
+local keys = redis.call("ZRANGEBYSCORE", KEYS[1], now, "+inf")
+for _, key in ipairs(keys) do
+	local workers = redis.call("HVALS", key)
+	for _, w in ipairs(workers) do
+		table.insert(res, w)
+	end  
+end
+redis.call("ZREMRANGEBYSCORE", KEYS[1], "-inf", now-1)
+return res`)
+
+// ListWorkers returns the list of worker stats.
+func (r *RDB) ListWorkers() ([]*base.WorkerInfo, error) {
+	res, err := listWorkersCmd.Run(r.client, []string{base.AllWorkers}, time.Now().UTC().Unix()).Result()
+	if err != nil {
+		return nil, err
+	}
+	data, err := cast.ToStringSliceE(res)
+	if err != nil {
+		return nil, err
+	}
+	var workers []*base.WorkerInfo
+	for _, s := range data {
+		var w base.WorkerInfo
+		err := json.Unmarshal([]byte(s), &w)
+		if err != nil {
+			continue // skip bad data
+		}
+		workers = append(workers, &w)
+	}
+	return workers, nil
+}
diff --git a/internal/rdb/rdb.go b/internal/rdb/rdb.go
index faf12ed..5a86316 100644
--- a/internal/rdb/rdb.go
+++ b/internal/rdb/rdb.go
@@ -372,6 +372,7 @@ func (r *RDB) forwardSingle(src, dst string) error {
 var writeProcessInfoCmd = redis.NewScript(`
 redis.call("SETEX", KEYS[1], ARGV[2], ARGV[3])
 redis.call("ZADD", KEYS[2], ARGV[1], KEYS[1])
+redis.call("DEL", KEYS[3])
 for i = 4, table.getn(ARGV)-1, 2 do
 	redis.call("HSET", KEYS[3], ARGV[i], ARGV[i+1])
 end
diff --git a/tools/asynqmon/cmd/workers.go b/tools/asynqmon/cmd/workers.go
new file mode 100644
index 0000000..40b21de
--- /dev/null
+++ b/tools/asynqmon/cmd/workers.go
@@ -0,0 +1,75 @@
+// 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 cmd
+
+import (
+	"fmt"
+	"io"
+	"os"
+	"sort"
+
+	"github.com/go-redis/redis/v7"
+	"github.com/hibiken/asynq/internal/rdb"
+	"github.com/spf13/cobra"
+	"github.com/spf13/viper"
+)
+
+// workersCmd represents the workers command
+var workersCmd = &cobra.Command{
+	Use:   "workers",
+	Short: "Shows all running workers information",
+	Long: `Workers (asynqmon workers) will show all running workers information.
+
+The command shows the follwoing for each worker:
+* Process in which the worker is running
+* ID of the task worker is processing
+* Type of the task worker is processing
+* Payload of the task worker is processing
+* Queue that the task was pulled from.
+* Time the worker started processing the task`,
+	Args: cobra.NoArgs,
+	Run:  workers,
+}
+
+func init() {
+	rootCmd.AddCommand(workersCmd)
+}
+
+func workers(cmd *cobra.Command, args []string) {
+	r := rdb.NewRDB(redis.NewClient(&redis.Options{
+		Addr:     viper.GetString("uri"),
+		DB:       viper.GetInt("db"),
+		Password: viper.GetString("password"),
+	}))
+
+	workers, err := r.ListWorkers()
+	if err != nil {
+		fmt.Println(err)
+		os.Exit(1)
+	}
+
+	if len(workers) == 0 {
+		fmt.Println("No workers")
+		return
+	}
+
+	// sort by started timestamp or ID.
+	sort.Slice(workers, func(i, j int) bool {
+		x, y := workers[i], workers[j]
+		if x.Started != y.Started {
+			return x.Started.Before(y.Started)
+		}
+		return x.ID.String() < y.ID.String()
+	})
+
+	cols := []string{"Process", "ID", "Type", "Payload", "Queue", "Started"}
+	printRows := func(w io.Writer, tmpl string) {
+		for _, wk := range workers {
+			fmt.Fprintf(w, tmpl,
+				fmt.Sprintf("%s:%d", wk.Host, wk.PID), wk.ID, wk.Type, wk.Payload, wk.Queue, timeAgo(wk.Started))
+		}
+	}
+	printTable(cols, printRows)
+}
diff --git a/tools/go.mod b/tools/go.mod
index 0b26e71..9685df1 100644
--- a/tools/go.mod
+++ b/tools/go.mod
@@ -3,7 +3,7 @@ module github.com/hibiken/asynq/tools
 go 1.13
 
 require (
-	github.com/go-redis/redis/v7 v7.1.0
+	github.com/go-redis/redis/v7 v7.2.0
 	github.com/hibiken/asynq v0.4.0
 	github.com/mitchellh/go-homedir v1.1.0
 	github.com/rs/xid v1.2.1
diff --git a/tools/go.sum b/tools/go.sum
index a478405..40205a1 100644
--- a/tools/go.sum
+++ b/tools/go.sum
@@ -28,6 +28,8 @@ github.com/go-redis/redis v6.15.7+incompatible h1:3skhDh95XQMpnqeqNftPkQD9jL9e5e
 github.com/go-redis/redis/v7 v7.0.0-beta.4/go.mod h1:xhhSbUMTsleRPur+Vgx9sUHtyN33bdjxY+9/0n9Ig8s=
 github.com/go-redis/redis/v7 v7.1.0 h1:I4C4a8UGbFejiVjtYVTRVOiMIJ5pm5Yru6ibvDX/OS0=
 github.com/go-redis/redis/v7 v7.1.0/go.mod h1:JDNMw23GTyLNC4GZu9njt15ctBQVn7xjRfnwdHj/Dcg=
+github.com/go-redis/redis/v7 v7.2.0 h1:CrCexy/jYWZjW0AyVoHlcJUeZN19VWlbepTh1Vq6dJs=
+github.com/go-redis/redis/v7 v7.2.0/go.mod h1:JDNMw23GTyLNC4GZu9njt15ctBQVn7xjRfnwdHj/Dcg=
 github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
 github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
 github.com/gogo/protobuf v1.2.1/go.mod h1:hp+jE20tsWTFYpLwKvXlhS1hjn+gTNwPg2I6zVXpSg4=


Test patch:
diff --git a/client_test.go b/client_test.go
index 1bd19ef..f0b8710 100644
--- a/client_test.go
+++ b/client_test.go
@@ -15,7 +15,7 @@ import (
 
 func TestClient(t *testing.T) {
 	r := setup(t)
-	client := NewClient(&RedisClientOpt{
+	client := NewClient(RedisClientOpt{
 		Addr: "localhost:6379",
 		DB:   14,
 	})
diff --git a/internal/asynqtest/asynqtest.go b/internal/asynqtest/asynqtest.go
index 7b8a0ae..07de078 100644
--- a/internal/asynqtest/asynqtest.go
+++ b/internal/asynqtest/asynqtest.go
@@ -53,6 +53,15 @@ var SortProcessInfoOpt = cmp.Transformer("SortProcessInfo", func(in []*base.Proc
 	return out
 })
 
+// SortWorkerInfoOpt is a cmp.Option to sort base.WorkerInfo for comparing slice of worker info.
+var SortWorkerInfoOpt = cmp.Transformer("SortWorkerInfo", func(in []*base.WorkerInfo) []*base.WorkerInfo {
+	out := append([]*base.WorkerInfo(nil), in...) // Copy input to avoid mutating it
+	sort.Slice(out, func(i, j int) bool {
+		return out[i].ID.String() < out[j].ID.String()
+	})
+	return out
+})
+
 // SortStringSliceOpt is a cmp.Option to sort string slice.
 var SortStringSliceOpt = cmp.Transformer("SortStringSlice", func(in []string) []string {
 	out := append([]string(nil), in...)
diff --git a/internal/rdb/inspect_test.go b/internal/rdb/inspect_test.go
index 48bcb56..c3b774e 100644
--- a/internal/rdb/inspect_test.go
+++ b/internal/rdb/inspect_test.go
@@ -2122,3 +2122,68 @@ func TestListProcesses(t *testing.T) {
 		}
 	}
 }
+
+func TestListWorkers(t *testing.T) {
+	r := setup(t)
+
+	const (
+		host = "127.0.0.1"
+		pid  = 4567
+	)
+
+	m1 := h.NewTaskMessage("send_email", map[string]interface{}{"user_id": "abc123"})
+	m2 := h.NewTaskMessage("gen_thumbnail", map[string]interface{}{"path": "some/path/to/image/file"})
+	m3 := h.NewTaskMessage("reindex", map[string]interface{}{})
+	t1 := time.Now().Add(-time.Second)
+	t2 := time.Now().Add(-10 * time.Second)
+	t3 := time.Now().Add(-time.Minute)
+
+	type workerStats struct {
+		msg     *base.TaskMessage
+		started time.Time
+	}
+
+	tests := []struct {
+		workers []*workerStats
+		want    []*base.WorkerInfo
+	}{
+		{
+			workers: []*workerStats{
+				{m1, t1},
+				{m2, t2},
+				{m3, t3},
+			},
+			want: []*base.WorkerInfo{
+				{Host: host, PID: pid, ID: m1.ID, Type: m1.Type, Queue: m1.Queue, Payload: m1.Payload, Started: t1},
+				{Host: host, PID: pid, ID: m2.ID, Type: m2.Type, Queue: m2.Queue, Payload: m2.Payload, Started: t2},
+				{Host: host, PID: pid, ID: m3.ID, Type: m3.Type, Queue: m3.Queue, Payload: m3.Payload, Started: t3},
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		h.FlushDB(t, r.client)
+
+		ps := base.NewProcessState(host, pid, 10, map[string]int{"default": 1}, false)
+
+		for _, w := range tc.workers {
+			ps.AddWorkerStats(w.msg, w.started)
+		}
+
+		err := r.WriteProcessState(ps, time.Minute)
+		if err != nil {
+			t.Errorf("could not write process state to redis: %v", err)
+			continue
+		}
+
+		got, err := r.ListWorkers()
+		if err != nil {
+			t.Errorf("(*RDB).ListWorkers() returned an error: %v", err)
+			continue
+		}
+
+		if diff := cmp.Diff(tc.want, got, h.SortWorkerInfoOpt); diff != "" {
+			t.Errorf("(*RDB).ListWorkers() = %v, want = %v; (-want,+got)\n%s", got, tc.want, diff)
+		}
+	}
+}