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
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
|
package builds
import (
"encoding/json"
"errors"
"net/http"
"time"
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/promauto"
"gitlab.com/gitlab-org/gitlab/workhorse/internal/helper"
"gitlab.com/gitlab-org/gitlab/workhorse/internal/redis"
)
const (
maxRegisterBodySize = 32 * 1024
runnerBuildQueue = "runner:build_queue:"
runnerBuildQueueHeaderKey = "Gitlab-Ci-Builds-Polling"
runnerBuildQueueHeaderValue = "yes"
)
var (
registerHandlerRequests = promauto.NewCounterVec(
prometheus.CounterOpts{
Name: "gitlab_workhorse_builds_register_handler_requests",
Help: "Describes how many requests in different states hit a register handler",
},
[]string{"status"},
)
registerHandlerOpen = promauto.NewGaugeVec(
prometheus.GaugeOpts{
Name: "gitlab_workhorse_builds_register_handler_open",
Help: "Describes how many requests is currently open in given state",
},
[]string{"state"},
)
registerHandlerOpenAtReading = registerHandlerOpen.WithLabelValues("reading")
registerHandlerOpenAtProxying = registerHandlerOpen.WithLabelValues("proxying")
registerHandlerOpenAtWatching = registerHandlerOpen.WithLabelValues("watching")
registerHandlerBodyReadErrors = registerHandlerRequests.WithLabelValues("body-read-error")
registerHandlerBodyParseErrors = registerHandlerRequests.WithLabelValues("body-parse-error")
registerHandlerMissingValues = registerHandlerRequests.WithLabelValues("missing-values")
registerHandlerWatchErrors = registerHandlerRequests.WithLabelValues("watch-error")
registerHandlerAlreadyChangedRequests = registerHandlerRequests.WithLabelValues("already-changed")
registerHandlerSeenChangeRequests = registerHandlerRequests.WithLabelValues("seen-change")
registerHandlerTimeoutRequests = registerHandlerRequests.WithLabelValues("timeout")
registerHandlerNoChangeRequests = registerHandlerRequests.WithLabelValues("no-change")
)
type largeBodyError struct{ error }
type WatchKeyHandler func(key, value string, timeout time.Duration) (redis.WatchKeyStatus, error)
type runnerRequest struct {
Token string `json:"token,omitempty"`
LastUpdate string `json:"last_update,omitempty"`
}
func readRunnerBody(w http.ResponseWriter, r *http.Request) ([]byte, error) {
registerHandlerOpenAtReading.Inc()
defer registerHandlerOpenAtReading.Dec()
return helper.ReadRequestBody(w, r, maxRegisterBodySize)
}
func readRunnerRequest(r *http.Request, body []byte) (*runnerRequest, error) {
if !helper.IsApplicationJson(r) {
return nil, errors.New("invalid content-type received")
}
var runnerRequest runnerRequest
err := json.Unmarshal(body, &runnerRequest)
if err != nil {
return nil, err
}
return &runnerRequest, nil
}
func proxyRegisterRequest(h http.Handler, w http.ResponseWriter, r *http.Request) {
registerHandlerOpenAtProxying.Inc()
defer registerHandlerOpenAtProxying.Dec()
h.ServeHTTP(w, r)
}
func watchForRunnerChange(watchHandler WatchKeyHandler, token, lastUpdate string, duration time.Duration) (redis.WatchKeyStatus, error) {
registerHandlerOpenAtWatching.Inc()
defer registerHandlerOpenAtWatching.Dec()
return watchHandler(runnerBuildQueue+token, lastUpdate, duration)
}
func RegisterHandler(h http.Handler, watchHandler WatchKeyHandler, pollingDuration time.Duration) http.Handler {
if pollingDuration == 0 {
return h
}
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set(runnerBuildQueueHeaderKey, runnerBuildQueueHeaderValue)
requestBody, err := readRunnerBody(w, r)
if err != nil {
registerHandlerBodyReadErrors.Inc()
helper.RequestEntityTooLarge(w, r, &largeBodyError{err})
return
}
newRequest := helper.CloneRequestWithNewBody(r, requestBody)
runnerRequest, err := readRunnerRequest(r, requestBody)
if err != nil {
registerHandlerBodyParseErrors.Inc()
proxyRegisterRequest(h, w, newRequest)
return
}
if runnerRequest.Token == "" || runnerRequest.LastUpdate == "" {
registerHandlerMissingValues.Inc()
proxyRegisterRequest(h, w, newRequest)
return
}
result, err := watchForRunnerChange(watchHandler, runnerRequest.Token,
runnerRequest.LastUpdate, pollingDuration)
if err != nil {
registerHandlerWatchErrors.Inc()
proxyRegisterRequest(h, w, newRequest)
return
}
switch result {
// It means that we detected a change before starting watching on change,
// We proxy request to Rails, to see whether we have a build to receive
case redis.WatchKeyStatusAlreadyChanged:
registerHandlerAlreadyChangedRequests.Inc()
proxyRegisterRequest(h, w, newRequest)
// It means that we detected a change after watching.
// We could potentially proxy request to Rails, but...
// We can end-up with unreliable responses,
// as don't really know whether ResponseWriter is still in a sane state,
// for example the connection is dead
case redis.WatchKeyStatusSeenChange:
registerHandlerSeenChangeRequests.Inc()
w.WriteHeader(http.StatusNoContent)
// When we receive one of these statuses, it means that we detected no change,
// so we return to runner 204, which means nothing got changed,
// and there's no new builds to process
case redis.WatchKeyStatusTimeout:
registerHandlerTimeoutRequests.Inc()
w.WriteHeader(http.StatusNoContent)
case redis.WatchKeyStatusNoChange:
registerHandlerNoChangeRequests.Inc()
w.WriteHeader(http.StatusNoContent)
}
})
}
|
