For reading and verification only. You may not copy, use, modify or redistribute this code. Full notice on the main page.
tunnel/storm_test.go
| 1 | package main |
| 2 | |
| 3 | import ( |
| 4 | "bufio" |
| 5 | "fmt" |
| 6 | "io" |
| 7 | "strings" |
| 8 | "sync" |
| 9 | "sync/atomic" |
| 10 | "syscall" |
| 11 | "testing" |
| 12 | "time" |
| 13 | ) |
| 14 | |
| 15 | // A node that restarts meets every device at once. The control plane is asked by at most authorize_max_concurrent calls at a time, what waits is |
| 16 | // bounded, the rest is told to try again, and the devices (who back off) all get in. |
| 17 | |
| 18 | func TestStorm_AFleetReconnectsAtOnceAfterANodeRestartAndTheControlPlaneNeverSeesMoreThanTheBound(t *testing.T) { |
| 19 | var lim syscall.Rlimit |
| 20 | _ = syscall.Getrlimit(syscall.RLIMIT_NOFILE, &lim) |
| 21 | devices := 1000 |
| 22 | if lim.Cur < 8192 { |
| 23 | devices = max(60, int(lim.Cur)/12) |
| 24 | } |
| 25 | e, cp := httpNode(t, nil, func(c *Config) { |
| 26 | c.HelloTimeout, c.FirstLineTimeout, c.AuthorizeTimeout = 10*time.Second, 20*time.Second, 10*time.Second |
| 27 | c.MaxUnauth = 64 // small, so that the storm really exceeds what may wait for the control plane |
| 28 | }) |
| 29 | cp.mu.Lock() |
| 30 | cp.delay = 25 * time.Millisecond |
| 31 | cp.mu.Unlock() |
| 32 | all := make([]*testDevice, devices) |
| 33 | for i := range all { |
| 34 | all[i] = e.newDevice() |
| 35 | } |
| 36 | |
| 37 | var connected, refusedBusy atomic.Int64 |
| 38 | var wg sync.WaitGroup |
| 39 | for _, d := range all { |
| 40 | wg.Add(1) |
| 41 | go func(d *testDevice) { |
| 42 | defer wg.Done() |
| 43 | for attempt := 0; attempt < 60; attempt++ { |
| 44 | line := helloOnce(e, d) |
| 45 | switch { |
| 46 | case strings.HasPrefix(line, "OK "): |
| 47 | connected.Add(1) |
| 48 | return |
| 49 | case line == "ERR control_unavailable" || strings.HasPrefix(line, "unreachable"): |
| 50 | refusedBusy.Add(1) |
| 51 | time.Sleep(time.Duration(30+attempt*10) * time.Millisecond) // the device's backoff |
| 52 | default: |
| 53 | t.Errorf("a device was refused with %q", line) |
| 54 | return |
| 55 | } |
| 56 | } |
| 57 | t.Errorf("a device never got in") |
| 58 | }(d) |
| 59 | } |
| 60 | wg.Wait() |
| 61 | if int(connected.Load()) != devices { |
| 62 | t.Fatalf("%d of %d devices connected", connected.Load(), devices) |
| 63 | } |
| 64 | if peak := cp.maxInflight(); peak > policyData.Authorization.AuthorizeMaxConcur { |
| 65 | t.Fatalf("the control plane saw %d authorizations at once, the bound is %d", peak, policyData.Authorization.AuthorizeMaxConcur) |
| 66 | } |
| 67 | if refusedBusy.Load() == 0 { |
| 68 | t.Log("(the storm never exceeded what may wait; the bound was still respected)") |
| 69 | } |
| 70 | for _, d := range all[:5] { |
| 71 | if !e.srv.Online(d.host) { |
| 72 | t.Fatal("a connected device is not online") |
| 73 | } |
| 74 | } |
| 75 | t.Log(fmt.Sprintf("%d devices, %d told to try again, peak %d at the control plane", devices, refusedBusy.Load(), cp.maxInflight())) |
| 76 | } |
| 77 | |
| 78 | // helloOnce is a whole HELLO of d on its own connection, as a device does it: the node's final line, or "unreachable: ..." when the node would not |
| 79 | // even take the connection (it is full of connections that have not authenticated yet). The connection is left open: a device that got in stays in. |
| 80 | func helloOnce(e *testEnv, d *testDevice) string { |
| 81 | c, err := e.dialNode() |
| 82 | if err != nil { |
| 83 | return "unreachable: " + err.Error() |
| 84 | } |
| 85 | br := bufio.NewReader(c) |
| 86 | read := func() string { |
| 87 | _ = c.SetReadDeadline(time.Now().Add(15 * time.Second)) |
| 88 | l, err := br.ReadString('\n') |
| 89 | if err != nil { |
| 90 | return "unreachable: " + err.Error() |
| 91 | } |
| 92 | return strings.TrimSuffix(l, "\n") |
| 93 | } |
| 94 | session := e.session(d, nil) |
| 95 | _, _ = io.WriteString(c, "HELLO 2 "+session+"\n") |
| 96 | line := read() |
| 97 | if nonce, ok := strings.CutPrefix(line, "CHALLENGE "); ok { |
| 98 | _, _ = io.WriteString(c, "PROOF "+e.proof(d, session, nonce, nil)+"\n") |
| 99 | line = read() |
| 100 | } |
| 101 | if !strings.HasPrefix(line, "OK ") { |
| 102 | _ = c.Close() |
| 103 | } else { |
| 104 | e.t.Cleanup(func() { _ = c.Close() }) |
| 105 | } |
| 106 | return line |
| 107 | } |