| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "fmt" |
| 6 | "net/http" |
| 7 | "os" |
| 8 | "path/filepath" |
| 9 | "testing" |
| 10 | "time" |
| 11 | |
| 12 | "github.com/carlosframework/platform/internal/activator" |
| 13 | "github.com/carlosframework/platform/internal/registry" |
| 14 | "github.com/carlosframework/platform/internal/schedule" |
| 15 | ) |
| 16 | |
| 17 | // genWakeBackend is fakeWakeBackend with the generation capability: with |
| 18 | // perGen it starts the fake instance on <dir>/g<n>.sock, never on rt.Socket, |
| 19 | // so a tick that reached the canonical path would find nothing. |
| 20 | type genWakeBackend struct { |
| 21 | fakeWakeBackend |
| 22 | dir string |
| 23 | } |
| 24 | |
| 25 | func (b *genWakeBackend) StartGen(_ context.Context, rt registry.Route, n uint64, perGen bool) (activator.Proc, activator.Gen, error) { |
| 26 | sock := rt.Socket |
| 27 | if perGen { |
| 28 | sock = filepath.Join(b.dir, fmt.Sprintf("g%d.sock", n)) |
| 29 | } |
| 30 | b.inst.socket = sock |
| 31 | if err := b.inst.start(); err != nil { |
| 32 | return nil, activator.Gen{}, err |
| 33 | } |
| 34 | return &fakeInstanceProc{inst: b.inst}, activator.Gen{N: n, Socket: sock, Control: sock + ".control"}, nil |
| 35 | } |
| 36 | |
| 37 | func (b *genWakeBackend) NextDueVia(context.Context, registry.Route, string) (int64, error) { |
| 38 | return 0, nil |
| 39 | } |
| 40 | |
| 41 | // genDeliveryFixture is wakeFixture over genWakeBackend, with the |
| 42 | // generations switch set before the activator reads it. |
| 43 | func genDeliveryFixture(t *testing.T, on bool) (*activator.Service, *fakeInstance, registry.Route) { |
| 44 | t.Helper() |
| 45 | prev := activator.GenerationsEnabled() |
| 46 | activator.SetGenerations(on) |
| 47 | t.Cleanup(func() { activator.SetGenerations(prev) }) |
| 48 | dd, err := os.MkdirTemp("", "h") |
| 49 | if err != nil { |
| 50 | t.Fatal(err) |
| 51 | } |
| 52 | t.Cleanup(func() { os.RemoveAll(dd) }) |
| 53 | reg, err := registry.Open(filepath.Join(dd, "r.db")) |
| 54 | if err != nil { |
| 55 | t.Fatal(err) |
| 56 | } |
| 57 | t.Cleanup(func() { reg.Close() }) |
| 58 | inst := &fakeInstance{handler: http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusNoContent) })} |
| 59 | rt := registry.Route{Host: "gen.bab.oncarlos.com", Kind: "instance", Account: "bab", App: "app", Status: "ready", |
| 60 | Hibernate: true, Socket: filepath.Join(dd, "canonical.sock"), DBPath: filepath.Join(dd, "app.db")} |
| 61 | if err := reg.Upsert(rt); err != nil { |
| 62 | t.Fatal(err) |
| 63 | } |
| 64 | svc := activator.New(activator.Config{Bucket: "test-bucket", Idle: time.Hour}, &genWakeBackend{fakeWakeBackend: fakeWakeBackend{inst: inst}, dir: dd}, reg) |
| 65 | svc.Start() |
| 66 | t.Cleanup(func() { |
| 67 | ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) |
| 68 | defer cancel() |
| 69 | svc.Shutdown(ctx) |
| 70 | }) |
| 71 | return svc, inst, rt |
| 72 | } |
| 73 | |
| 74 | func deliverTick(t *testing.T, d *activatorDeliverer, rt registry.Route) int { |
| 75 | t.Helper() |
| 76 | req, err := schedule.BuildRequest(rt, registry.Timer{Name: "sync", Kind: schedule.KindSchedule, Path: "/jobs/sync", NextAt: 1, DueAt: 1}, "tok", "test") |
| 77 | if err != nil { |
| 78 | t.Fatal(err) |
| 79 | } |
| 80 | status, err := d.Deliver(context.Background(), rt, req) |
| 81 | if err != nil { |
| 82 | t.Fatalf("Deliver: %v", err) |
| 83 | } |
| 84 | return status |
| 85 | } |
| 86 | |
| 87 | // Scheduled delivery dials AppSocket (spec §7.1): the tick wakes the |
| 88 | // instance and reaches its generation socket, though nothing listens on the |
| 89 | // route's canonical path. |
| 90 | func TestScheduledDeliveryReachesTheCurrentGeneration(t *testing.T) { |
| 91 | requireUnixSockets(t) |
| 92 | svc, inst, rt := genDeliveryFixture(t, true) |
| 93 | d := &activatorDeliverer{svc: svc} |
| 94 | if got := deliverTick(t, d, rt); got != http.StatusNoContent { |
| 95 | t.Fatalf("tick status = %d, want 204 from the instance", got) |
| 96 | } |
| 97 | cur, ok := svc.AppSocket(rt.Host) |
| 98 | if !ok || inst.socket != cur || cur == rt.Socket { |
| 99 | t.Fatalf("instance bound %q, AppSocket = %q (%v); want a generation socket, not %q", inst.socket, cur, ok, rt.Socket) |
| 100 | } |
| 101 | if len(inst.served()) != 1 { |
| 102 | t.Fatalf("instance served %d ticks, want 1", len(inst.served())) |
| 103 | } |
| 104 | if d.genPools == nil || d.genPools.Len() != 1 { |
| 105 | t.Fatal("the tick did not go through the leg's generation pools") |
| 106 | } |
| 107 | } |
| 108 | |
| 109 | // Flag off: the canonical socket and the cached per-socket proxy, as before. |
| 110 | func TestScheduledDeliveryWithGenerationsOffUsesTheCanonicalSocket(t *testing.T) { |
| 111 | requireUnixSockets(t) |
| 112 | svc, inst, rt := genDeliveryFixture(t, false) |
| 113 | d := &activatorDeliverer{svc: svc} |
| 114 | if got := deliverTick(t, d, rt); got != http.StatusNoContent { |
| 115 | t.Fatalf("tick status = %d, want 204", got) |
| 116 | } |
| 117 | if inst.socket != rt.Socket { |
| 118 | t.Fatalf("instance bound %q, want the canonical %q", inst.socket, rt.Socket) |
| 119 | } |
| 120 | if d.genPools != nil { |
| 121 | t.Fatal("generation pools were made with the switch off") |
| 122 | } |
| 123 | if _, ok := d.proxies[rt.Socket]; !ok { |
| 124 | t.Fatal("the canonical per-socket proxy was not used") |
| 125 | } |
| 126 | } |
| 127 | |
| 128 | // The status probe asks about each managed route's current generation, and |
| 129 | // about everything else exactly as before. |
| 130 | func TestCurrentSocketProbePointsAtTheCurrentGeneration(t *testing.T) { |
| 131 | routes := []registry.Route{ |
| 132 | {Host: "awake.example.com", Kind: "instance", Hibernate: true, Socket: "/run/carlos/awake.example.com.sock"}, |
| 133 | {Host: "asleep.example.com", Kind: "instance", Hibernate: true, Socket: "/run/carlos/asleep.example.com.sock"}, |
| 134 | {Host: "svc.example.com", Kind: "service", Socket: "/run/carlos/svc.example.com.sock"}, |
| 135 | } |
| 136 | appSocket := func(host string) (string, bool) { |
| 137 | if host == "awake.example.com" { |
| 138 | return "/run/carlos/g/0123456789ab.4.sock", true |
| 139 | } |
| 140 | return "", false |
| 141 | } |
| 142 | var asked []string |
| 143 | probe := currentSocketProbe(routes, appSocket, func(sock string) *bool { |
| 144 | asked = append(asked, sock) |
| 145 | v := true |
| 146 | return &v |
| 147 | }) |
| 148 | for _, r := range routes { |
| 149 | probe(r.Socket) |
| 150 | } |
| 151 | want := []string{"/run/carlos/g/0123456789ab.4.sock", "/run/carlos/asleep.example.com.sock", "/run/carlos/svc.example.com.sock"} |
| 152 | if fmt.Sprint(asked) != fmt.Sprint(want) { |
| 153 | t.Fatalf("probed %v, want %v", asked, want) |
| 154 | } |
| 155 | } |
| 156 | |
| 157 | // Flag off (or no activator): the status loop's probe is listeningProbe. |
| 158 | func TestStatusSocketProbeIsUnchangedWithGenerationsOff(t *testing.T) { |
| 159 | got := statusSocketProbe(nil, nil) |
| 160 | if fmt.Sprintf("%p", got) != fmt.Sprintf("%p", listeningProbe) { |
| 161 | t.Fatal("with no activator the status probe is not listeningProbe") |
| 162 | } |
| 163 | svc, _, _ := genDeliveryFixture(t, false) |
| 164 | got = statusSocketProbe(nil, svc) |
| 165 | if fmt.Sprintf("%p", got) != fmt.Sprintf("%p", listeningProbe) { |
| 166 | t.Fatal("with generations off the status probe is not listeningProbe") |
| 167 | } |
| 168 | } |
| 169 | |
amadan