| 1 | package activator |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "io" |
| 8 | "log" |
| 9 | "net" |
| 10 | "path/filepath" |
| 11 | "strings" |
| 12 | "sync" |
| 13 | "sync/atomic" |
| 14 | "time" |
| 15 | |
| 16 | "github.com/carlosframework/platform/internal/edge" |
| 17 | "github.com/carlosframework/platform/internal/instancectl" |
| 18 | "github.com/carlosframework/platform/internal/instancekey" |
| 19 | "github.com/carlosframework/platform/internal/registry" |
| 20 | ) |
| 21 | |
| 22 | // CmdRunDirName is the directory under an instance's own data directory |
| 23 | // that holds each run's --file uploads (run spec §5.2 step 3). |
| 24 | const CmdRunDirName = ".carlos-run" |
| 25 | |
| 26 | // cmdErrSocket is the stderr listener's name; stdout's is the role's. |
| 27 | const cmdErrSocket = "cmd-err" |
| 28 | |
| 29 | // cmdDrainBound is how long a stream may take to reach EOF once its unit |
| 30 | // has stopped. |
| 31 | const cmdDrainBound = 5 * time.Second |
| 32 | |
| 33 | // ErrCmdNotNested: the instance is not on the nested layout, so a command |
| 34 | // would run with the agent's reach (run spec §5.2 step 1, refused "flat"). |
| 35 | var ErrCmdNotNested = errors.New("activator: the instance is not on the nested layout") |
| 36 | |
| 37 | // CmdSpec is one run's launch, as the box's run pass hands it over. |
| 38 | type CmdSpec struct { |
| 39 | RunID string |
| 40 | // Args is exactly what followed `--`; the binary is the host's live link. |
| 41 | Args []string |
| 42 | // RunDir is CARLOS_RUN_DIR: <instance dir>/.carlos-run/<RunID>, made and |
| 43 | // filled by the caller. |
| 44 | RunDir string |
| 45 | // Stdout and Stderr receive the unit's two streams, byte for byte. |
| 46 | Stdout, Stderr io.Writer |
| 47 | // Prepare runs after the run is attached to the awake entry and before |
| 48 | // the unit starts (Service.RunCmd); an error ends the run before launch. |
| 49 | Prepare func(ctx context.Context) error |
| 50 | // Recheck, when set, is the run's launch eligibility checked again |
| 51 | // after RunCmd has waited for a swap's switch (re-review 5): Prepare's |
| 52 | // last check is that old by then. An error ends the run before launch, |
| 53 | // as Prepare's does. |
| 54 | Recheck func(ctx context.Context) error |
| 55 | |
| 56 | // serving is the serving generation's staged build, set by RunCmd at its |
| 57 | // last awake check when that generation names one, generations on or |
| 58 | // off (zero-downtime spec §7.5 and settings ruling R-EXEC-SIDECAR: what |
| 59 | // runs beside an instance is its serving artifact). Nil runs the live |
| 60 | // link. Callers cannot set it. |
| 61 | serving *instancectl.StagedExe |
| 62 | // genEnv is the file environment the run's generation launched with, |
| 63 | // from its handle (unitProc.LaunchEnv), set by RunCmd with serving |
| 64 | // (merge ruling M3). The run's variables are composed over it, never |
| 65 | // over the delivered files, which may hold a config revision that |
| 66 | // generation never activated. StartCmd refuses a run without one. |
| 67 | // Callers cannot set it. |
| 68 | genEnv map[string]string |
| 69 | // genKeyLayer is the layer genEnv's CARLOS_INSTANCE_KEY came from, from |
| 70 | // the same handle (unitProc.LaunchKeyLayer), for the replica gate. |
| 71 | // Callers cannot set it. |
| 72 | genKeyLayer instancekey.Layer |
| 73 | } |
| 74 | |
| 75 | // errNoGenEnv: the run's generation kept no launch environment, so the run |
| 76 | // would have to read the delivered files, which may differ from what the |
| 77 | // instance is serving with. It is refused instead (merge ruling M3). |
| 78 | var errNoGenEnv = errors.New("activator: the serving generation kept no launch environment for a run") |
| 79 | |
| 80 | // cmdExeStarter is the cmd start with the executable named (Client.CmdExe). |
| 81 | // A separate interface, like GenControl, so a Control without it still |
| 82 | // satisfies Control; StartCmd asks for it only when a run names a build. |
| 83 | type cmdExeStarter interface { |
| 84 | CmdExe(ctx context.Context, host string, memoryMax int64, x instancectl.StagedExe) (string, error) |
| 85 | } |
| 86 | |
| 87 | var _ cmdExeStarter = (*instancectl.Client)(nil) |
| 88 | |
| 89 | // errNoCmdExe: a run names its serving generation's build and the control |
| 90 | // channel cannot carry it. The run is refused rather than run on the live |
| 91 | // link, which after a fallback names a build the instance is not serving. |
| 92 | var errNoCmdExe = errors.New("activator: the instance-control client cannot name a run's build") |
| 93 | |
| 94 | // CmdUnit is a started cmd unit. |
| 95 | type CmdUnit interface { |
| 96 | Invocation() string |
| 97 | // Poll long-polls the invocation, at most about 30 seconds. |
| 98 | Poll(ctx context.Context) (instancectl.CmdStatus, error) |
| 99 | // StopAndSettle stops the unit (SIGTERM, then SIGKILL after 20 s), |
| 100 | // proves its cgroup empty and reads both streams to their end. It is |
| 101 | // safe to call more than once: once a call has succeeded, later ones |
| 102 | // return nil at once, and a call after a failure tries again (merge |
| 103 | // ruling M1). |
| 104 | StopAndSettle(ctx context.Context) error |
| 105 | } |
| 106 | |
| 107 | // CmdBackend is the backend capability Service.RunCmd needs. StartCmd may |
| 108 | // return a unit together with an error: the start failed after the unit |
| 109 | // may have begun, and the unit must still be stopped and settled. |
| 110 | type CmdBackend interface { |
| 111 | StartCmd(ctx context.Context, rt registry.Route, spec CmdSpec) (CmdUnit, error) |
| 112 | } |
| 113 | |
| 114 | var _ CmdBackend = (*ExecBackend)(nil) |
| 115 | |
| 116 | // cmdVars is a run's environment over fileEnv, the file environment its |
| 117 | // generation launched with (run spec §4, codex 7; merge ruling M3). That |
| 118 | // environment carries CARLOS_INSTANCE_KEY and CARLOS_ADMIN_TOKEN through: |
| 119 | // the command runs as the instance, on the instance's sealed data, with |
| 120 | // the key the serving generation holds and never one of its own. Over it: |
| 121 | // the control-channel variables stripped as seedVars strips them, |
| 122 | // CARLOS_REGION set or removed from the backend's region as agentOwnedVars |
| 123 | // does, CARLOS_HOST from the row, and the three run variables. |
| 124 | func (b *ExecBackend) cmdVars(rt registry.Route, fileEnv map[string]string, dbPath, runID, runDir string) map[string]string { |
| 125 | v := withoutVars(fileEnv, "CARLOS_CONTROL_SOCKET", "CARLOS_PURGE") |
| 126 | if b.Region != "" { |
| 127 | v = withVar(v, "CARLOS_REGION", b.Region) |
| 128 | } else { |
| 129 | v = withoutVars(v, "CARLOS_REGION") |
| 130 | } |
| 131 | v = withVar(v, "CARLOS_HOST", rt.Host) |
| 132 | v = withVar(v, "CARLOS_DB", dbPath) |
| 133 | v = withVar(v, "CARLOS_RUN_ID", runID) |
| 134 | return withVar(v, "CARLOS_RUN_DIR", runDir) |
| 135 | } |
| 136 | |
| 137 | // checkCmdRunDir refuses a run id that is not one plain path element and a |
| 138 | // run directory other than <instance dir>/.carlos-run/<RunID>. The run |
| 139 | // directory is deleted after the run, so it must never name anything |
| 140 | // outside the instance's own .carlos-run. It does no disk work. |
| 141 | func checkCmdRunDir(dataDir string, spec CmdSpec) error { |
| 142 | id := spec.RunID |
| 143 | if id == "" || id == "." || id == ".." || filepath.Base(id) != id || strings.ContainsAny(id, `/\`) { |
| 144 | return fmt.Errorf("run id %q is not a single path element", id) |
| 145 | } |
| 146 | root := filepath.Join(dataDir, CmdRunDirName) |
| 147 | if want := filepath.Join(root, id); spec.RunDir != want || filepath.Dir(spec.RunDir) != root { |
| 148 | return fmt.Errorf("run directory %q is not %q", spec.RunDir, want) |
| 149 | } |
| 150 | return nil |
| 151 | } |
| 152 | |
| 153 | func (b *ExecBackend) cmdLayout(rt registry.Route) (Layout, string) { |
| 154 | if b.layoutFn != nil { |
| 155 | return b.layoutFn(rt) |
| 156 | } |
| 157 | return b.Layout(rt) |
| 158 | } |
| 159 | |
| 160 | // StartCmd starts one run's cmd unit (run spec §5.4): the layout check, the |
| 161 | // nested gate, the live link, the instance's directories, the environment |
| 162 | // its generation launched with (spec.genEnv, merge ruling M3), the two |
| 163 | // stream listeners, launch.cmd.json, then POST /cmd. It returns |
| 164 | // once the unit has an invocation. The environment is never logged. A |
| 165 | // failure after the POST was sent returns a unit with the error: the unit |
| 166 | // may be running, and the caller must stop and settle it. |
| 167 | // |
| 168 | // A run that names its serving generation's staged build (spec.serving) |
| 169 | // has root mount that build instead of the live link, and needs no live |
| 170 | // link of its own. The argv is the same /opt/carlos/live/<host> inside the |
| 171 | // namespace either way. |
| 172 | func (b *ExecBackend) StartCmd(ctx context.Context, rt registry.Route, spec CmdSpec) (CmdUnit, error) { |
| 173 | if l, why := b.cmdLayout(rt); l != LayoutNested { |
| 174 | return nil, fmt.Errorf("%w: %s is %s (%s)", ErrCmdNotNested, rt.Host, l, why) |
| 175 | } |
| 176 | if err := checkCmdRunDir(b.pathsFor(rt).dataDir, spec); err != nil { |
| 177 | return nil, fmt.Errorf("%s: %w", rt.Host, err) |
| 178 | } |
| 179 | if err := b.nestedGate(rt); err != nil { |
| 180 | return nil, err |
| 181 | } |
| 182 | var exeStart cmdExeStarter |
| 183 | if spec.serving != nil { |
| 184 | cs, ok := b.Control.(cmdExeStarter) |
| 185 | if !ok { |
| 186 | return nil, fmt.Errorf("%s: %w", rt.Host, errNoCmdExe) |
| 187 | } |
| 188 | exeStart = cs |
| 189 | } else if err := b.ensureLiveLink(ctx, rt.Host); err != nil { |
| 190 | return nil, err |
| 191 | } |
| 192 | if spec.genEnv == nil { |
| 193 | return nil, fmt.Errorf("%s: %w", rt.Host, errNoGenEnv) |
| 194 | } |
| 195 | // A run's command is a launch like any other (instance-key rulings 26 |
| 196 | // to 29): on a replica it starts only with its home's key, checked on |
| 197 | // the environment it is about to be given and the layer that key came |
| 198 | // from. A command has no control socket (cmdVars strips it), so a |
| 199 | // replica's command with no key is refused, as a seed or a worker is: |
| 200 | // an app on the instance-key helper would otherwise mint its own key |
| 201 | // over the home's data. Nothing is laid out or started before this. |
| 202 | if err := b.replicaKeyGate(rt, spec.genEnv, spec.genKeyLayer, false); err != nil { |
| 203 | return nil, err |
| 204 | } |
| 205 | p, _, _, _, err := b.prepare(rt, instancectl.RoleCmd) |
| 206 | if err != nil { |
| 207 | return nil, err |
| 208 | } |
| 209 | out, err := listenCmdStream(b.logSocketNamed(rt.Host, string(instancectl.RoleCmd)), spec.Stdout) |
| 210 | if err != nil { |
| 211 | return nil, fmt.Errorf("stdout listener for %s: %w", rt.Host, err) |
| 212 | } |
| 213 | errs, err := listenCmdStream(b.logSocketNamed(rt.Host, cmdErrSocket), spec.Stderr) |
| 214 | if err != nil { |
| 215 | out.Close() |
| 216 | return nil, fmt.Errorf("stderr listener for %s: %w", rt.Host, err) |
| 217 | } |
| 218 | fail := func(err error) (CmdUnit, error) { |
| 219 | out.Close() |
| 220 | errs.Close() |
| 221 | return nil, err |
| 222 | } |
| 223 | argv := append([]string{b.liveLink(rt.Host)}, spec.Args...) |
| 224 | env := childEnv(b.cmdVars(rt, spec.genEnv, p.dbPath, spec.RunID, spec.RunDir)) |
| 225 | if err := b.writeLaunch(p, instancectl.RoleCmd, rt.Host, argv, env); err != nil { |
| 226 | return fail(err) |
| 227 | } |
| 228 | var inv string |
| 229 | if exeStart != nil { |
| 230 | inv, err = exeStart.CmdExe(ctx, rt.Host, b.tier(rt).MemoryMax, *spec.serving) |
| 231 | } else { |
| 232 | inv, err = b.Control.Cmd(ctx, rt.Host, b.tier(rt).MemoryMax) |
| 233 | } |
| 234 | if err != nil { |
| 235 | // The request may have reached systemd before it failed (a timeout, |
| 236 | // a lost response, a 502 mid-roll), so the unit may be running. |
| 237 | // Hand back a unit whose StopAndSettle stops the host's cmd role |
| 238 | // and waits for its cgroup; it needs no invocation. |
| 239 | return &cmdUnit{b: b, rt: rt, out: out, errs: errs}, fmt.Errorf("%s: start cmd unit: %w", rt.Host, err) |
| 240 | } |
| 241 | log.Printf("activator: %s: started %s for run %s", rt.Host, instancectl.UnitName(instancectl.RoleCmd, rt.Host), spec.RunID) |
| 242 | return &cmdUnit{b: b, rt: rt, inv: inv, out: out, errs: errs}, nil |
| 243 | } |
| 244 | |
| 245 | type cmdUnit struct { |
| 246 | b *ExecBackend |
| 247 | rt registry.Route |
| 248 | inv string |
| 249 | out, errs *cmdStream |
| 250 | mu sync.Mutex // serialises StopAndSettle's attempts |
| 251 | proven bool // an attempt succeeded; under mu |
| 252 | } |
| 253 | |
| 254 | func (u *cmdUnit) Invocation() string { return u.inv } |
| 255 | |
| 256 | func (u *cmdUnit) Poll(ctx context.Context) (instancectl.CmdStatus, error) { |
| 257 | return u.b.Control.CmdStatus(ctx, u.rt.Host, u.inv) |
| 258 | } |
| 259 | |
| 260 | // StopAndSettle is one attempt at the proof: the cmd role stopped, then |
| 261 | // its cgroup waited empty, then both streams read to their end. The |
| 262 | // streams are drained only once the unit is proven gone, so a unit that |
| 263 | // outlives a failed attempt keeps its output flowing to the next one. |
| 264 | func (u *cmdUnit) StopAndSettle(ctx context.Context) error { |
| 265 | u.mu.Lock() |
| 266 | defer u.mu.Unlock() |
| 267 | if u.proven { |
| 268 | return nil |
| 269 | } |
| 270 | sctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), instancectl.DeadlineControl) |
| 271 | stopErr := u.b.Control.Stop(sctx, u.rt.Host, instancectl.RoleCmd) |
| 272 | cancel() |
| 273 | wctx, wcancel := context.WithTimeout(context.WithoutCancel(ctx), actStopTimeout) |
| 274 | emptyErr := u.b.waitRoleCgroupEmpty(wctx, u.rt, instancectl.RoleCmd) |
| 275 | wcancel() |
| 276 | if err := errors.Join(stopErr, emptyErr); err != nil { |
| 277 | return err |
| 278 | } |
| 279 | if err := errors.Join(u.out.Drain(cmdDrainBound), u.errs.Drain(cmdDrainBound)); err != nil { |
| 280 | return err |
| 281 | } |
| 282 | u.proven = true |
| 283 | return nil |
| 284 | } |
| 285 | |
| 286 | // cmdStream is the agent's end of one cmd stream (run spec §5.5): the |
| 287 | // unit's StandardOutput=file: or StandardError=file: connects once, at |
| 288 | // exec, and every byte is copied to w unchanged until EOF. Nothing goes to |
| 289 | // the journal or the app log. One connection is taken; a second is closed |
| 290 | // unread. |
| 291 | type cmdStream struct { |
| 292 | ln net.Listener |
| 293 | w io.Writer |
| 294 | accepted atomic.Bool |
| 295 | closed atomic.Bool |
| 296 | mu sync.Mutex |
| 297 | conn net.Conn |
| 298 | copied chan struct{} |
| 299 | looped chan struct{} // closed when the accept loop has returned |
| 300 | wg sync.WaitGroup |
| 301 | } |
| 302 | |
| 303 | func listenCmdStream(path string, w io.Writer) (*cmdStream, error) { |
| 304 | if w == nil { |
| 305 | w = io.Discard |
| 306 | } |
| 307 | ln, err := edge.SecureUnixListen(path) |
| 308 | if err != nil { |
| 309 | return nil, err |
| 310 | } |
| 311 | s := &cmdStream{ln: ln, w: w, copied: make(chan struct{}), looped: make(chan struct{})} |
| 312 | go s.accept() |
| 313 | return s, nil |
| 314 | } |
| 315 | |
| 316 | func (s *cmdStream) accept() { |
| 317 | defer close(s.looped) |
| 318 | for { |
| 319 | c, err := s.ln.Accept() |
| 320 | if err != nil { |
| 321 | return |
| 322 | } |
| 323 | s.take(c) |
| 324 | } |
| 325 | } |
| 326 | |
| 327 | // take starts copying c if it is the stream's first connection, and closes |
| 328 | // it unread otherwise. |
| 329 | func (s *cmdStream) take(c net.Conn) { |
| 330 | if s.accepted.Swap(true) { |
| 331 | c.Close() |
| 332 | return |
| 333 | } |
| 334 | s.mu.Lock() |
| 335 | if s.closed.Load() { |
| 336 | s.mu.Unlock() |
| 337 | c.Close() |
| 338 | return |
| 339 | } |
| 340 | s.conn = c |
| 341 | s.wg.Add(1) |
| 342 | s.mu.Unlock() |
| 343 | go s.copy(c) |
| 344 | } |
| 345 | |
| 346 | func (s *cmdStream) copy(c net.Conn) { |
| 347 | defer s.wg.Done() |
| 348 | defer close(s.copied) |
| 349 | defer c.Close() |
| 350 | _, _ = io.Copy(s.w, c) |
| 351 | } |
| 352 | |
| 353 | // cmdAcceptGrace is how long Drain waits for a connection still queued in |
| 354 | // the listener's backlog. A var so tests can shrink it. |
| 355 | var cmdAcceptGrace = time.Second |
| 356 | |
| 357 | // Drain waits up to bound for the stream to reach EOF (its unit has |
| 358 | // stopped, so every writer has closed), then closes the listener. The unit |
| 359 | // connected at exec, so the connection may still be queued in the backlog |
| 360 | // when the accept loop has not yet run: Drain stops the loop and takes any |
| 361 | // queued connection itself, so the command's last bytes are never dropped. |
| 362 | // A stream that never connected has nothing to drain. |
| 363 | func (s *cmdStream) Drain(bound time.Duration) error { |
| 364 | if dl, ok := s.ln.(interface{ SetDeadline(time.Time) error }); ok { |
| 365 | _ = dl.SetDeadline(time.Now()) |
| 366 | <-s.looped |
| 367 | if !s.accepted.Load() { |
| 368 | _ = dl.SetDeadline(time.Now().Add(cmdAcceptGrace)) |
| 369 | if c, err := s.ln.Accept(); err == nil { |
| 370 | s.take(c) |
| 371 | } |
| 372 | } |
| 373 | } |
| 374 | var err error |
| 375 | if s.accepted.Load() { |
| 376 | select { |
| 377 | case <-s.copied: |
| 378 | case <-time.After(bound): |
| 379 | err = fmt.Errorf("a cmd stream did not reach its end within %s", bound) |
| 380 | } |
| 381 | } |
| 382 | s.Close() |
| 383 | return err |
| 384 | } |
| 385 | |
| 386 | func (s *cmdStream) Close() error { |
| 387 | if s.closed.Swap(true) { |
| 388 | return nil |
| 389 | } |
| 390 | err := s.ln.Close() |
| 391 | s.mu.Lock() |
| 392 | if s.conn != nil { |
| 393 | s.conn.Close() |
| 394 | } |
| 395 | s.mu.Unlock() |
| 396 | <-s.looped |
| 397 | s.wg.Wait() |
| 398 | return err |
| 399 | } |
| 400 | |
amadan