| 1 | package activator |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "net/http" |
| 6 | "net/http/httptest" |
| 7 | "testing" |
| 8 | "time" |
| 9 | |
| 10 | "github.com/carlosframework/platform/internal/registry" |
| 11 | "github.com/carlosframework/platform/internal/seeds" |
| 12 | ) |
| 13 | |
| 14 | // Final review F4: an import's row is cached at the edge; the rollback |
| 15 | // closes the gate under its fence, puts the legacy address-backed route |
| 16 | // back and lifts the fence; a verify request then arrives before the edge |
| 17 | // reloads. The wake refreshes the row, finds the legacy route, and stops: |
| 18 | // no lease is taken and nothing is restored. The edge is never reloaded. |
| 19 | func TestAVerifyOnACachedImportRowAfterItsRollbackWakesNothing(t *testing.T) { |
| 20 | t.Parallel() |
| 21 | a, b, _, reg := testActivator(t, time.Hour) |
| 22 | a.Routes = reg |
| 23 | const host = "old.shop.example.com" |
| 24 | if err := reg.Upsert(registry.Route{Host: host, Kind: "instance", Addr: "127.0.0.1:8401", App: "shop", Account: "bab"}); err != nil { |
| 25 | t.Fatal(err) |
| 26 | } |
| 27 | original, err := reg.Get(host) |
| 28 | if err != nil || original == nil { |
| 29 | t.Fatalf("the legacy route: %+v, %v", original, err) |
| 30 | } |
| 31 | id, err := seeds.NewID() |
| 32 | if err != nil { |
| 33 | t.Fatal(err) |
| 34 | } |
| 35 | token := "s3cret-token" |
| 36 | imported := registry.Route{Host: host, Kind: "instance", App: "shop", Account: "bab", Provisioned: true, Hibernate: true, |
| 37 | Socket: "/run/carlos/instances/old/app.sock", DBPath: "/data/instances/old/" + host + ".db", |
| 38 | ImportGate: seeds.HashToken(token), SeedSource: seeds.RowSeed{ID: id, Kind: seeds.KindImport, State: seeds.StateSwapped}.Encode()} |
| 39 | if err := reg.SwapForImport(registry.ImportJournal{Host: host, SeedID: id, Original: *original, |
| 40 | RecordKey: "apps/bab/shop/instances/" + host + ".json", At: 1}, imported); err != nil { |
| 41 | t.Fatal(err) |
| 42 | } |
| 43 | edge, err := newTestRouter(reg, a) // the edge caches the gated import row |
| 44 | if err != nil { |
| 45 | t.Fatal(err) |
| 46 | } |
| 47 | |
| 48 | // The rollback, steps 1 to 5. |
| 49 | const gen = 7 |
| 50 | if err := reg.CloseImportGate(host, id, gen); err != nil { |
| 51 | t.Fatal(err) |
| 52 | } |
| 53 | fctx, cancel := context.WithTimeout(context.Background(), time.Second) |
| 54 | _, _ = a.FenceAndStopIfUnfenced(fctx, host, gen) // nothing is running: the stop is unproven, the fence is held |
| 55 | cancel() |
| 56 | if a.FenceGeneration(host) != gen { |
| 57 | t.Fatal("the rollback's fence was not installed") |
| 58 | } |
| 59 | if err := reg.RestoreImportRoute(host, id); err != nil { |
| 60 | t.Fatal(err) |
| 61 | } |
| 62 | if !a.LiftFence(host, gen) { |
| 63 | t.Fatal("the rollback's fence was not lifted") |
| 64 | } |
| 65 | |
| 66 | req := httptest.NewRequest(http.MethodGet, "http://"+host+"/up", nil) |
| 67 | req.Header.Set("Carlos-Import-Verify", token) |
| 68 | rec := httptest.NewRecorder() |
| 69 | edge.ServeHTTP(rec, req) |
| 70 | if rec.Code == http.StatusOK { |
| 71 | t.Fatalf("the verify was answered %d through the restored legacy route's wake", rec.Code) |
| 72 | } |
| 73 | b.mu.Lock() |
| 74 | acquires, restores := b.acquires, b.restores |
| 75 | b.mu.Unlock() |
| 76 | if acquires != 0 || restores != 0 { |
| 77 | t.Fatalf("the wake took %d leases and ran %d restores for a route the activator no longer manages", acquires, restores) |
| 78 | } |
| 79 | if got, _ := reg.Get(host); got == nil || got.Addr != original.Addr || got.ImportGate != "" { |
| 80 | t.Fatalf("the legacy route after the refused wake = %+v", got) |
| 81 | } |
| 82 | } |
| 83 | |
| 84 | // The refreshed row is compared with the row the wake began from: a row |
| 85 | // the activator does not manage, or one that is another route (account, |
| 86 | // app, address, import gate, seed), is refused. Paths, versions and a |
| 87 | // seed's own state moving on are not. |
| 88 | func TestWakeRouteChanged(t *testing.T) { |
| 89 | id, _ := seeds.NewID() |
| 90 | other, _ := seeds.NewID() |
| 91 | began := testRoute("h.example.com") |
| 92 | began.Account, began.App = "bab", "shop" |
| 93 | began.SeedSource = seeds.RowSeed{ID: id, Kind: seeds.KindReplica, State: seeds.StateReady}.Encode() |
| 94 | for _, c := range []struct { |
| 95 | name string |
| 96 | edit func(*registry.Route) |
| 97 | refused bool |
| 98 | }{ |
| 99 | {"unchanged", func(*registry.Route) {}, false}, |
| 100 | {"paths moved", func(r *registry.Route) { |
| 101 | r.DBPath, r.Socket = "/data/instances/h/h.db", "/run/carlos/instances/h/app.sock" |
| 102 | }, false}, |
| 103 | {"the seed placed", func(r *registry.Route) { |
| 104 | r.SeedSource = seeds.RowSeed{ID: id, Kind: seeds.KindReplica, State: seeds.StatePlaced}.Encode() |
| 105 | }, false}, |
| 106 | {"not hibernating", func(r *registry.Route) { r.Hibernate = false }, true}, |
| 107 | {"no socket", func(r *registry.Route) { r.Socket, r.Addr = "", "127.0.0.1:8401" }, true}, |
| 108 | {"unit-backed", func(r *registry.Route) { r.Backing = "unit" }, true}, |
| 109 | {"another account", func(r *registry.Route) { r.Account = "cdef" }, true}, |
| 110 | {"another app", func(r *registry.Route) { r.App = "blog" }, true}, |
| 111 | {"an address", func(r *registry.Route) { r.Addr = "127.0.0.1:8401" }, true}, |
| 112 | {"gated", func(r *registry.Route) { r.ImportGate = registry.ImportGateClosed }, true}, |
| 113 | {"another seed", func(r *registry.Route) { |
| 114 | r.SeedSource = seeds.RowSeed{ID: other, Kind: seeds.KindImport, State: seeds.StateSwapped}.Encode() |
| 115 | }, true}, |
| 116 | {"no seed", func(r *registry.Route) { r.SeedSource = "" }, true}, |
| 117 | } { |
| 118 | now := began |
| 119 | c.edit(&now) |
| 120 | if err := wakeRouteChanged(began, now, false); (err != nil) != c.refused { |
| 121 | t.Errorf("%s: %v, want refused %v", c.name, err, c.refused) |
| 122 | } |
| 123 | } |
| 124 | } |
| 125 | |
| 126 | // Re-review finding 9: a standalone sidecar wakes through WakeWorker on the |
| 127 | // same path, with Routes wired to the registry as production has it. Its |
| 128 | // row is no ActivatorManaged instance (no socket, kind sidecar), and it |
| 129 | // must still wake; a sidecar whose identity changed under the wake is still |
| 130 | // refused before any lease. |
| 131 | func TestWakeWorkerWithTheRegistryWired(t *testing.T) { |
| 132 | t.Parallel() |
| 133 | a, b, _, reg := testActivator(t, 10*time.Second) |
| 134 | a.Routes = reg |
| 135 | rt := testWorkerRoute("bot.shop.example.com") |
| 136 | rt.Account, rt.App = "bab", "shop" |
| 137 | if err := reg.Upsert(rt); err != nil { |
| 138 | t.Fatal(err) |
| 139 | } |
| 140 | if err := a.WakeWorker(rt); err != nil { |
| 141 | t.Fatalf("WakeWorker of an unchanged sidecar: %v", err) |
| 142 | } |
| 143 | if got := b.events(); !contains(got, "startWorker:bot.shop.example.com") { |
| 144 | t.Fatalf("events %v; want the sidecar started", got) |
| 145 | } |
| 146 | |
| 147 | moved := testWorkerRoute("bot2.shop.example.com") |
| 148 | moved.Account, moved.App = "bab", "shop" |
| 149 | if err := reg.Upsert(moved); err != nil { |
| 150 | t.Fatal(err) |
| 151 | } |
| 152 | stale := moved |
| 153 | stale.App = "blog" // the caller's copy names another app |
| 154 | b.mu.Lock() |
| 155 | before := b.acquires |
| 156 | b.mu.Unlock() |
| 157 | if err := a.WakeWorker(stale); err == nil { |
| 158 | t.Fatal("a sidecar whose identity changed under the wake woke") |
| 159 | } |
| 160 | b.mu.Lock() |
| 161 | after := b.acquires |
| 162 | b.mu.Unlock() |
| 163 | if after != before { |
| 164 | t.Fatalf("the refused wake took %d leases", after-before) |
| 165 | } |
| 166 | } |
| 167 | |
| 168 | // The activator is handed instance rows that are not ActivatorManaged too, |
| 169 | // with Routes wired as production has it: a unit-backed row (SplitBackend's |
| 170 | // Unit half) and a row with no socket column. A wake that began for one |
| 171 | // goes on while the row keeps its shape, and is refused when the shape or |
| 172 | // the identity changed under it. A row that began ActivatorManaged is held |
| 173 | // to that (TestWakeRouteChanged). |
| 174 | func TestWakeRouteChangedForARowOutsideActivatorManaged(t *testing.T) { |
| 175 | unit := testRoute("u.example.com") |
| 176 | unit.Account, unit.App, unit.Backing = "bab", "shop", registry.BackingUnit |
| 177 | bare := registry.Route{Host: "bare.example.com", Kind: "instance", Hibernate: true, Account: "bab", App: "shop"} |
| 178 | for _, began := range []registry.Route{unit, bare} { |
| 179 | if ActivatorManaged(began) { |
| 180 | t.Fatalf("%s: the fixture is ActivatorManaged", began.Host) |
| 181 | } |
| 182 | if err := wakeRouteChanged(began, began, false); err != nil { |
| 183 | t.Fatalf("%s: an unchanged row: %v", began.Host, err) |
| 184 | } |
| 185 | for name, edit := range map[string]func(*registry.Route){ |
| 186 | "another backing": func(r *registry.Route) { |
| 187 | if r.Backing == "" { |
| 188 | r.Backing = registry.BackingUnit |
| 189 | } else { |
| 190 | r.Backing = "" |
| 191 | } |
| 192 | }, |
| 193 | "not hibernating": func(r *registry.Route) { r.Hibernate = false }, |
| 194 | "another kind": func(r *registry.Route) { r.Kind = "service" }, |
| 195 | "a tunnel": func(r *registry.Route) { r.Tunnel = "t1" }, |
| 196 | "the socket came or went": func(r *registry.Route) { |
| 197 | if r.Socket == "" { |
| 198 | r.Socket = "/run/carlos/x.sock" |
| 199 | } else { |
| 200 | r.Socket = "" |
| 201 | } |
| 202 | }, |
| 203 | "another account": func(r *registry.Route) { r.Account = "cdef" }, |
| 204 | "another app": func(r *registry.Route) { r.App = "blog" }, |
| 205 | "an address": func(r *registry.Route) { r.Addr = "127.0.0.1:8401" }, |
| 206 | "gated": func(r *registry.Route) { r.ImportGate = registry.ImportGateClosed }, |
| 207 | } { |
| 208 | now := began |
| 209 | edit(&now) |
| 210 | if err := wakeRouteChanged(began, now, false); err == nil { |
| 211 | t.Errorf("%s, %s: a row that changed under its wake was not refused", began.Host, name) |
| 212 | } |
| 213 | } |
| 214 | } |
| 215 | } |
| 216 | |
| 217 | // The role decides the shape: a worker entry needs a sidecar row, an |
| 218 | // instance entry an ActivatorManaged one; the identity checks hold for both. |
| 219 | func TestWakeRouteChangedByRole(t *testing.T) { |
| 220 | sidecar := testWorkerRoute("bot.shop.example.com") |
| 221 | sidecar.Account, sidecar.App = "bab", "shop" |
| 222 | if err := wakeRouteChanged(sidecar, sidecar, true); err != nil { |
| 223 | t.Fatalf("an unchanged sidecar for a worker entry: %v", err) |
| 224 | } |
| 225 | if err := wakeRouteChanged(sidecar, sidecar, false); err == nil { |
| 226 | t.Fatal("a sidecar row passed for an instance entry") |
| 227 | } |
| 228 | inst := testRoute("h.example.com") |
| 229 | if err := wakeRouteChanged(inst, inst, true); err == nil { |
| 230 | t.Fatal("an instance row passed for a worker entry") |
| 231 | } |
| 232 | other := sidecar |
| 233 | other.Account = "cdef" |
| 234 | if err := wakeRouteChanged(sidecar, other, true); err == nil { |
| 235 | t.Fatal("a worker entry woke a sidecar of another account") |
| 236 | } |
| 237 | } |
| 238 | |
amadan