forked from hibiken/asynqmon
-
Notifications
You must be signed in to change notification settings - Fork 0
/
queue_handlers.go
138 lines (125 loc) · 3.89 KB
/
queue_handlers.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
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
package asynqmon
import (
"encoding/json"
"errors"
"net/http"
"github.com/gorilla/mux"
"github.com/hibiken/asynq"
)
// ****************************************************************************
// This file defines:
// - http.Handler(s) for queue related endpoints
// ****************************************************************************
func newListQueuesHandlerFunc(inspector *asynq.Inspector) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
qnames, err := inspector.Queues()
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
snapshots := make([]*queueStateSnapshot, len(qnames))
for i, qname := range qnames {
qinfo, err := inspector.GetQueueInfo(qname)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
snapshots[i] = toQueueStateSnapshot(qinfo)
}
payload := map[string]interface{}{"queues": snapshots}
json.NewEncoder(w).Encode(payload)
}
}
func newGetQueueHandlerFunc(inspector *asynq.Inspector) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
vars := mux.Vars(r)
qname := vars["qname"]
payload := make(map[string]interface{})
qinfo, err := inspector.GetQueueInfo(qname)
if err != nil {
// TODO: Check for queue not found error.
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
payload["current"] = toQueueStateSnapshot(qinfo)
// TODO: make this n a variable
data, err := inspector.History(qname, 10)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
var dailyStats []*dailyStats
for _, s := range data {
dailyStats = append(dailyStats, toDailyStats(s))
}
payload["history"] = dailyStats
json.NewEncoder(w).Encode(payload)
}
}
func newDeleteQueueHandlerFunc(inspector *asynq.Inspector) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
vars := mux.Vars(r)
qname := vars["qname"]
if err := inspector.DeleteQueue(qname, false); err != nil {
if errors.Is(err, asynq.ErrQueueNotFound) {
http.Error(w, err.Error(), http.StatusNotFound)
return
}
if errors.Is(err, asynq.ErrQueueNotEmpty) {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
w.WriteHeader(http.StatusNoContent)
}
}
func newPauseQueueHandlerFunc(inspector *asynq.Inspector) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
vars := mux.Vars(r)
qname := vars["qname"]
if err := inspector.PauseQueue(qname); err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
w.WriteHeader(http.StatusNoContent)
}
}
func newResumeQueueHandlerFunc(inspector *asynq.Inspector) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
vars := mux.Vars(r)
qname := vars["qname"]
if err := inspector.UnpauseQueue(qname); err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
w.WriteHeader(http.StatusNoContent)
}
}
type listQueueStatsResponse struct {
Stats map[string][]*dailyStats `json:"stats"`
}
func newListQueueStatsHandlerFunc(inspector *asynq.Inspector) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
qnames, err := inspector.Queues()
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
resp := listQueueStatsResponse{Stats: make(map[string][]*dailyStats)}
const numdays = 90 // Get stats for the last 90 days.
for _, qname := range qnames {
stats, err := inspector.History(qname, numdays)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
resp.Stats[qname] = toDailyStatsList(stats)
}
if err := json.NewEncoder(w).Encode(resp); err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
}
}