| 1 | package edge |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "io" |
| 7 | "net" |
| 8 | "net/http" |
| 9 | "net/http/httptest" |
| 10 | "strings" |
| 11 | "sync" |
| 12 | "syscall" |
| 13 | "testing" |
| 14 | "time" |
| 15 | |
| 16 | "github.com/carlosframework/platform/internal/apperr" |
| 17 | "github.com/carlosframework/platform/internal/registry" |
| 18 | "github.com/carlosframework/platform/internal/release" |
| 19 | ) |
| 20 | |
| 21 | type memErrSink struct { |
| 22 | mu sync.Mutex |
| 23 | recs []apperr.Record |
| 24 | apps []string |
| 25 | } |
| 26 | |
| 27 | func (m *memErrSink) Add(acct, app string, r apperr.Record) { |
| 28 | m.mu.Lock() |
| 29 | defer m.mu.Unlock() |
| 30 | m.recs = append(m.recs, r) |
| 31 | m.apps = append(m.apps, acct+"/"+app) |
| 32 | } |
| 33 | |
| 34 | // snap copies the sink's records under the lock — every assertion reads |
| 35 | // through this rather than the fields directly, so a test can never |
| 36 | // observe a torn append from a concurrent Add. |
| 37 | func (m *memErrSink) snap() []apperr.Record { |
| 38 | m.mu.Lock() |
| 39 | defer m.mu.Unlock() |
| 40 | out := make([]apperr.Record, len(m.recs)) |
| 41 | copy(out, m.recs) |
| 42 | return out |
| 43 | } |
| 44 | |
| 45 | func TestEdgeEmitsErrorOnUpstream5xx(t *testing.T) { |
| 46 | backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| 47 | switch r.URL.Path { |
| 48 | case "/boom": |
| 49 | http.Error(w, "kaboom", http.StatusInternalServerError) |
| 50 | case "/missing": |
| 51 | http.NotFound(w, r) |
| 52 | default: |
| 53 | w.Header().Set("Content-Type", "text/html") |
| 54 | w.Write([]byte("<p>ok</p>")) |
| 55 | } |
| 56 | })) |
| 57 | t.Cleanup(backend.Close) |
| 58 | reg := openReg(t) |
| 59 | rt := registry.Route{Host: "app.example.com", Kind: "service", Addr: backend.Listener.Addr().String(), Status: "ready", App: "web", Version: "v7"} |
| 60 | if err := reg.Upsert(rt); err != nil { |
| 61 | t.Fatal(err) |
| 62 | } |
| 63 | e, err := New(reg, nil) |
| 64 | if err != nil { |
| 65 | t.Fatal(err) |
| 66 | } |
| 67 | sink := &memErrSink{} |
| 68 | e.EnableErrors(sink) |
| 69 | srv := httptest.NewServer(e) |
| 70 | t.Cleanup(srv.Close) |
| 71 | |
| 72 | do := func(method, path string) int { |
| 73 | req, _ := http.NewRequest(method, srv.URL+path, nil) |
| 74 | req.Host = "app.example.com" |
| 75 | resp, err := http.DefaultClient.Do(req) |
| 76 | if err != nil { |
| 77 | t.Fatal(err) |
| 78 | } |
| 79 | resp.Body.Close() |
| 80 | return resp.StatusCode |
| 81 | } |
| 82 | if do("POST", "/boom?token=abc") != 500 || do("GET", "/missing") != 404 || do("GET", "/") != 200 { |
| 83 | t.Fatal("upstream statuses must pass through unchanged") |
| 84 | } |
| 85 | recs := sink.snap() |
| 86 | if len(recs) != 1 { |
| 87 | t.Fatalf("errors emitted = %d, want 1 (the 500 only)", len(recs)) |
| 88 | } |
| 89 | r := recs[0] |
| 90 | if r.Source != apperr.SourceEdge || r.Kind != apperr.KindHTTP || r.Status != 500 || r.Host != "app.example.com" || r.Release != "v7" { |
| 91 | t.Fatalf("record = %+v", r) |
| 92 | } |
| 93 | if r.Name != "500 POST /boom" || r.Path != "/boom" || strings.Contains(r.Name, "token") { |
| 94 | t.Fatalf("name/path = %q %q", r.Name, r.Path) |
| 95 | } |
| 96 | // The route carries no Account, so RouteAccount falls back to |
| 97 | // OpsAccount: attribution must land on the ops account, not "". |
| 98 | if sink.apps[0] != release.OpsAccount+"/web" { |
| 99 | t.Fatalf("attribution = %q, want %q", sink.apps[0], release.OpsAccount+"/web") |
| 100 | } |
| 101 | } |
| 102 | |
| 103 | // answering503 is an Activator that answers every request itself with the |
| 104 | // wake-path 503 the real one writes, and marks it as the activator's own — |
| 105 | // the same MarkAnswered call internal/activator's serve makes before each of |
| 106 | // its four wake-path http.Error(…, 503) writes. |
| 107 | type answering503 struct{} |
| 108 | |
| 109 | func (answering503) ServeHTTP(w http.ResponseWriter, r *http.Request, rt registry.Route, proxy http.Handler) bool { |
| 110 | MarkAnswered(r) |
| 111 | http.Error(w, "instance failed to wake", http.StatusServiceUnavailable) |
| 112 | return true |
| 113 | } |
| 114 | func (answering503) RetireExcept(map[string]bool) {} |
| 115 | |
| 116 | func TestEdgeSkipsActivatorAnswered503(t *testing.T) { |
| 117 | reg := openReg(t) |
| 118 | rt := registry.Route{Host: "sleepy.example.com", Kind: "instance", Socket: t.TempDir() + "/x.sock", Status: "ready", Hibernate: true, App: "web"} |
| 119 | if err := reg.Upsert(rt); err != nil { |
| 120 | t.Fatal(err) |
| 121 | } |
| 122 | e, err := New(reg, answering503{}) |
| 123 | if err != nil { |
| 124 | t.Fatal(err) |
| 125 | } |
| 126 | sink := &memErrSink{} |
| 127 | e.EnableErrors(sink) |
| 128 | rec := httptest.NewRecorder() |
| 129 | req := httptest.NewRequest("GET", "http://sleepy.example.com/", nil) |
| 130 | e.ServeHTTP(rec, req) |
| 131 | if rec.Code != 503 { |
| 132 | t.Fatalf("status = %d", rec.Code) |
| 133 | } |
| 134 | if recs := sink.snap(); len(recs) != 0 { |
| 135 | t.Fatalf("an activator-answered 503 is already a platform log error; got %d edge errors", len(recs)) |
| 136 | } |
| 137 | } |
| 138 | |
| 139 | // plain503 is answering503's negative twin: it writes the same wake-path |
| 140 | // 503 but never calls MarkAnswered, so the edge must not mistake it for the |
| 141 | // activator's own already-logged error. |
| 142 | type plain503 struct{} |
| 143 | |
| 144 | func (plain503) ServeHTTP(w http.ResponseWriter, r *http.Request, rt registry.Route, proxy http.Handler) bool { |
| 145 | http.Error(w, "instance failed to wake", http.StatusServiceUnavailable) |
| 146 | return true |
| 147 | } |
| 148 | func (plain503) RetireExcept(map[string]bool) {} |
| 149 | |
| 150 | // TestEdgeEmitsUnmarked503OnHibernatingRoute is the positive control for |
| 151 | // TestEdgeSkipsActivatorAnswered503: with the MarkAnswered call gone, the |
| 152 | // same hibernating route's 503 must be filed as exactly one edge error. |
| 153 | func TestEdgeEmitsUnmarked503OnHibernatingRoute(t *testing.T) { |
| 154 | reg := openReg(t) |
| 155 | rt := registry.Route{Host: "sleepy.example.com", Kind: "instance", Socket: t.TempDir() + "/x.sock", Status: "ready", Hibernate: true, App: "web"} |
| 156 | if err := reg.Upsert(rt); err != nil { |
| 157 | t.Fatal(err) |
| 158 | } |
| 159 | e, err := New(reg, plain503{}) |
| 160 | if err != nil { |
| 161 | t.Fatal(err) |
| 162 | } |
| 163 | sink := &memErrSink{} |
| 164 | e.EnableErrors(sink) |
| 165 | rec := httptest.NewRecorder() |
| 166 | req := httptest.NewRequest("GET", "http://sleepy.example.com/", nil) |
| 167 | e.ServeHTTP(rec, req) |
| 168 | if rec.Code != 503 { |
| 169 | t.Fatalf("status = %d", rec.Code) |
| 170 | } |
| 171 | recs := sink.snap() |
| 172 | if len(recs) != 1 { |
| 173 | t.Fatalf("errors emitted = %d, want 1 (the unmarked 503)", len(recs)) |
| 174 | } |
| 175 | if recs[0].Status != 503 || recs[0].Source != apperr.SourceEdge { |
| 176 | t.Fatalf("record = %+v", recs[0]) |
| 177 | } |
| 178 | } |
| 179 | |
| 180 | // TestEdgeFilesUnavailableByCause: a 502/503/504 goes under Availability, |
| 181 | // named by status and cause rather than path, so one dead instance probed |
| 182 | // on two paths is one fingerprint — and a 503 the app wrote itself says so. |
| 183 | func TestEdgeFilesUnavailableByCause(t *testing.T) { |
| 184 | backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| 185 | http.Error(w, "busy", http.StatusServiceUnavailable) |
| 186 | })) |
| 187 | t.Cleanup(backend.Close) |
| 188 | dead := httptest.NewServer(http.NotFoundHandler()) |
| 189 | deadAddr := dead.Listener.Addr().String() |
| 190 | dead.Close() // nothing listens here now: every dial is refused |
| 191 | |
| 192 | reg := openReg(t) |
| 193 | for _, rt := range []registry.Route{ |
| 194 | {Host: "busy.example.com", Kind: "service", Addr: backend.Listener.Addr().String(), Status: "ready", App: "web", Version: "v7"}, |
| 195 | {Host: "dead.example.com", Kind: "service", Addr: deadAddr, Status: "ready", App: "web", Version: "v7"}, |
| 196 | } { |
| 197 | if err := reg.Upsert(rt); err != nil { |
| 198 | t.Fatal(err) |
| 199 | } |
| 200 | } |
| 201 | e, err := New(reg, nil) |
| 202 | if err != nil { |
| 203 | t.Fatal(err) |
| 204 | } |
| 205 | sink := &memErrSink{} |
| 206 | e.EnableErrors(sink) |
| 207 | srv := httptest.NewServer(e) |
| 208 | t.Cleanup(srv.Close) |
| 209 | do := func(host, path string) int { |
| 210 | req, _ := http.NewRequest("GET", srv.URL+path, nil) |
| 211 | req.Host = host |
| 212 | resp, err := http.DefaultClient.Do(req) |
| 213 | if err != nil { |
| 214 | t.Fatal(err) |
| 215 | } |
| 216 | resp.Body.Close() |
| 217 | return resp.StatusCode |
| 218 | } |
| 219 | if got := do("busy.example.com", "/sandbox"); got != 503 { |
| 220 | t.Fatalf("busy status = %d", got) |
| 221 | } |
| 222 | if do("dead.example.com", "/.env") != 502 || do("dead.example.com", "/phpinfo.php") != 502 { |
| 223 | t.Fatal("a refused dial must answer 502") |
| 224 | } |
| 225 | recs := sink.snap() |
| 226 | if len(recs) != 3 { |
| 227 | t.Fatalf("records = %d, want 3", len(recs)) |
| 228 | } |
| 229 | busy := recs[0] |
| 230 | if busy.Kind != apperr.KindUnavailable || busy.Cause != apperr.CauseUpstream || busy.Name != "503 · the app answered it itself" || busy.Path != "/sandbox" { |
| 231 | t.Fatalf("busy record = %+v", busy) |
| 232 | } |
| 233 | for _, r := range recs[1:] { |
| 234 | if r.Kind != apperr.KindUnavailable || r.Cause != apperr.CauseRefused || r.Status != 502 || strings.Contains(r.Name, "/") { |
| 235 | t.Fatalf("dead record = %+v", r) |
| 236 | } |
| 237 | } |
| 238 | if recs[1].Name != recs[2].Name || recs[1].Path == recs[2].Path { |
| 239 | t.Fatalf("two paths of one refused instance must share a name and keep their own paths: %+v", recs[1:]) |
| 240 | } |
| 241 | if !apperr.IsAvailability(busy) { |
| 242 | t.Fatal("IsAvailability(busy) = false") |
| 243 | } |
| 244 | } |
| 245 | |
| 246 | // TestProxyCauseClientGone: a proxy failure after the client left is the |
| 247 | // client's doing, whatever the error says, and is never filed. |
| 248 | func TestProxyCauseClientGone(t *testing.T) { |
| 249 | ctx, cancel := context.WithCancel(context.Background()) |
| 250 | cancel() |
| 251 | r := httptest.NewRequest("GET", "http://x/", nil).WithContext(ctx) |
| 252 | if got := proxyCause(r, syscall.ECONNRESET); got != causeClientGone { |
| 253 | t.Fatalf("proxyCause = %q, want %q", got, causeClientGone) |
| 254 | } |
| 255 | r, out := withOutcome(httptest.NewRequest("GET", "http://x/", nil)) |
| 256 | for _, c := range []struct { |
| 257 | err error |
| 258 | want string |
| 259 | }{ |
| 260 | {&net.OpError{Op: "dial", Err: syscall.ECONNREFUSED}, apperr.CauseRefused}, |
| 261 | {io.ErrUnexpectedEOF, apperr.CauseReset}, |
| 262 | {context.DeadlineExceeded, apperr.CauseTimeout}, |
| 263 | {errors.New("malformed"), apperr.CauseProxy}, |
| 264 | } { |
| 265 | if got := proxyCause(r, c.err); got != c.want { |
| 266 | t.Errorf("proxyCause(%v) = %q, want %q", c.err, got, c.want) |
| 267 | } |
| 268 | } |
| 269 | // The same leaving after a long wait is the instance's doing: a dead |
| 270 | // socket redialled for 13s, or a wedged instance nothing times out. |
| 271 | gone, o := withOutcome(httptest.NewRequest("GET", "http://x/", nil).WithContext(ctx)) |
| 272 | o.start = time.Now().Add(-clientGoneGrace - time.Second) |
| 273 | if got := proxyCause(gone, syscall.ECONNREFUSED); got != apperr.CauseAbandoned { |
| 274 | t.Fatalf("long wait then gone: proxyCause = %q, want %q", got, apperr.CauseAbandoned) |
| 275 | } |
| 276 | if got := tunnelCause(gone, syscall.ECONNREFUSED); got != apperr.CauseAbandoned { |
| 277 | t.Fatalf("long wait then gone via tunnel: %q", got) |
| 278 | } |
| 279 | noteCause(r, apperr.CauseReset) |
| 280 | if out.cause != apperr.CauseReset { |
| 281 | t.Fatalf("noteCause did not reach the outcome: %+v", out) |
| 282 | } |
| 283 | sink := &memErrSink{} |
| 284 | e := &Edge{} |
| 285 | e.EnableErrors(sink) |
| 286 | e.emitEdgeError(registry.Route{Host: "h", App: "web"}, false, r, 502, causeClientGone) |
| 287 | if n := len(sink.snap()); n != 0 { |
| 288 | t.Fatalf("a client-gone 502 was filed (%d records)", n) |
| 289 | } |
| 290 | } |
| 291 | |
| 292 | // TestEdgeFilesOnlyItsOwnFailuresThroughATunnel: when this box proxies over |
| 293 | // a reverse channel, the upstream is another box's edge, which files what |
| 294 | // it saw serving the app. This box files the hop failing, never a status |
| 295 | // that merely passed through it. |
| 296 | func TestEdgeFilesOnlyItsOwnFailuresThroughATunnel(t *testing.T) { |
| 297 | sink := &memErrSink{} |
| 298 | e := &Edge{} |
| 299 | e.EnableErrors(sink) |
| 300 | r, _ := withOutcome(httptest.NewRequest("GET", "http://x/admin", nil)) |
| 301 | rt := registry.Route{Host: "t.example.com", App: "web", Tunnel: "box1"} |
| 302 | e.emitEdgeError(rt, true, r, 503, "") // the serving box's own 503, passed through |
| 303 | e.emitEdgeError(rt, true, r, 500, "") // the app's 500, filed on the serving box |
| 304 | if n := len(sink.snap()); n != 0 { |
| 305 | t.Fatalf("pass-through statuses were filed on the front box: %+v", sink.snap()) |
| 306 | } |
| 307 | e.emitEdgeError(rt, true, r, 502, apperr.CauseTunnel) |
| 308 | e.emitEdgeError(rt, true, r, 503, apperr.CauseFleetDark) |
| 309 | if recs := sink.snap(); len(recs) != 2 || recs[0].Cause != apperr.CauseTunnel || recs[1].Cause != apperr.CauseFleetDark { |
| 310 | t.Fatalf("the hop's own failures must be filed: %+v", recs) |
| 311 | } |
| 312 | } |
| 313 | |
| 314 | // TestEdgeServingFleetBoxFilesItsOwn is the other end of the same route: |
| 315 | // the fleet box that SERVES a fleet-placed app carries the placement on its |
| 316 | // row too, but has no hub and serves from the app's own address. Everything |
| 317 | // the app answers there must be filed — it is the only box that can. |
| 318 | func TestEdgeServingFleetBoxFilesItsOwn(t *testing.T) { |
| 319 | backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| 320 | if r.URL.Path == "/busy" { |
| 321 | http.Error(w, "busy", http.StatusServiceUnavailable) |
| 322 | return |
| 323 | } |
| 324 | http.Error(w, "boom", http.StatusInternalServerError) |
| 325 | })) |
| 326 | t.Cleanup(backend.Close) |
| 327 | reg := openReg(t) |
| 328 | rt := registry.Route{Host: "f.example.com", Kind: "service", Addr: backend.Listener.Addr().String(), Status: "ready", App: "web", |
| 329 | Region: registry.PlacementTarget("eleven"), Tunnel: "box1"} |
| 330 | if err := reg.Upsert(rt); err != nil { |
| 331 | t.Fatal(err) |
| 332 | } |
| 333 | e, err := New(reg, nil) // no hub: this box is the far end, not the front |
| 334 | if err != nil { |
| 335 | t.Fatal(err) |
| 336 | } |
| 337 | sink := &memErrSink{} |
| 338 | e.EnableErrors(sink) |
| 339 | for _, path := range []string{"/boom", "/busy"} { |
| 340 | rec := httptest.NewRecorder() |
| 341 | e.ServeHTTP(rec, httptest.NewRequest("GET", "http://f.example.com"+path, nil)) |
| 342 | } |
| 343 | recs := sink.snap() |
| 344 | if len(recs) != 2 || recs[0].Status != 500 || recs[1].Cause != apperr.CauseUpstream { |
| 345 | t.Fatalf("the serving box must file its app's 500 and 503: %+v", recs) |
| 346 | } |
| 347 | } |
| 348 | |
| 349 | // TestEdgeFilesAbandonedWaitEndToEnd drives a real request at an instance |
| 350 | // that never answers, through ServeHTTP and the proxy, with a client that |
| 351 | // gives up after the grace: the wait is measured from where the edge |
| 352 | // attaches the outcome, and the outage is filed rather than hidden. |
| 353 | func TestEdgeFilesAbandonedWaitEndToEnd(t *testing.T) { |
| 354 | old := clientGoneGrace |
| 355 | clientGoneGrace = 50 * time.Millisecond |
| 356 | t.Cleanup(func() { clientGoneGrace = old }) |
| 357 | release := make(chan struct{}) |
| 358 | backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| 359 | select { |
| 360 | case <-release: |
| 361 | case <-r.Context().Done(): |
| 362 | } |
| 363 | })) |
| 364 | t.Cleanup(func() { close(release); backend.Close() }) |
| 365 | reg := openReg(t) |
| 366 | if err := reg.Upsert(registry.Route{Host: "wedged.example.com", Kind: "service", Addr: backend.Listener.Addr().String(), Status: "ready", App: "web"}); err != nil { |
| 367 | t.Fatal(err) |
| 368 | } |
| 369 | e, err := New(reg, nil) |
| 370 | if err != nil { |
| 371 | t.Fatal(err) |
| 372 | } |
| 373 | sink := &memErrSink{} |
| 374 | e.EnableErrors(sink) |
| 375 | ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond) |
| 376 | defer cancel() |
| 377 | req := httptest.NewRequest("GET", "http://wedged.example.com/slow", nil).WithContext(ctx) |
| 378 | e.ServeHTTP(httptest.NewRecorder(), req) |
| 379 | recs := sink.snap() |
| 380 | if len(recs) != 1 || recs[0].Cause != apperr.CauseAbandoned || recs[0].Status != 502 { |
| 381 | t.Fatalf("records = %+v, want one abandoned 502", recs) |
| 382 | } |
| 383 | } |
| 384 | |
| 385 | // TestEdgeFilesHopFailuresThroughServeHTTP pins the other half of the |
| 386 | // tunnel rule end to end: the front box files nothing a proxy did not note |
| 387 | // a cause for, so each hop failure it produces itself MUST note one, or it |
| 388 | // is filed nowhere. A hub with no connected channel answers a -tunnel route |
| 389 | // with 502 (tunnel) and a fleet route with 503 (fleet_dark). |
| 390 | func TestEdgeFilesHopFailuresThroughServeHTTP(t *testing.T) { |
| 391 | reg := openReg(t) |
| 392 | if err := reg.UpsertFleet(registry.Fleet{Name: "acme", Account: "amadan", Bucket: "acme-data"}); err != nil { |
| 393 | t.Fatal(err) |
| 394 | } |
| 395 | if err := reg.AddFleetBox(registry.FleetBox{Fleet: "acme", Label: "pi-1", TokenHash: sha256Hex("tok1"), Status: "active"}); err != nil { |
| 396 | t.Fatal(err) |
| 397 | } |
| 398 | for _, rt := range []registry.Route{ |
| 399 | {Host: "fleet.example", Kind: "instance", Status: "ready", App: "web", Region: registry.PlacementTarget("acme"), FleetBox: "pi-1"}, |
| 400 | {Host: "tun.example", Kind: "service", Status: "ready", App: "web", Addr: "127.0.0.1:1", Tunnel: "pi-box"}, |
| 401 | } { |
| 402 | if err := reg.Upsert(rt); err != nil { |
| 403 | t.Fatal(err) |
| 404 | } |
| 405 | } |
| 406 | if err := reg.SetTunnel("tun.example", "pi-box"); err != nil { |
| 407 | t.Fatal(err) |
| 408 | } |
| 409 | e, err := New(reg, nil) |
| 410 | if err != nil { |
| 411 | t.Fatal(err) |
| 412 | } |
| 413 | e.EnableTunnels(NewTunnelHub("tunnel.test", map[string]string{"pi-box": "sekrit"})) |
| 414 | if err := e.Reload(); err != nil { |
| 415 | t.Fatal(err) |
| 416 | } |
| 417 | sink := &memErrSink{} |
| 418 | e.EnableErrors(sink) |
| 419 | for host, want := range map[string]int{"fleet.example": 503, "tun.example": 502} { |
| 420 | rec := httptest.NewRecorder() |
| 421 | e.ServeHTTP(rec, httptest.NewRequest("GET", "http://"+host+"/", nil)) |
| 422 | if rec.Code != want { |
| 423 | t.Fatalf("%s = %d, want %d", host, rec.Code, want) |
| 424 | } |
| 425 | } |
| 426 | causes := map[string]string{} |
| 427 | for _, r := range sink.snap() { |
| 428 | causes[r.Host] = r.Cause |
| 429 | } |
| 430 | if len(causes) != 2 || causes["fleet.example"] != apperr.CauseFleetDark || causes["tun.example"] != apperr.CauseTunnel { |
| 431 | t.Fatalf("hop failures filed = %+v, want fleet_dark and tunnel, one each", sink.snap()) |
| 432 | } |
| 433 | } |
| 434 | |
amadan