| 1 | package edge |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "net/http" |
| 7 | "net/http/httptest" |
| 8 | "os" |
| 9 | "path/filepath" |
| 10 | "strings" |
| 11 | "testing" |
| 12 | "time" |
| 13 | |
| 14 | "golang.org/x/crypto/acme/autocert" |
| 15 | |
| 16 | "github.com/carlosframework/platform/internal/lbhealth" |
| 17 | "github.com/carlosframework/platform/internal/registry" |
| 18 | "github.com/carlosframework/platform/internal/release" |
| 19 | ) |
| 20 | |
| 21 | // sweeper is an Activator whose only working method is the sweep beat. |
| 22 | type sweeper struct { |
| 23 | Activator |
| 24 | at time.Time |
| 25 | } |
| 26 | |
| 27 | func (s sweeper) LastSweep() time.Time { return s.at } |
| 28 | |
| 29 | func lbEnv(v string) func(string) string { |
| 30 | return func(k string) string { |
| 31 | if k == "CARLOS_REGION_HA" { |
| 32 | return v |
| 33 | } |
| 34 | return "" |
| 35 | } |
| 36 | } |
| 37 | |
| 38 | func TestLBHealthStateFollowsRegionHA(t *testing.T) { |
| 39 | if s := lbHealthFromEnv(lbEnv(""), nil); s != nil { |
| 40 | t.Fatal("region HA unset armed the health answer") |
| 41 | } |
| 42 | if s := lbHealthFromEnv(lbEnv("0"), nil); s != nil { |
| 43 | t.Fatal("region HA off armed the health answer") |
| 44 | } |
| 45 | at := time.Now() |
| 46 | s := lbHealthFromEnv(lbEnv("1"), nil) |
| 47 | if s != lbhealth.Default { |
| 48 | t.Fatal("region HA on: not the process's state") |
| 49 | } |
| 50 | t.Cleanup(func() { lbhealth.Default.SetActivator(nil) }) |
| 51 | lbHealthFromEnv(lbEnv("true"), sweeper{at: at}) |
| 52 | if got := lbhealth.Default.Failing(at); strings.Contains(strings.Join(got, ","), lbhealth.CondActivator) { |
| 53 | t.Fatalf("the activator's sweep was not wired: %v", got) |
| 54 | } |
| 55 | } |
| 56 | |
| 57 | // fallbackGet drives the :80 handler with one request. |
| 58 | func fallbackGet(e *Edge, method, path, host string) *httptest.ResponseRecorder { |
| 59 | w := httptest.NewRecorder() |
| 60 | r := httptest.NewRequest(method, path, nil) |
| 61 | r.Host = host |
| 62 | e.httpFallbackHandler().ServeHTTP(w, r) |
| 63 | return w |
| 64 | } |
| 65 | |
| 66 | // healthyState meets every condition now. |
| 67 | func healthyState() *lbhealth.State { |
| 68 | s := &lbhealth.State{} |
| 69 | now := time.Now() |
| 70 | s.SetHosting(true) |
| 71 | s.MapLoaded() |
| 72 | s.KeyLoaded() |
| 73 | s.CertsLoaded() |
| 74 | s.EdgeBeat(now) |
| 75 | s.RefreshAttempted(now) |
| 76 | s.SetActivator(func() time.Time { return now }) |
| 77 | return s |
| 78 | } |
| 79 | |
| 80 | // The answer is on :80, for any host (the load balancer sends the |
| 81 | // target's address), GET or HEAD, ahead of the redirect and unknown-host |
| 82 | // branches. With region HA off the path is ordinary :80 traffic. |
| 83 | func TestLBHealthAnswersOnPort80ForAnyHost(t *testing.T) { |
| 84 | e, err := New(openReg(t), nil) |
| 85 | if err != nil { |
| 86 | t.Fatal(err) |
| 87 | } |
| 88 | t.Cleanup(e.probers.Stop) |
| 89 | e.lbHealth = healthyState() |
| 90 | for _, host := range []string{"10.0.1.7", "10.0.1.7:80", "unknown.example.com"} { |
| 91 | w := fallbackGet(e, http.MethodGet, lbhealth.Path, host) |
| 92 | if w.Code != http.StatusOK || w.Body.String() != "ok\n" { |
| 93 | t.Fatalf("host %s: %d %q", host, w.Code, w.Body.String()) |
| 94 | } |
| 95 | } |
| 96 | if w := fallbackGet(e, http.MethodHead, lbhealth.Path, "10.0.1.7"); w.Code != http.StatusOK { |
| 97 | t.Fatalf("HEAD: %d", w.Code) |
| 98 | } |
| 99 | joining := healthyState() |
| 100 | joining.SetHosting(false) |
| 101 | e.lbHealth = joining |
| 102 | if w := fallbackGet(e, http.MethodGet, lbhealth.Path, "10.0.1.7"); w.Code != http.StatusServiceUnavailable || !strings.Contains(w.Body.String(), lbhealth.CondHosting) { |
| 103 | t.Fatalf("joining: %d %q", w.Code, w.Body.String()) |
| 104 | } |
| 105 | // A POST, or another path, is ordinary :80 traffic. |
| 106 | if w := fallbackGet(e, http.MethodPost, lbhealth.Path, "10.0.1.7"); w.Code == http.StatusOK || w.Code == http.StatusServiceUnavailable { |
| 107 | t.Fatalf("POST answered by the health check: %d", w.Code) |
| 108 | } |
| 109 | e.lbHealth = nil |
| 110 | if w := fallbackGet(e, http.MethodGet, lbhealth.Path, "10.0.1.7"); w.Code == http.StatusOK || strings.Contains(w.Body.String(), "unavailable:") { |
| 111 | t.Fatalf("region HA off answered the health check: %d %q", w.Code, w.Body.String()) |
| 112 | } |
| 113 | } |
| 114 | |
| 115 | // blockingStore makes every List wait until release is closed, and fail |
| 116 | // while down is set: a refresh stuck inside one attempt, and a bucket |
| 117 | // outage. |
| 118 | type blockingStore struct { |
| 119 | release.Store |
| 120 | block chan struct{} |
| 121 | down bool |
| 122 | entered chan struct{} |
| 123 | } |
| 124 | |
| 125 | func (s *blockingStore) List(ctx context.Context, prefix string) ([]string, error) { |
| 126 | if s.block != nil { |
| 127 | s.entered <- struct{}{} |
| 128 | <-s.block |
| 129 | } |
| 130 | if s.down { |
| 131 | return nil, errors.New("bucket unreachable") |
| 132 | } |
| 133 | return s.Store.List(ctx, prefix) |
| 134 | } |
| 135 | |
| 136 | func (s *blockingStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| 137 | if s.down { |
| 138 | return nil, "", errors.New("bucket unreachable") |
| 139 | } |
| 140 | return s.Store.Get(ctx, key) |
| 141 | } |
| 142 | |
| 143 | // The home map latches its first loaded map and key, stamps every attempt |
| 144 | // it finishes, failed or not, and stamps nothing while one is stuck. |
| 145 | func TestHomeMapFeedsTheHealthState(t *testing.T) { |
| 146 | inner := meshTestStore(t) |
| 147 | if err := mintMeshKey(context.Background(), inner); err != nil { |
| 148 | t.Fatal(err) |
| 149 | } |
| 150 | putJSON(t, inner, "control/edges/betty.json", `{"v":1,"box":"betty","eip":"2.2.2.2","hosts":{"b.test":{"role":"home"}}}`) |
| 151 | s := &blockingStore{Store: inner} |
| 152 | h := &lbhealth.State{} |
| 153 | m := newHomeMap(s, "annie") |
| 154 | m.health = h |
| 155 | refreshHomeMap(t, m) |
| 156 | now := time.Now() |
| 157 | failing := strings.Join(h.Failing(now), ",") |
| 158 | if strings.Contains(failing, lbhealth.CondForwarding) || strings.Contains(failing, lbhealth.CondRefresh) { |
| 159 | t.Fatalf("after a good refresh: %s", failing) |
| 160 | } |
| 161 | |
| 162 | // A bucket outage: the attempt fails but finishes, and the latches hold. |
| 163 | s.down = true |
| 164 | before := time.Now() |
| 165 | if err := m.Refresh(context.Background()); err == nil { |
| 166 | t.Fatal("refresh against a down bucket succeeded") |
| 167 | } |
| 168 | if got := strings.Join(h.Failing(before.Add(lbhealth.RefreshWindow)), ","); strings.Contains(got, lbhealth.CondForwarding) || strings.Contains(got, lbhealth.CondRefresh) { |
| 169 | t.Fatalf("a failed attempt dropped forwarding or went unstamped: %s", got) |
| 170 | } |
| 171 | |
| 172 | // A stuck attempt stamps nothing, at entry or otherwise: the last stamp |
| 173 | // is the failed attempt above, taken more than a second before t0, so |
| 174 | // a minute and a bit short of the window after t0 it is already stale. |
| 175 | // A stamp taken at entry would be fresh here. |
| 176 | s.down = false |
| 177 | s.block, s.entered = make(chan struct{}), make(chan struct{}, 1) |
| 178 | time.Sleep(1200 * time.Millisecond) |
| 179 | t0 := time.Now() |
| 180 | done := make(chan struct{}) |
| 181 | go func() { _ = m.Refresh(context.Background()); close(done) }() |
| 182 | <-s.entered |
| 183 | if got := strings.Join(h.Failing(t0.Add(lbhealth.RefreshWindow-time.Second)), ","); !strings.Contains(got, lbhealth.CondRefresh) { |
| 184 | t.Fatalf("a refresh stuck inside its attempt was stamped at entry: %s", got) |
| 185 | } |
| 186 | stuckAt := time.Now() |
| 187 | if got := strings.Join(h.Failing(stuckAt.Add(lbhealth.RefreshWindow+time.Second)), ","); !strings.Contains(got, lbhealth.CondRefresh) { |
| 188 | t.Fatalf("a refresh stuck for longer than the window still counts: %s", got) |
| 189 | } |
| 190 | close(s.block) |
| 191 | <-done |
| 192 | } |
| 193 | |
| 194 | // A key that never loads leaves forwarding unarmed, whatever the map did. |
| 195 | func TestHomeMapWithNoMeshKeyIsNotArmed(t *testing.T) { |
| 196 | s := meshTestStore(t) |
| 197 | putJSON(t, s, "control/edges/betty.json", `{"v":1,"box":"betty","eip":"2.2.2.2","hosts":{}}`) |
| 198 | h := &lbhealth.State{} |
| 199 | m := newHomeMap(s, "annie") |
| 200 | m.health = h |
| 201 | refreshHomeMap(t, m) |
| 202 | if got := strings.Join(h.Failing(time.Now()), ","); !strings.Contains(got, lbhealth.CondForwarding) { |
| 203 | t.Fatalf("no mesh key, yet forwarding armed: %s", got) |
| 204 | } |
| 205 | } |
| 206 | |
| 207 | func TestCertMirrorLatchesItsFirstLoad(t *testing.T) { |
| 208 | s := meshTestStore(t) |
| 209 | h := &lbhealth.State{} |
| 210 | m := newCertMirror(s, autocert.DirCache(t.TempDir())) |
| 211 | m.health = h |
| 212 | if !strings.Contains(strings.Join(h.Failing(time.Now()), ","), lbhealth.CondCerts) { |
| 213 | t.Fatal("certificates counted before any load") |
| 214 | } |
| 215 | if err := m.Refresh(context.Background()); err != nil { |
| 216 | t.Fatal(err) |
| 217 | } |
| 218 | if strings.Contains(strings.Join(h.Failing(time.Now()), ","), lbhealth.CondCerts) { |
| 219 | t.Fatal("a loaded mirror did not latch the certificate condition") |
| 220 | } |
| 221 | } |
| 222 | |
| 223 | // The edge's registry watch beats on every poll it completes. |
| 224 | func TestRegistryWatchBeats(t *testing.T) { |
| 225 | e, err := New(openReg(t), nil) |
| 226 | if err != nil { |
| 227 | t.Fatal(err) |
| 228 | } |
| 229 | t.Cleanup(e.probers.Stop) |
| 230 | h := &lbhealth.State{} |
| 231 | e.lbHealth = h |
| 232 | stop := make(chan struct{}) |
| 233 | t.Cleanup(func() { close(stop) }) |
| 234 | go e.Watch(stop) |
| 235 | deadline := time.Now().Add(5 * time.Second) |
| 236 | for strings.Contains(strings.Join(h.Failing(time.Now()), ","), lbhealth.CondEdge) { |
| 237 | if time.Now().After(deadline) { |
| 238 | t.Fatal("the registry watch never beat") |
| 239 | } |
| 240 | time.Sleep(50 * time.Millisecond) |
| 241 | } |
| 242 | } |
| 243 | |
| 244 | // A certificate mirror whose first list fails has loaded nothing: the |
| 245 | // certificate condition stays unlatched, and a later good refresh latches it. |
| 246 | func TestCertMirrorFailedRefreshLeavesCertsUnlatched(t *testing.T) { |
| 247 | inner := meshTestStore(t) |
| 248 | s := &blockingStore{Store: inner, down: true} |
| 249 | h := &lbhealth.State{} |
| 250 | m := newCertMirror(s, autocert.DirCache(t.TempDir())) |
| 251 | m.health = h |
| 252 | if err := m.Refresh(context.Background()); err == nil { |
| 253 | t.Fatal("refresh against a down bucket succeeded") |
| 254 | } |
| 255 | if !strings.Contains(strings.Join(h.Failing(time.Now()), ","), lbhealth.CondCerts) { |
| 256 | t.Fatal("a failed refresh latched the certificate condition") |
| 257 | } |
| 258 | s.down = false |
| 259 | if err := m.Refresh(context.Background()); err != nil { |
| 260 | t.Fatal(err) |
| 261 | } |
| 262 | if strings.Contains(strings.Join(h.Failing(time.Now()), ","), lbhealth.CondCerts) { |
| 263 | t.Fatal("the next good refresh did not latch") |
| 264 | } |
| 265 | } |
| 266 | |
| 267 | // failingFingerprint is a registry whose file stops being a database after |
| 268 | // it opened (a directory stands where it was), so Fingerprint errors on every call. |
| 269 | func failingFingerprint(t *testing.T) *registry.Registry { |
| 270 | t.Helper() |
| 271 | path := filepath.Join(t.TempDir(), "registry.db") |
| 272 | reg, err := registry.Open(path) |
| 273 | if err != nil { |
| 274 | t.Fatal(err) |
| 275 | } |
| 276 | for _, f := range []string{path, path + "-wal", path + "-shm"} { |
| 277 | _ = os.Remove(f) |
| 278 | } |
| 279 | if err := os.Mkdir(path, 0o700); err != nil { |
| 280 | t.Fatal(err) |
| 281 | } |
| 282 | if _, err := reg.Fingerprint(); err == nil { |
| 283 | t.Fatal("setup: Fingerprint still works") |
| 284 | } |
| 285 | return reg |
| 286 | } |
| 287 | |
| 288 | // A registry poll that fails does not beat. |
| 289 | func TestRegistryWatchDoesNotBeatOnFingerprintError(t *testing.T) { |
| 290 | e, err := New(openReg(t), nil) |
| 291 | if err != nil { |
| 292 | t.Fatal(err) |
| 293 | } |
| 294 | t.Cleanup(e.probers.Stop) |
| 295 | e.reg = failingFingerprint(t) |
| 296 | h := &lbhealth.State{} |
| 297 | e.lbHealth = h |
| 298 | stop := make(chan struct{}) |
| 299 | t.Cleanup(func() { close(stop) }) |
| 300 | go e.Watch(stop) |
| 301 | time.Sleep(5 * time.Second) |
| 302 | if !strings.Contains(strings.Join(h.Failing(time.Now()), ","), lbhealth.CondEdge) { |
| 303 | t.Fatal("a failing Fingerprint beat") |
| 304 | } |
| 305 | } |
| 306 | |
| 307 | // wedgedProbe runs an edge over a state that is healthy at setup (every beat |
| 308 | // taken then), so only the edge's beat can still move, and reports what is |
| 309 | // failing a little before the window ends. |
| 310 | func wedgedProbe(t *testing.T, wedge bool) (failing string) { |
| 311 | t.Helper() |
| 312 | old := servingProbeTimeout |
| 313 | servingProbeTimeout = 100 * time.Millisecond |
| 314 | t.Cleanup(func() { servingProbeTimeout = old }) |
| 315 | e, err := New(openReg(t), nil) |
| 316 | if err != nil { |
| 317 | t.Fatal(err) |
| 318 | } |
| 319 | t.Cleanup(e.probers.Stop) |
| 320 | e.EnableCache(NewCache(10, 1<<20)) |
| 321 | h := healthyState() |
| 322 | h.SetActivator(time.Now) // the activator is not what this test wedges |
| 323 | e.lbHealth = h |
| 324 | if wedge { |
| 325 | // A request path deadlocked inside the cache: every cacheable GET |
| 326 | // takes this lock. Released at cleanup so the probe goroutines end. |
| 327 | e.cache.mu.Lock() |
| 328 | t.Cleanup(e.cache.mu.Unlock) |
| 329 | } |
| 330 | stop := make(chan struct{}) |
| 331 | t.Cleanup(func() { close(stop) }) |
| 332 | go e.Watch(stop) |
| 333 | time.Sleep(5 * time.Second) |
| 334 | // Set the clock 57s ahead: a beat from the last tick or two is still |
| 335 | // inside the 60s window, one from before the wedge (setup, 5s+ ago) is not. |
| 336 | now := time.Now().Add(lbhealth.HeartbeatWindow - 3*time.Second) |
| 337 | got := h.Failing(now) |
| 338 | w := httptest.NewRecorder() |
| 339 | lbhealth.Serve(w, httptest.NewRequest(http.MethodGet, lbhealth.Path, nil), h, now) |
| 340 | if wedge != (w.Code == http.StatusServiceUnavailable) { |
| 341 | t.Fatalf("wedge=%v: the health answer was %d %q", wedge, w.Code, w.Body.String()) |
| 342 | } |
| 343 | return strings.Join(got, ",") |
| 344 | } |
| 345 | |
| 346 | // A deadlocked request path keeps the registry poll running, but the edge |
| 347 | // stops beating, and the health answer turns 503 naming the edge once the |
| 348 | // last beat ages out. |
| 349 | func TestEdgeBeatStopsWhenTheServingPathIsWedged(t *testing.T) { |
| 350 | if got := wedgedProbe(t, true); !strings.Contains(got, lbhealth.CondEdge) { |
| 351 | t.Fatalf("a wedged serving path kept beating: %q", got) |
| 352 | } |
| 353 | } |
| 354 | |
| 355 | func TestEdgeBeatContinuesWhenTheServingPathIsFree(t *testing.T) { |
| 356 | if got := wedgedProbe(t, false); strings.Contains(got, lbhealth.CondEdge) { |
| 357 | t.Fatalf("a free serving path stopped beating: %q", got) |
| 358 | } |
| 359 | } |
| 360 | |
amadan