| 1 | package activator |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "context" |
| 6 | "encoding/json" |
| 7 | "errors" |
| 8 | "net" |
| 9 | "os" |
| 10 | "path/filepath" |
| 11 | "strings" |
| 12 | "sync" |
| 13 | "testing" |
| 14 | "time" |
| 15 | |
| 16 | "github.com/carlosframework/platform/internal/instancectl" |
| 17 | "github.com/carlosframework/platform/internal/provision" |
| 18 | "github.com/carlosframework/platform/internal/registry" |
| 19 | ) |
| 20 | |
| 21 | // syncBuffer is a bytes.Buffer two goroutines may write. |
| 22 | type syncBuffer struct { |
| 23 | mu sync.Mutex |
| 24 | b bytes.Buffer |
| 25 | } |
| 26 | |
| 27 | func (s *syncBuffer) Write(p []byte) (int, error) { |
| 28 | s.mu.Lock() |
| 29 | defer s.mu.Unlock() |
| 30 | return s.b.Write(p) |
| 31 | } |
| 32 | func (s *syncBuffer) String() string { s.mu.Lock(); defer s.mu.Unlock(); return s.b.String() } |
| 33 | |
| 34 | func TestCmdVarsComposeTheRunEnvironment(t *testing.T) { |
| 35 | b := &ExecBackend{Region: "ie"} |
| 36 | rt := registry.Route{Host: "h.example.com"} |
| 37 | file := map[string]string{ |
| 38 | "APP_KEY": "k", "CARLOS_ADMIN_TOKEN": "tok", "CARLOS_CONTROL_SOCKET": "/x", "CARLOS_PURGE": "header", |
| 39 | "CARLOS_REGION": "us", "CARLOS_HOST": "liar.example.com", "CARLOS_RUN_ID": "planted", |
| 40 | } |
| 41 | got := b.cmdVars(rt, file, "/data/instances/h.example.com/app.db", "run_a", "/data/instances/h.example.com/.carlos-run/run_a") |
| 42 | want := map[string]string{ |
| 43 | "APP_KEY": "k", "CARLOS_ADMIN_TOKEN": "tok", "CARLOS_REGION": "ie", "CARLOS_HOST": "h.example.com", |
| 44 | "CARLOS_DB": "/data/instances/h.example.com/app.db", "CARLOS_RUN_ID": "run_a", |
| 45 | "CARLOS_RUN_DIR": "/data/instances/h.example.com/.carlos-run/run_a", |
| 46 | } |
| 47 | if len(got) != len(want) { |
| 48 | t.Fatalf("env = %v", got) |
| 49 | } |
| 50 | for k, v := range want { |
| 51 | if got[k] != v { |
| 52 | t.Errorf("%s = %q, want %q", k, got[k], v) |
| 53 | } |
| 54 | } |
| 55 | b.Region = "" |
| 56 | if _, ok := b.cmdVars(rt, file, "/db", "run_a", "/d")["CARLOS_REGION"]; ok { |
| 57 | t.Error("an empty backend region must remove CARLOS_REGION (codex 7)") |
| 58 | } |
| 59 | } |
| 60 | |
| 61 | func startTestCmd(t *testing.T) (*ExecBackend, registry.Route, *fakeControl, *fakeUnit, CmdUnit, *syncBuffer, *syncBuffer) { |
| 62 | t.Helper() |
| 63 | ctl := &fakeControl{cmdInv: "inv1"} |
| 64 | unit := &fakeUnit{} |
| 65 | b, rt := nestedBackend(t, ctl, unit) |
| 66 | b.layoutFn = func(registry.Route) (Layout, string) { return LayoutNested, "" } |
| 67 | out, errs := &syncBuffer{}, &syncBuffer{} |
| 68 | dir := filepath.Join(provision.InstanceDir(b.DataDir, rt.Host), CmdRunDirName, "run_a") |
| 69 | u, err := b.StartCmd(context.Background(), rt, CmdSpec{RunID: "run_a", Args: []string{"import", "--dry-run"}, RunDir: dir, Stdout: out, Stderr: errs, |
| 70 | genEnv: map[string]string{"CARLOS_ADMIN_TOKEN": "tok"}}) |
| 71 | if err != nil { |
| 72 | t.Fatalf("StartCmd: %v", err) |
| 73 | } |
| 74 | return b, rt, ctl, unit, u, out, errs |
| 75 | } |
| 76 | |
| 77 | // shrinkCmdAcceptGrace sets how long a cmd stream's Drain waits for a |
| 78 | // connection that never came, for one test. The fake units here never |
| 79 | // connect their streams, so every Drain waits the whole grace. |
| 80 | func shrinkCmdAcceptGrace(t *testing.T, d time.Duration) { |
| 81 | t.Helper() |
| 82 | old := cmdAcceptGrace |
| 83 | cmdAcceptGrace = d |
| 84 | t.Cleanup(func() { cmdAcceptGrace = old }) |
| 85 | } |
| 86 | |
| 87 | func TestStartCmdWritesTheLaunchFileAndStartsTheUnit(t *testing.T) { |
| 88 | shrinkCmdAcceptGrace(t, 50*time.Millisecond) // the cleanup drains two unconnected streams |
| 89 | b, rt, ctl, _, u, _, _ := startTestCmd(t) |
| 90 | t.Cleanup(func() { u.StopAndSettle(context.Background()) }) |
| 91 | if u.Invocation() != "inv1" || !strings.Contains(ctl.joined(), "cmd "+rt.Host) { |
| 92 | t.Fatalf("invocation %q, calls\n%s", u.Invocation(), ctl.joined()) |
| 93 | } |
| 94 | raw, err := os.ReadFile(filepath.Join(provision.InstanceDir(b.DataDir, rt.Host), "launch.cmd.json")) |
| 95 | if err != nil { |
| 96 | t.Fatal(err) |
| 97 | } |
| 98 | var l instancectl.Launch |
| 99 | if err := json.Unmarshal(raw, &l); err != nil { |
| 100 | t.Fatal(err) |
| 101 | } |
| 102 | if len(l.Argv) != 3 || l.Argv[0] != b.liveLink(rt.Host) || l.Argv[1] != "import" || l.Argv[2] != "--dry-run" { |
| 103 | t.Errorf("argv = %q: exactly the binary and what followed --", l.Argv) |
| 104 | } |
| 105 | env := strings.Join(l.Env, "\n") |
| 106 | for _, want := range []string{"CARLOS_RUN_ID=run_a", "CARLOS_HOST=" + rt.Host, "CARLOS_DB=" + rt.DBPath} { |
| 107 | if !strings.Contains(env, want) { |
| 108 | t.Errorf("env lacks %q", want) |
| 109 | } |
| 110 | } |
| 111 | if strings.Contains(env, "CARLOS_CONTROL_SOCKET=") { |
| 112 | t.Error("a command has no control socket") |
| 113 | } |
| 114 | } |
| 115 | |
| 116 | func TestStartCmdRefusesAFlatInstanceAndAForeignRunDir(t *testing.T) { |
| 117 | ctl := &fakeControl{cmdInv: "inv1"} |
| 118 | b, rt := nestedBackend(t, ctl, &fakeUnit{}) |
| 119 | b.layoutFn = func(registry.Route) (Layout, string) { return LayoutFlat, "test" } |
| 120 | if _, err := b.StartCmd(context.Background(), rt, CmdSpec{RunID: "run_a"}); !errors.Is(err, ErrCmdNotNested) { |
| 121 | t.Errorf("flat: %v, want ErrCmdNotNested", err) |
| 122 | } |
| 123 | b.layoutFn = func(registry.Route) (Layout, string) { return LayoutNested, "" } |
| 124 | if _, err := b.StartCmd(context.Background(), rt, CmdSpec{RunID: "run_a", RunDir: "/elsewhere"}); err == nil { |
| 125 | t.Error("a run directory outside the instance's .carlos-run was accepted") |
| 126 | } |
| 127 | if strings.Contains(ctl.joined(), "cmd ") { |
| 128 | t.Errorf("a refused start reached the control service:\n%s", ctl.joined()) |
| 129 | } |
| 130 | } |
| 131 | |
| 132 | func TestStartCmdRefusesAnUnsafeRunID(t *testing.T) { |
| 133 | for _, id := range []string{"../x", "", "a/b", ".", "..", "../../other.example.com", `a\b`} { |
| 134 | ctl := &fakeControl{cmdInv: "inv1"} |
| 135 | b, rt := nestedBackend(t, ctl, &fakeUnit{}) |
| 136 | b.layoutFn = func(registry.Route) (Layout, string) { return LayoutNested, "" } |
| 137 | dir := provision.InstanceDir(b.DataDir, rt.Host) |
| 138 | // the RunDir that matches the id after cleaning: the caller is |
| 139 | // "consistent", so only the id itself can be refused |
| 140 | runDir := filepath.Join(dir, CmdRunDirName, id) |
| 141 | if _, err := b.StartCmd(context.Background(), rt, CmdSpec{RunID: id, RunDir: runDir}); err == nil { |
| 142 | t.Errorf("run id %q was accepted", id) |
| 143 | } |
| 144 | if strings.Contains(ctl.joined(), "cmd ") { |
| 145 | t.Errorf("run id %q reached the control service", id) |
| 146 | } |
| 147 | if _, err := os.Stat(dir); !os.IsNotExist(err) { |
| 148 | t.Errorf("run id %q: the instance directory was made (%v)", id, err) |
| 149 | } |
| 150 | } |
| 151 | } |
| 152 | |
| 153 | func TestStopAndSettleDrainsBothStreams(t *testing.T) { |
| 154 | t.Parallel() |
| 155 | b, rt, ctl, unit, u, out, errs := startTestCmd(t) |
| 156 | unit.set("active", "inv1", true, true) |
| 157 | write := func(path, s string) { |
| 158 | c, err := net.Dial("unix", path) |
| 159 | if err != nil { |
| 160 | t.Fatal(err) |
| 161 | } |
| 162 | if _, err := c.Write([]byte(s)); err != nil { |
| 163 | t.Fatal(err) |
| 164 | } |
| 165 | c.Close() |
| 166 | } |
| 167 | write(b.logSocketNamed(rt.Host, string(instancectl.RoleCmd)), "out-1\nout-2") |
| 168 | write(b.logSocketNamed(rt.Host, cmdErrSocket), "err-1") |
| 169 | go func() { |
| 170 | time.Sleep(30 * time.Millisecond) |
| 171 | unit.set("inactive", "inv1", false, false) |
| 172 | }() |
| 173 | if err := u.StopAndSettle(context.Background()); err != nil { |
| 174 | t.Fatalf("StopAndSettle: %v", err) |
| 175 | } |
| 176 | if out.String() != "out-1\nout-2" || errs.String() != "err-1" { |
| 177 | t.Errorf("stdout %q stderr %q", out.String(), errs.String()) |
| 178 | } |
| 179 | if !strings.Contains(ctl.joined(), "stop "+rt.Host+" cmd") { |
| 180 | t.Errorf("the cmd unit was not stopped:\n%s", ctl.joined()) |
| 181 | } |
| 182 | if err := u.StopAndSettle(context.Background()); err != nil { |
| 183 | t.Errorf("a second StopAndSettle: %v", err) |
| 184 | } |
| 185 | } |
| 186 | |
| 187 | // A connection the unit made at exec may still be queued in the listener's |
| 188 | // backlog when the unit has exited: Drain must take it and deliver its |
| 189 | // bytes (run spec §5.5; plan Review Focus 5). |
| 190 | func TestCmdStreamDrainTakesAConnectionStillInTheBacklog(t *testing.T) { |
| 191 | dir, err := os.MkdirTemp("", "cs") |
| 192 | if err != nil { |
| 193 | t.Fatal(err) |
| 194 | } |
| 195 | defer os.RemoveAll(dir) |
| 196 | for i := 0; i < 200; i++ { |
| 197 | var w syncBuffer |
| 198 | s, err := listenCmdStream(filepath.Join(dir, "s.sock"), &w) |
| 199 | if err != nil { |
| 200 | t.Fatal(err) |
| 201 | } |
| 202 | c, err := net.Dial("unix", filepath.Join(dir, "s.sock")) |
| 203 | if err != nil { |
| 204 | t.Fatal(err) |
| 205 | } |
| 206 | if _, err := c.Write([]byte("last words")); err != nil { |
| 207 | t.Fatal(err) |
| 208 | } |
| 209 | c.Close() |
| 210 | if err := s.Drain(time.Second); err != nil { |
| 211 | t.Fatalf("round %d: Drain: %v", i, err) |
| 212 | } |
| 213 | if w.String() != "last words" { |
| 214 | t.Fatalf("round %d: got %q", i, w.String()) |
| 215 | } |
| 216 | } |
| 217 | } |
| 218 | |
| 219 | func TestCmdStreamDrainOfAnUnconnectedStreamEndsAfterTheGrace(t *testing.T) { |
| 220 | shrinkCmdAcceptGrace(t, 100*time.Millisecond) |
| 221 | dir, err := os.MkdirTemp("", "cs") |
| 222 | if err != nil { |
| 223 | t.Fatal(err) |
| 224 | } |
| 225 | defer os.RemoveAll(dir) |
| 226 | s, err := listenCmdStream(filepath.Join(dir, "s.sock"), nil) |
| 227 | if err != nil { |
| 228 | t.Fatal(err) |
| 229 | } |
| 230 | start := time.Now() |
| 231 | if err := s.Drain(time.Second); err != nil { |
| 232 | t.Fatal(err) |
| 233 | } |
| 234 | if d := time.Since(start); d > 3*cmdAcceptGrace { |
| 235 | t.Errorf("took %s", d) |
| 236 | } |
| 237 | } |
| 238 | |
amadan