forked from jrallison/go-workers
-
Notifications
You must be signed in to change notification settings - Fork 0
/
stats.go
executable file
·97 lines (81 loc) · 2.1 KB
/
stats.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
package workers
import (
"encoding/json"
"fmt"
"net/http"
"strconv"
)
type stats struct {
Processed int `json:"processed"`
Failed int `json:"failed"`
Jobs interface{} `json:"jobs"`
Enqueued interface{} `json:"enqueued"`
Retries int64 `json:"retries"`
}
func Stats(w http.ResponseWriter, req *http.Request) {
w.Header().Set("Content-Type", "application/json; charset=utf-8")
w.Header().Set("Access-Control-Allow-Origin", "*")
jobs := make(map[string][]*map[string]interface{})
enqueued := make(map[string]string)
for _, m := range managers {
queue := m.queueName()
jobs[queue] = make([]*map[string]interface{}, 0)
enqueued[queue] = ""
for _, worker := range m.workers {
message := worker.currentMsg
startedAt := worker.startedAt
if message != nil && startedAt > 0 {
jobs[queue] = append(jobs[queue], &map[string]interface{}{
"message": message,
"started_at": startedAt,
})
}
}
}
stats := stats{
0,
0,
jobs,
enqueued,
0,
}
conn := Config.Pool.Get()
defer conn.Close()
conn.Send("multi")
conn.Send("get", Config.Namespace+"stat:processed")
conn.Send("get", Config.Namespace+"stat:failed")
conn.Send("zcard", Config.Namespace+RETRY_KEY)
for key, _ := range enqueued {
conn.Send("llen", fmt.Sprintf("%squeue:%s", Config.Namespace, key))
}
r, err := conn.Do("exec")
if err != nil {
Logger.Println("couldn't retrieve stats:", err)
}
results := r.([]interface{})
if len(results) == (3 + len(enqueued)) {
for index, result := range results {
if index == 0 && result != nil {
stats.Processed, _ = strconv.Atoi(string(result.([]byte)))
continue
}
if index == 1 && result != nil {
stats.Failed, _ = strconv.Atoi(string(result.([]byte)))
continue
}
if index == 2 && result != nil {
stats.Retries = result.(int64)
continue
}
queueIndex := 0
for key, _ := range enqueued {
if queueIndex == (index - 3) {
enqueued[key] = fmt.Sprintf("%d", result.(int64))
}
queueIndex++
}
}
}
body, _ := json.MarshalIndent(stats, "", " ")
fmt.Fprintln(w, string(body))
}