ALL

For reading and verification only. You may not copy, use, modify or redistribute this code. Full notice on the main page.

tunnel/drain_test.go

SHA-256: 8c60403b772237195c8c4f04a3ef0be5e9904b432891ee3a562581318b07e424
424 lines · 14054 bytes · View on GitHub

1package main
2
3import (
4 "encoding/json"
5 "os"
6 "path/filepath"
7 "strconv"
8 "strings"
9 "sync/atomic"
10 "syscall"
11 "testing"
12 "time"
13)
14
15// SIGTERM drains (the design plan, section 5): nothing new is accepted, what is open finishes, the devices are told to prefer the other node,
16// and the node is gone by the end of the quiet period.
17
18func drainAsync(e *testEnv, quiet time.Duration) (done chan struct{}, started time.Time) {
19 done = make(chan struct{})
20 started = time.Now()
21 go func() {
22 e.srv.Drain(quiet)
23 close(done)
24 }()
25 return done, started
26}
27
28func TestDrain_NothingNewIsAcceptedOpenStreamsContinueDrainIsDeliveredAndTheNodeLeavesWhenTheyEnd(t *testing.T) {
29 e := newTestEnv(t, nil)
30 d, other := e.newDevice(), e.newDevice()
31 dc := e.connectOK(d)
32 ds := dc.serve(nil)
33 oc := e.connectOK(other)
34 otherServer := oc.serve(nil)
35
36 c, err := e.clientTLS(d)
37 if err != nil {
38 t.Fatal(err)
39 }
40 if resp, err := postOver(c, d.host, []byte("before")); err != nil || string(resp) != "echo:before" {
41 t.Fatalf("%q %v", resp, err)
42 }
43
44 quiet := 8 * time.Second
45 done, started := drainAsync(e, quiet)
46 for name, dev := range map[string]*devServer{"first": ds, "second": otherServer} {
47 dev := dev
48 eventually(t, name+" told DRAIN", 3*time.Second, func() bool { return containsLine(dev.otherLines(), "DRAIN "+strconv.Itoa(int(quiet/time.Second))) })
49 }
50
51 // no new client: the TLS alert, no OPEN, nothing forwarded
52 opens := ds.openCount()
53 if _, err := e.clientTLS(d); err == nil {
54 t.Fatal("a client was accepted by a draining node")
55 }
56 if _, err := e.clientTLS(other); err == nil {
57 t.Fatal("a client of another device was accepted by a draining node")
58 }
59 time.Sleep(100 * time.Millisecond)
60 if ds.openCount() != opens || otherServer.openCount() != 0 {
61 t.Fatal("a draining node sent an OPEN")
62 }
63 // no new device: told to go to the other node
64 stranger := e.newDevice()
65 sc := e.connect(stranger)
66 if line := sc.hello(func() string { return e.session(stranger, nil) }, func(s, n string) string { return e.proof(stranger, s, n, nil) }); line != "ERR control_unavailable" {
67 t.Fatalf("a HELLO during a drain: %q", line)
68 }
69 // the stream that was open goes on, and so does a RENEW of the device that holds it
70 if resp, err := postOver(c, d.host, []byte("during")); err != nil || string(resp) != "echo:during" {
71 t.Fatalf("an open stream during a drain: %q %v", resp, err)
72 }
73 dc.drainFeed() // the DRAIN line the device was sent is still unread
74 if r := dc.renew(); !strings.HasPrefix(r, "OK ") {
75 t.Fatalf("renew during a drain: %q", r)
76 }
77 select {
78 case <-done:
79 t.Fatal("the node left with a stream open")
80 case <-time.After(300 * time.Millisecond):
81 }
82
83 // the last stream ends: the node closes the devices (CLOSE drain) and exits well inside the quiet period
84 _ = c.Close()
85 select {
86 case <-done:
87 case <-time.After(5 * time.Second):
88 t.Fatal("the node did not leave when its last stream ended")
89 }
90 if took := time.Since(started); took >= quiet {
91 t.Fatalf("it took %s, the quiet period is %s", took, quiet)
92 }
93 <-ds.done
94 if !containsLine(ds.otherLines(), "CLOSE drain") {
95 t.Fatalf("lines %q", ds.otherLines())
96 }
97 if _, err := e.dialNode(); err == nil {
98 t.Fatal("the node still listens after it drained")
99 }
100}
101
102func TestDrain_ALongStreamIsCutAtTheEndOfTheQuietPeriodAndNotBefore(t *testing.T) {
103 e := newTestEnv(t, nil)
104 d := e.newDevice()
105 dc := e.connectOK(d)
106 _ = dc.serve(nil)
107 stream := streamOpen(t, e, d)
108
109 quiet := 2 * time.Second
110 done, started := drainAsync(e, quiet)
111 time.Sleep(quiet / 2)
112 if stream.dead(50 * time.Millisecond) {
113 t.Fatal("the stream was cut before the quiet period was over")
114 }
115 select {
116 case <-done:
117 case <-time.After(quiet + 4*time.Second):
118 t.Fatal("the node did not exit by the end of the quiet period")
119 }
120 if took := time.Since(started); took < quiet-100*time.Millisecond || took > quiet+3*time.Second {
121 t.Fatalf("it took %s for a quiet period of %s", took, quiet)
122 }
123 if !stream.dead(2 * time.Second) {
124 t.Fatal("the stream outlived the node")
125 }
126}
127
128func TestDrain_AnAttachForAStreamThatAlreadyHasItsOpenIsStillServed(t *testing.T) {
129 e := newTestEnv(t, nil)
130 d := e.newDevice()
131 dc := e.connectOK(d)
132 ds := dc.serve(nil)
133 ds.noAttach = true // the device is slow: it has been sent the OPEN and has not attached yet
134
135 type result struct {
136 resp []byte
137 err error
138 }
139 got := make(chan result, 1)
140 go func() {
141 c, err := e.clientTLS(d)
142 if err != nil {
143 got <- result{nil, err}
144 return
145 }
146 defer c.Close()
147 resp, err := postOver(c, d.host, []byte("late"))
148 got <- result{resp, err}
149 }()
150 eventually(t, "the OPEN", 3*time.Second, func() bool { return ds.openCount() == 1 })
151
152 done, _ := drainAsync(e, 10*time.Second)
153 eventually(t, "draining", 2*time.Second, func() bool { return e.srv.Draining() })
154 open := ds.openList()[0]
155 ds.wg.Add(1)
156 go ds.attach(open.str("stream_id"), open.str("secret"))
157 select {
158 case r := <-got:
159 if r.err != nil || string(r.resp) != "echo:late" {
160 t.Fatalf("a client whose stream was being set up when the drain began: %q %v", r.resp, r.err)
161 }
162 case <-time.After(5 * time.Second):
163 t.Fatal("the stream that was being set up was not served")
164 }
165 select {
166 case <-done:
167 case <-time.After(5 * time.Second):
168 t.Fatal("no exit after the last stream")
169 }
170}
171
172func TestDrain_ADeviceThatIsAuthenticatingWhenTheDrainStartsIsToldToo(t *testing.T) {
173 // A HELLO past the gate whose authorize is still running when SIGTERM comes: its OK is followed by DRAIN, so it never serves as if nothing happened.
174 e := newTestEnv(t, nil)
175 holder := e.newDevice()
176 hc := e.connectOK(holder)
177 _ = hc.serve(nil)
178 held := streamOpen(t, e, holder) // a stream that keeps the node draining for a while
179 defer held.Close()
180 d := e.newDevice()
181 e.plan(d).delay = 400 * time.Millisecond
182 dc := e.connect(d)
183 dc.session = e.session(d, nil)
184 dc.send("HELLO 2 " + dc.session)
185 m, err := parseLine(dc.recv(), dirN2D)
186 if err != nil || m.str("verb") != "CHALLENGE" {
187 t.Fatal("no challenge")
188 }
189 dc.send("PROOF " + e.proof(d, dc.session, m.str("nonce"), nil))
190 time.Sleep(100 * time.Millisecond) // inside the slow authorize
191 done, _ := drainAsync(e, 5*time.Second)
192 eventually(t, "draining", 2*time.Second, func() bool { return e.srv.Draining() })
193 if line := dc.recv(); !strings.HasPrefix(line, "OK ") {
194 t.Fatalf("%q", line)
195 }
196 if line := dc.recv(); line != "DRAIN 5" {
197 t.Fatalf("after the OK of a device that authenticated during the drain: %q", line)
198 }
199 _ = dc.c.Close()
200 _ = held.Close()
201 <-done
202}
203
204func TestDrain_ARevokeInTheMiddleOfADrainStillCuts(t *testing.T) {
205 f := &scriptedFeed{}
206 e := withFeed(t, f)
207 d := e.newDevice()
208 dc := e.connectOK(d)
209 ds := dc.serve(nil)
210 stream := streamOpen(t, e, d)
211 done, _ := drainAsync(e, 6*time.Second)
212 eventually(t, "draining", 2*time.Second, func() bool { return e.srv.Draining() })
213 f.push(e.event(feedRevoke, d))
214 if !stream.dead(3 * time.Second) {
215 t.Fatal("a revoke during a drain did not cut the stream")
216 }
217 eventually(t, "CLOSE revoked", 2*time.Second, func() bool { return containsLine(ds.otherLines(), "CLOSE revoked") })
218 select {
219 case <-done:
220 case <-time.After(5 * time.Second):
221 t.Fatal("the node did not leave once nothing was open")
222 }
223}
224
225func TestDrain_ASecondDrainIsNoOp(t *testing.T) {
226 e := newTestEnv(t, nil)
227 done, _ := drainAsync(e, time.Second)
228 <-done
229 finished := make(chan struct{})
230 go func() { e.srv.Drain(time.Second); close(finished) }()
231 select {
232 case <-finished:
233 case <-time.After(2 * time.Second):
234 t.Fatal("a second Drain blocked")
235 }
236}
237
238// A call of 900 s that started before SIGTERM finishes; the quiet period is the policy's twenty minutes, on a clock the test moves.
239func TestDrain_ASigtermDuringA900SecondStreamLetsItFinishAndOneThatNeverEndsIsCutAtTwentyMinutes(t *testing.T) {
240 if policyData.Timing.DrainQuietS != 1200 {
241 t.Fatalf("the plan says 20 minutes, the policy says %d s", policyData.Timing.DrainQuietS)
242 }
243 for _, endsAt := range []time.Duration{900 * time.Second, 0} {
244 e := newClockEnv(t, nil)
245 d := e.newDevice()
246 dc := e.connectOK(d)
247 ds := dc.serve(nil)
248 stream := streamOpen(t, e, d)
249 var exited atomic.Bool
250 done := make(chan struct{})
251 go func() {
252 e.srv.Drain(e.srv.cfg.DrainQuiet)
253 exited.Store(true)
254 close(done)
255 }()
256 eventually(t, "DRAIN delivered", 3*time.Second, func() bool { return containsLine(ds.otherLines(), "DRAIN 1200") })
257
258 e.clk.Advance(899 * time.Second)
259 time.Sleep(400 * time.Millisecond)
260 if exited.Load() || stream.dead(50*time.Millisecond) {
261 t.Fatalf("ends at %s: the stream or the node was cut at 899 s", endsAt)
262 }
263 if endsAt > 0 {
264 // the call finishes at 900 s: the node leaves at once, 300 s early
265 e.clk.Advance(time.Second)
266 _ = stream.Close()
267 select {
268 case <-done:
269 case <-time.After(5 * time.Second):
270 t.Fatal("the node did not leave after the 900 s stream finished")
271 }
272 continue
273 }
274 e.clk.Advance(300 * time.Second) // 1199 s
275 time.Sleep(400 * time.Millisecond)
276 if exited.Load() {
277 t.Fatal("the node left before the quiet period was over")
278 }
279 e.clk.Advance(2 * time.Second) // 1201 s
280 select {
281 case <-done:
282 case <-time.After(6 * time.Second):
283 t.Fatal("a stream that never ends kept the node past the quiet period")
284 }
285 if !stream.dead(2 * time.Second) {
286 t.Fatal("the stream outlived the node")
287 }
288 }
289}
290
291// ---------------------------------------------------------------------------------------------------------------------------- the real process
292
293func TestTestNodeBinary_SigtermDrainsAndTheProcessExitsInsideTheQuietPeriod(t *testing.T) {
294 for _, longStream := range []bool{false, true} {
295 p := launchTestNode(t, 600, "-drain-quiet", "3s")
296 e, d, cmd := p.e, p.d, p.cmd
297 dc := e.connectOK(d)
298 ds := dc.serve(nil)
299 stream := streamOpen(t, e, d)
300 if !longStream {
301 _ = stream.Close()
302 }
303 start := time.Now()
304 if err := cmd.Process.Signal(syscall.SIGTERM); err != nil {
305 t.Fatal(err)
306 }
307 eventually(t, "DRAIN 3 delivered", 3*time.Second, func() bool { return containsLine(ds.otherLines(), "DRAIN 3") })
308 if longStream {
309 // no new client; the open one goes on
310 if _, err := e.clientTLS(d); err == nil {
311 t.Fatal("a client was accepted by a draining process")
312 }
313 if resp, err := postOver(stream.Conn, d.host, []byte("still")); err != nil || string(resp) != "echo:still" {
314 t.Fatalf("the open stream during the drain: %q %v", resp, err)
315 }
316 }
317 select {
318 case err := <-p.exited:
319 if err != nil {
320 t.Fatalf("exit: %v\n%s", err, p.logs.String())
321 }
322 case <-time.After(10 * time.Second):
323 t.Fatalf("the process did not exit (long stream: %v)\n%s", longStream, p.logs.String())
324 }
325 took := time.Since(start)
326 switch {
327 case !longStream && took > 2*time.Second:
328 t.Errorf("an idle node took %s to drain", took)
329 case longStream && (took < 2500*time.Millisecond || took > 3*time.Second+3*time.Second):
330 t.Errorf("a node with a stream that never ends left after %s; the quiet period is 3 s", took)
331 }
332 <-ds.done
333 if !containsLine(ds.otherLines(), "CLOSE drain") {
334 t.Errorf("lines %q", ds.otherLines())
335 }
336 }
337}
338
339func TestTestNodeBinary_ASecondSigtermDuringADrainStopsWaiting(t *testing.T) {
340 p := launchTestNode(t, 600, "-drain-quiet", "5m")
341 dc := p.e.connectOK(p.d)
342 _ = dc.serve(nil)
343 stream := streamOpen(t, p.e, p.d)
344 defer stream.Close()
345 if err := p.cmd.Process.Signal(syscall.SIGTERM); err != nil {
346 t.Fatal(err)
347 }
348 time.Sleep(500 * time.Millisecond)
349 select {
350 case <-p.exited:
351 t.Fatal("the process left with a stream open and five minutes of quiet period left")
352 default:
353 }
354 if err := p.cmd.Process.Signal(syscall.SIGTERM); err != nil {
355 t.Fatal(err)
356 }
357 select {
358 case err := <-p.exited:
359 if err != nil {
360 t.Fatalf("exit: %v", err)
361 }
362 case <-time.After(5 * time.Second):
363 t.Fatalf("a second SIGTERM did not stop the drain\n%s", p.logs.String())
364 }
365}
366
367func TestTestNodeBinary_AFeedFileRevokesAStreamAndTheKillSwitchClosesEverything(t *testing.T) {
368 dir := t.TempDir()
369 feed := filepath.Join(dir, "feed.json")
370 writeFeed := func(epoch int, events ...map[string]any) {
371 t.Helper()
372 list := []any{}
373 for _, ev := range events {
374 list = append(list, ev)
375 }
376 b, _ := json.Marshal(map[string]any{"epoch": epoch, "events": list})
377 if err := os.WriteFile(feed+".tmp", b, 0o600); err != nil {
378 t.Fatal(err)
379 }
380 if err := os.Rename(feed+".tmp", feed); err != nil {
381 t.Fatal(err)
382 }
383 }
384 writeFeed(0)
385 p := launchTestNode(t, 600, "-test-feed", feed, "-feed-poll", "100ms")
386 e, d := p.e, p.d
387 dc := e.connectOK(d)
388 ds := dc.serve(nil)
389 stream := streamOpen(t, e, d)
390
391 at := time.Now().Unix()
392 writeFeed(1, map[string]any{"epoch": 1, "type": "revoke", "device_id": d.id, "label": d.label, "at": at})
393 eventually(t, "CLOSE revoked", 5*time.Second, func() bool { return containsLine(ds.otherLines(), "CLOSE revoked") })
394 if !stream.dead(3 * time.Second) {
395 t.Fatal("the revoked device's stream is still open")
396 }
397 // the kill switch: nobody is served
398 writeFeed(2, map[string]any{"epoch": 1, "type": "revoke", "device_id": d.id, "label": d.label, "at": at},
399 map[string]any{"epoch": 2, "type": "kill", "device_id": nil, "label": nil, "at": at})
400 time.Sleep(600 * time.Millisecond)
401 c := e.connect(d)
402 if line := c.hello(func() string { return e.session(d, nil) }, func(s, n string) string { return e.proof(d, s, n, nil) }); line != "ERR control_unavailable" {
403 t.Fatalf("a HELLO while the kill switch is on: %q", line)
404 }
405 writeFeed(3, map[string]any{"epoch": 3, "type": "kill_clear", "device_id": nil, "label": nil, "at": at})
406 eventually(t, "served again", 5*time.Second, func() bool {
407 cc := e.connect(d)
408 return strings.HasPrefix(cc.hello(func() string { return e.session(d, nil) }, func(s, n string) string { return e.proof(d, s, n, nil) }), "OK ")
409 })
410 if strings.Contains(p.logs.String(), d.id) || strings.Contains(p.logs.String(), d.label) {
411 t.Fatal("the process logs name a device")
412 }
413}
414
415// drainFeed discards what the node already sent and the test did not read.
416func (dc *devConn) drainFeed() {
417 for {
418 select {
419 case <-dc.feed:
420 default:
421 return
422 }
423 }
424}