-
-
Notifications
You must be signed in to change notification settings - Fork 2.5k
/
supervisor.go
385 lines (340 loc) · 11.3 KB
/
supervisor.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
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
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
package host
import (
"context"
"crypto/tls"
"errors"
"net"
"net/http"
"strings"
"sync"
"sync/atomic"
"time"
"golang.org/x/crypto/acme/autocert"
"github.com/kataras/iris/v12/core/netutil"
)
// Configurator provides an easy way to modify
// the Supervisor.
//
// Look the `Configure` func for more.
type Configurator func(su *Supervisor)
// Supervisor is the wrapper and the manager for a compatible server
// and it's relative actions, called Tasks.
//
// Interfaces are separated to return relative functionality to them.
type Supervisor struct {
Server *http.Server
closedManually int32 // future use, accessed atomically (non-zero means we've called the Shutdown)
manuallyTLS bool // we need that in order to determinate what to output on the console before the server begin.
shouldWait int32 // non-zero means that the host should wait for unblocking
unblockChan chan struct{}
mu sync.Mutex
onServe []func(TaskHost)
// IgnoreErrors should contains the errors that should be ignored
// on both serve functions return statements and error handlers.
//
// i.e: http.ErrServerClosed.Error().
//
// Note that this will match the string value instead of the equality of the type's variables.
//
// Defaults to empty.
IgnoredErrors []string
onErr []func(error)
onShutdown []func()
}
// New returns a new host supervisor
// based on a native net/http "srv".
//
// It contains all native net/http's Server methods.
// Plus you can add tasks on specific events.
// It has its own flow, which means that you can prevent
// to return and exit and restore the flow too.
func New(srv *http.Server) *Supervisor {
return &Supervisor{
Server: srv,
unblockChan: make(chan struct{}, 1),
}
}
// Configure accepts one or more `Configurator`.
// With this function you can use simple functions
// that are spread across your app to modify
// the supervisor, these Configurators can be
// used on any Supervisor instance.
//
// Look `Configurator` too.
//
// Returns itself.
func (su *Supervisor) Configure(configurators ...Configurator) *Supervisor {
for _, conf := range configurators {
conf(su)
}
return su
}
// DeferFlow defers the flow of the exeuction,
// i.e: when server should return error and exit
// from app, a DeferFlow call inside a Task
// can wait for a `RestoreFlow` to exit or not exit if
// host's server is "fixed".
//
// See `RestoreFlow` too.
func (su *Supervisor) DeferFlow() {
atomic.StoreInt32(&su.shouldWait, 1)
}
// RestoreFlow restores the flow of the execution,
// if called without a `DeferFlow` call before
// then it does nothing.
// See tests to understand how that can be useful on specific cases.
//
// See `DeferFlow` too.
func (su *Supervisor) RestoreFlow() {
if su.isWaiting() {
atomic.StoreInt32(&su.shouldWait, 0)
su.mu.Lock()
su.unblockChan <- struct{}{}
su.mu.Unlock()
}
}
func (su *Supervisor) isWaiting() bool {
return atomic.LoadInt32(&su.shouldWait) != 0
}
func (su *Supervisor) newListener() (net.Listener, error) {
// this will not work on "unix" as network
// because UNIX doesn't supports the kind of
// restarts we may want for the server.
//
// User still be able to call .Serve instead.
// l, err := netutil.TCPKeepAlive(su.Server.Addr)
l, err := netutil.TCP(su.Server.Addr)
if err != nil {
return nil, err
}
// here we can check for sure, without the need of the supervisor's `manuallyTLS` field.
if netutil.IsTLS(su.Server) {
// means tls
tlsl := tls.NewListener(l, su.Server.TLSConfig)
return tlsl, nil
}
return l, nil
}
// RegisterOnError registers a function to call when errors occurred by the underline http server.
func (su *Supervisor) RegisterOnError(cb func(error)) {
su.mu.Lock()
su.onErr = append(su.onErr, cb)
su.mu.Unlock()
}
func (su *Supervisor) validateErr(err error) error {
if err == nil {
return nil
}
su.mu.Lock()
defer su.mu.Unlock()
for _, e := range su.IgnoredErrors {
if err.Error() == e {
return nil
}
}
return err
}
func (su *Supervisor) notifyErr(err error) {
err = su.validateErr(err)
if err != nil {
su.mu.Lock()
for _, f := range su.onErr {
go f(err)
}
su.mu.Unlock()
}
}
// RegisterOnServe registers a function to call on
// Serve/ListenAndServe/ListenAndServeTLS/ListenAndServeAutoTLS.
func (su *Supervisor) RegisterOnServe(cb func(TaskHost)) {
su.mu.Lock()
su.onServe = append(su.onServe, cb)
su.mu.Unlock()
}
func (su *Supervisor) notifyServe(host TaskHost) {
su.mu.Lock()
for _, f := range su.onServe {
go f(host)
}
su.mu.Unlock()
}
// Remove all channels, do it with events
// or with channels but with a different channel on each task proc
// I don't know channels are not so safe, when go func and race risk..
// so better with callbacks....
func (su *Supervisor) supervise(blockFunc func() error) error {
host := createTaskHost(su)
su.notifyServe(host)
err := blockFunc()
su.notifyErr(err)
if su.isWaiting() {
blockStatement:
for {
select {
case <-su.unblockChan:
break blockStatement
}
}
}
return su.validateErr(err)
}
// Serve accepts incoming connections on the Listener l, creating a
// new service goroutine for each. The service goroutines read requests and
// then call su.server.Handler to reply to them.
//
// For HTTP/2 support, server.TLSConfig should be initialized to the
// provided listener's TLS Config before calling Serve. If
// server.TLSConfig is non-nil and doesn't include the string "h2" in
// Config.NextProtos, HTTP/2 support is not enabled.
//
// Serve always returns a non-nil error. After Shutdown or Close, the
// returned error is http.ErrServerClosed.
func (su *Supervisor) Serve(l net.Listener) error {
return su.supervise(func() error { return su.Server.Serve(l) })
}
// ListenAndServe listens on the TCP network address addr
// and then calls Serve with handler to handle requests
// on incoming connections.
// Accepted connections are configured to enable TCP keep-alives.
func (su *Supervisor) ListenAndServe() error {
l, err := su.newListener()
if err != nil {
return err
}
return su.Serve(l)
}
// ListenAndServeTLS acts identically to ListenAndServe, except that it
// expects HTTPS connections. Additionally, files containing a certificate and
// matching private key for the server must be provided. If the certificate
// is signed by a certificate authority, the certFile should be the concatenation
// of the server's certificate, any intermediates, and the CA's certificate.
func (su *Supervisor) ListenAndServeTLS(certFile string, keyFile string) error {
su.manuallyTLS = true
if certFile != "" && keyFile != "" {
cfg := new(tls.Config)
var err error
cfg.Certificates = make([]tls.Certificate, 1)
if cfg.Certificates[0], err = tls.LoadX509KeyPair(certFile, keyFile); err != nil {
return err
}
// manually inserted as pre-go 1.9 for any case.
cfg.NextProtos = []string{"h2", "http/1.1"}
su.Server.TLSConfig = cfg
// It does nothing more than the su.Server.ListenAndServeTLS anymore.
// - no hurt if we let it as it is
// - no problem if we remove it as well
// but let's comment this as proposed, fewer code is better:
// return su.ListenAndServe()
}
if su.Server.TLSConfig == nil {
return errors.New("empty certFile or keyFile and Server.TLSConfig")
}
return su.supervise(func() error { return su.Server.ListenAndServeTLS("", "") })
}
// ListenAndServeAutoTLS acts identically to ListenAndServe, except that it
// expects HTTPS connections. Server's certificates are auto generated from LETSENCRYPT using
// the golang/x/net/autocert package.
//
// The whitelisted domains are separated by whitespace in "domain" argument, i.e "iris-go.com".
// If empty, all hosts are currently allowed. This is not recommended,
// as it opens a potential attack where clients connect to a server
// by IP address and pretend to be asking for an incorrect host name.
// Manager will attempt to obtain a certificate for that host, incorrectly,
// eventually reaching the CA's rate limit for certificate requests
// and making it impossible to obtain actual certificates.
//
// For an "e-mail" use a non-public one, letsencrypt needs that for your own security.
//
// The "cacheDir" is being, optionally, used to provide cache
// stores and retrieves previously-obtained certificates.
// If empty, certs will only be cached for the lifetime of the auto tls manager.
//
// Note: The domain should be like "iris-go.com www.iris-go.com",
// the e-mail like "[email protected]" and the cacheDir like "letscache"
// The `ListenAndServeAutoTLS` will start a new server for you,
// which will redirect all http versions to their https, including subdomains as well.
func (su *Supervisor) ListenAndServeAutoTLS(domain string, email string, cacheDir string) error {
var (
cache autocert.Cache
hostPolicy autocert.HostPolicy
)
if cacheDir != "" {
cache = autocert.DirCache(cacheDir)
}
if domain != "" {
domains := strings.Split(domain, " ")
hostPolicy = autocert.HostWhitelist(domains...)
}
autoTLSManager := &autocert.Manager{
Prompt: autocert.AcceptTOS,
HostPolicy: hostPolicy,
Email: email,
Cache: cache,
ForceRSA: true,
}
srv2 := &http.Server{
ReadTimeout: 30 * time.Second,
WriteTimeout: 60 * time.Second,
Addr: ":http",
Handler: autoTLSManager.HTTPHandler(nil), // nil for redirect.
}
// register a shutdown callback to this
// supervisor in order to close the "secondary redirect server" as well.
su.RegisterOnShutdown(func() {
// give it some time to close itself...
timeout := 5 * time.Second
ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
srv2.Shutdown(ctx)
})
go srv2.ListenAndServe()
su.Server.TLSConfig = &tls.Config{
MinVersion: tls.VersionTLS10,
GetCertificate: autoTLSManager.GetCertificate,
PreferServerCipherSuites: true,
// Keep the defaults.
CurvePreferences: []tls.CurveID{
tls.X25519,
tls.CurveP256,
tls.CurveP384,
tls.CurveP521,
},
}
return su.ListenAndServeTLS("", "")
}
// RegisterOnShutdown registers a function to call on Shutdown.
// This can be used to gracefully shutdown connections that have
// undergone NPN/ALPN protocol upgrade or that have been hijacked.
// This function should start protocol-specific graceful shutdown,
// but should not wait for shutdown to complete.
func (su *Supervisor) RegisterOnShutdown(cb func()) {
// when go1.9: replace the following lines with su.Server.RegisterOnShutdown(f)
su.mu.Lock()
su.onShutdown = append(su.onShutdown, cb)
su.mu.Unlock()
}
func (su *Supervisor) notifyShutdown() {
// when go1.9: remove the lines below
su.mu.Lock()
for _, f := range su.onShutdown {
go f()
}
su.mu.Unlock()
// end
}
// Shutdown gracefully shuts down the server without interrupting any
// active connections. Shutdown works by first closing all open
// listeners, then closing all idle connections, and then waiting
// indefinitely for connections to return to idle and then shut down.
// If the provided context expires before the shutdown is complete,
// then the context's error is returned.
//
// Shutdown does not attempt to close nor wait for hijacked
// connections such as WebSockets. The caller of Shutdown should
// separately notify such long-lived connections of shutdown and wait
// for them to close, if desired.
func (su *Supervisor) Shutdown(ctx context.Context) error {
atomic.AddInt32(&su.closedManually, 1) // future-use
su.notifyShutdown()
return su.Server.Shutdown(ctx)
}