| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "net" |
| 6 | "net/http" |
| 7 | "os" |
| 8 | "path/filepath" |
| 9 | "strings" |
| 10 | "sync" |
| 11 | "testing" |
| 12 | "time" |
| 13 | |
| 14 | "github.com/carlosframework/platform/internal/activator" |
| 15 | "github.com/carlosframework/platform/internal/hostwrite" |
| 16 | "github.com/carlosframework/platform/internal/lease" |
| 17 | "github.com/carlosframework/platform/internal/moves" |
| 18 | "github.com/carlosframework/platform/internal/registry" |
| 19 | ) |
| 20 | |
| 21 | // instanceForTest serves h on a unix socket, the way an exec instance |
| 22 | // listens, and returns the route that names it. |
| 23 | func instanceForTest(t *testing.T, h http.HandlerFunc) registry.Route { |
| 24 | t.Helper() |
| 25 | dir, err := os.MkdirTemp("", "s1") |
| 26 | if err != nil { |
| 27 | t.Fatal(err) |
| 28 | } |
| 29 | t.Cleanup(func() { os.RemoveAll(dir) }) |
| 30 | sock := filepath.Join(dir, "i.sock") |
| 31 | ln, err := net.Listen("unix", sock) |
| 32 | if err != nil { |
| 33 | t.Fatal(err) |
| 34 | } |
| 35 | srv := &http.Server{Handler: h} |
| 36 | go srv.Serve(ln) |
| 37 | t.Cleanup(func() { srv.Close() }) |
| 38 | return registry.Route{Host: p1Host, Kind: "instance", Hibernate: true, Status: "ready", Socket: sock} |
| 39 | } |
| 40 | |
| 41 | // wakeThenProxy is an activator that wakes and hands the request to the |
| 42 | // probe's proxy, as Service.ServeScheduled does for an awake instance. |
| 43 | func wakeThenProxy(w http.ResponseWriter, r *http.Request, _ registry.Route, proxy http.Handler) bool { |
| 44 | proxy.ServeHTTP(w, r) |
| 45 | return true |
| 46 | } |
| 47 | |
| 48 | // S1 must prove the instance itself answered, in full, with a 2xx, within |
| 49 | // its deadline (Task 14 review finding 4). |
| 50 | func TestS1ProbeNeedsTheInstanceItself(t *testing.T) { |
| 51 | t.Parallel() |
| 52 | ctx := context.Background() |
| 53 | answer := func(code int) registry.Route { |
| 54 | return instanceForTest(t, func(w http.ResponseWriter, r *http.Request) { |
| 55 | if r.Host != p1Host || r.URL.Path != "/ready" { |
| 56 | w.WriteHeader(http.StatusTeapot) |
| 57 | return |
| 58 | } |
| 59 | w.WriteHeader(code) |
| 60 | w.Write([]byte("ok")) |
| 61 | }) |
| 62 | } |
| 63 | if c := probeInstance(ctx, wakeThenProxy, answer(200), "/ready"); !c.OK { |
| 64 | t.Fatalf("a 200 from the instance: %+v", c) |
| 65 | } |
| 66 | if c := probeInstance(ctx, wakeThenProxy, answer(204), "/ready"); !c.OK { |
| 67 | t.Fatalf("a 204 from the instance: %+v", c) |
| 68 | } |
| 69 | for name, code := range map[string]int{"redirect": 302, "password gate": 401, "server error": 503, "not found": 404} { |
| 70 | if c := probeInstance(ctx, wakeThenProxy, answer(code), "/ready"); c.OK || c.Reason != s1NoAnswer(p1Host) { |
| 71 | t.Errorf("%s (%d) from the instance passed S1: %+v", name, code, c) |
| 72 | } |
| 73 | } |
| 74 | |
| 75 | // Answers that never reached the instance: the edge's unknown-host |
| 76 | // fallback redirect, a gate's 401 and even a 200 written by the |
| 77 | // activator itself (a fenced or refused wake) all fail. |
| 78 | rt := answer(200) |
| 79 | for name, wake := range map[string]func(http.ResponseWriter, *http.Request, registry.Route, http.Handler) bool{ |
| 80 | "fallback redirect": func(w http.ResponseWriter, r *http.Request, _ registry.Route, _ http.Handler) bool { |
| 81 | http.Redirect(w, r, "https://example.com/", http.StatusFound) |
| 82 | return true |
| 83 | }, |
| 84 | "password gate": func(w http.ResponseWriter, _ *http.Request, _ registry.Route, _ http.Handler) bool { |
| 85 | w.WriteHeader(http.StatusUnauthorized) |
| 86 | return true |
| 87 | }, |
| 88 | "a 200 not from the instance": func(w http.ResponseWriter, _ *http.Request, _ registry.Route, _ http.Handler) bool { |
| 89 | w.WriteHeader(http.StatusOK) |
| 90 | return true |
| 91 | }, |
| 92 | } { |
| 93 | if c := probeInstance(ctx, wake, rt, "/ready"); c.OK { |
| 94 | t.Errorf("%s passed S1", name) |
| 95 | } |
| 96 | } |
| 97 | |
| 98 | // 200 headers, part of a body, then nothing until the deadline. |
| 99 | slow := instanceForTest(t, func(w http.ResponseWriter, r *http.Request) { |
| 100 | w.Header().Set("Content-Length", "100") |
| 101 | w.WriteHeader(http.StatusOK) |
| 102 | w.Write([]byte("partial")) |
| 103 | w.(http.Flusher).Flush() |
| 104 | select { |
| 105 | case <-r.Context().Done(): |
| 106 | case <-time.After(5 * time.Second): |
| 107 | } |
| 108 | }) |
| 109 | short, cancel := context.WithTimeout(ctx, 300*time.Millisecond) |
| 110 | defer cancel() |
| 111 | if c := probeInstance(short, wakeThenProxy, slow, "/ready"); c.OK { |
| 112 | t.Fatal("a partial 200 that ran out its deadline passed S1") |
| 113 | } |
| 114 | |
| 115 | // The instance answered in time, but the wake held the probe past its |
| 116 | // deadline: the probe expired, so S1 fails. |
| 117 | late, cancelLate := context.WithTimeout(ctx, 200*time.Millisecond) |
| 118 | defer cancelLate() |
| 119 | overrun := func(w http.ResponseWriter, r *http.Request, rt registry.Route, proxy http.Handler) bool { |
| 120 | proxy.ServeHTTP(w, r) |
| 121 | <-r.Context().Done() |
| 122 | return true |
| 123 | } |
| 124 | if c := probeInstance(late, overrun, answer(200), "/ready"); c.OK { |
| 125 | t.Fatal("a probe that ran out its deadline passed S1") |
| 126 | } |
| 127 | |
| 128 | // A route the activator does not manage is asked directly. |
| 129 | notManaged := func(http.ResponseWriter, *http.Request, registry.Route, http.Handler) bool { return false } |
| 130 | if c := probeInstance(ctx, notManaged, answer(200), "/ready"); !c.OK { |
| 131 | t.Fatalf("an always-on instance: %+v", c) |
| 132 | } |
| 133 | } |
| 134 | |
| 135 | // The declared readiness path reaches the instance as declared: an escaped |
| 136 | // slash stays escaped (Task 14 re-review), so an app that tells |
| 137 | // /ready%2Fcheck from /ready/check is asked the one it declared. |
| 138 | func TestS1ProbeSendsTheDeclaredPathVerbatim(t *testing.T) { |
| 139 | var got []string |
| 140 | rt := instanceForTest(t, func(w http.ResponseWriter, r *http.Request) { |
| 141 | got = append(got, r.RequestURI) |
| 142 | w.WriteHeader(http.StatusOK) |
| 143 | }) |
| 144 | for _, path := range []string{"/ready%2Fcheck", "/ready?probe=a%2Fb", "/plain"} { |
| 145 | got = nil |
| 146 | if c := probeInstance(context.Background(), wakeThenProxy, rt, path); !c.OK { |
| 147 | t.Fatalf("%s: %+v", path, c) |
| 148 | } |
| 149 | if len(got) != 1 || got[0] != path { |
| 150 | t.Fatalf("declared %q, the instance received %q", path, got) |
| 151 | } |
| 152 | } |
| 153 | } |
| 154 | |
| 155 | // armedProbe is the production S1 (armPhase1's ReadyProbe) over a real |
| 156 | // activator and one sleeping instance whose row carries gate as its import |
| 157 | // gate. The socket directory is short, as a unix socket path must be. |
| 158 | func armedProbe(t *testing.T, gate string) (func(context.Context, string, string) moves.Check, *countingWakeBackend) { |
| 159 | t.Helper() |
| 160 | dir, err := os.MkdirTemp("", "s1") |
| 161 | if err != nil { |
| 162 | t.Fatal(err) |
| 163 | } |
| 164 | t.Cleanup(func() { os.RemoveAll(dir) }) |
| 165 | reg, err := registry.Open(filepath.Join(dir, "r.db")) |
| 166 | if err != nil { |
| 167 | t.Fatal(err) |
| 168 | } |
| 169 | t.Cleanup(func() { reg.Close() }) |
| 170 | inst := &fakeInstance{socket: filepath.Join(dir, "i.sock"), handler: http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { |
| 171 | w.WriteHeader(http.StatusOK) |
| 172 | })} |
| 173 | if err := reg.Upsert(registry.Route{Host: p1Host, Kind: "instance", Account: "bab", App: "app", Status: "ready", |
| 174 | Hibernate: true, Socket: inst.socket, DBPath: filepath.Join(dir, "app.db"), ImportGate: gate}); err != nil { |
| 175 | t.Fatal(err) |
| 176 | } |
| 177 | b := &countingWakeBackend{fakeWakeBackend: &fakeWakeBackend{inst: inst}} |
| 178 | svc := activator.New(activator.Config{Bucket: "test-bucket", Idle: time.Hour}, b, reg) |
| 179 | svc.Start() |
| 180 | t.Cleanup(func() { |
| 181 | ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) |
| 182 | defer cancel() |
| 183 | svc.Shutdown(ctx) |
| 184 | }) |
| 185 | var cfg moveCfg |
| 186 | hib := activator.Config{Bucket: "data-ie", Region: "eu-west-1", Prefix: "instances"} |
| 187 | armPhase1(&cfg, &movesWorker{mirrors: map[string]lease.MirrorStore{}}, hib, dir, reg, svc, nil, hostwrite.Config{}, func(string) string { return "" }) |
| 188 | if cfg.ReadyProbe == nil { |
| 189 | t.Fatal("armPhase1 wired no S1 probe") |
| 190 | } |
| 191 | return cfg.ReadyProbe, b |
| 192 | } |
| 193 | |
| 194 | // countingWakeBackend counts every wake the activator begins (its lease |
| 195 | // acquire, the wake's first step), whether or not the spawn then succeeds. |
| 196 | type countingWakeBackend struct { |
| 197 | *fakeWakeBackend |
| 198 | mu sync.Mutex |
| 199 | acquires int |
| 200 | } |
| 201 | |
| 202 | func (b *countingWakeBackend) Acquire(ctx context.Context, rt registry.Route) (activator.Lease, error) { |
| 203 | b.mu.Lock() |
| 204 | b.acquires++ |
| 205 | b.mu.Unlock() |
| 206 | return b.fakeWakeBackend.Acquire(ctx, rt) |
| 207 | } |
| 208 | |
| 209 | func (b *countingWakeBackend) wakes() int { |
| 210 | b.mu.Lock() |
| 211 | defer b.mu.Unlock() |
| 212 | return b.acquires |
| 213 | } |
| 214 | |
| 215 | // The moves probe never wakes a gated import (seeded-instances spec §5.4: |
| 216 | // only the app's own verify requests reach it; §5.8: it moves only once |
| 217 | // committed). ServeScheduled does not read the gate, so the probe must: |
| 218 | // whatever the move's state, S1 fails without a wake. |
| 219 | func TestS1ProbeNeverWakesAGatedImport(t *testing.T) { |
| 220 | t.Parallel() |
| 221 | for _, gate := range []string{strings.Repeat("a", 64), registry.ImportGateClosed} { |
| 222 | probe, b := armedProbe(t, gate) |
| 223 | got := probe(context.Background(), p1Host, "/") |
| 224 | if got.OK || got.Reason != s1NoAnswer(p1Host) { |
| 225 | t.Fatalf("gate %q: S1 = %+v, want %q", gate, got, s1NoAnswer(p1Host)) |
| 226 | } |
| 227 | if n := b.wakes(); n != 0 { |
| 228 | t.Fatalf("gate %q: the probe began %d wakes of a gated import", gate, n) |
| 229 | } |
| 230 | if reqs := b.inst.served(); len(reqs) != 0 { |
| 231 | t.Fatalf("gate %q: the gated instance served %d requests", gate, len(reqs)) |
| 232 | } |
| 233 | } |
| 234 | // The control: the same row with no gate is woken and answers (this |
| 235 | // half listens on a unix socket, so it needs a run outside a sandbox |
| 236 | // that forbids one). |
| 237 | probe, b := armedProbe(t, "") |
| 238 | if got := probe(context.Background(), p1Host, "/"); !got.OK { |
| 239 | t.Fatalf("an ungated instance: S1 = %+v", got) |
| 240 | } |
| 241 | if b.wakes() != 1 || b.inst.startCount() != 1 { |
| 242 | t.Fatalf("an ungated instance: %d wakes, %d starts", b.wakes(), b.inst.startCount()) |
| 243 | } |
| 244 | } |
| 245 | |
amadan