| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "crypto/rand" |
| 6 | "encoding/hex" |
| 7 | "errors" |
| 8 | "fmt" |
| 9 | "log" |
| 10 | "os" |
| 11 | "sync" |
| 12 | "sync/atomic" |
| 13 | "time" |
| 14 | |
| 15 | "github.com/carlosframework/platform/internal/activator" |
| 16 | "github.com/carlosframework/platform/internal/provision" |
| 17 | "github.com/carlosframework/platform/internal/registry" |
| 18 | "github.com/carlosframework/platform/internal/release" |
| 19 | "github.com/carlosframework/platform/internal/runs" |
| 20 | ) |
| 21 | |
| 22 | // runLauncher is what the runs worker needs from the activator. |
| 23 | type runLauncher interface { |
| 24 | RunCmd(ctx context.Context, rt registry.Route, spec activator.CmdSpec) (activator.CmdHandle, error) |
| 25 | } |
| 26 | |
| 27 | var _ runLauncher = (*activator.Service)(nil) |
| 28 | |
| 29 | // runHandle is the worker's line to one executing run. |
| 30 | type runHandle struct { |
| 31 | cancel chan struct{} |
| 32 | once sync.Once |
| 33 | } |
| 34 | |
| 35 | func newRunHandle() *runHandle { return &runHandle{cancel: make(chan struct{})} } |
| 36 | |
| 37 | func (h *runHandle) requestCancel() { h.once.Do(func() { close(h.cancel) }) } |
| 38 | |
| 39 | // runsWorker is the box's half of carlos run (run spec §5.2). boot is this |
| 40 | // agent process's own id: a run picked under another boot of this box lost |
| 41 | // its unit when the edge stopped (BindsTo=carlos-edge.service). |
| 42 | type runsWorker struct { |
| 43 | store release.Store |
| 44 | routes func(host string) (*registry.Route, error) |
| 45 | launch runLauncher |
| 46 | dataDir string |
| 47 | box string |
| 48 | boot string |
| 49 | now func() time.Time |
| 50 | exec func(ctx context.Context, rec provision.Record, req runs.Request, rt registry.Route, h *runHandle) |
| 51 | // events is where a run's start and end reach `carlos logs` |
| 52 | // (runevents.go); nil on a box with no log store. |
| 53 | events activator.EventSink |
| 54 | // start runs a picked run's exec: on its own goroutine in the agent, |
| 55 | // inline in tests that count launches. |
| 56 | start func(func()) |
| 57 | // booted is set when the boot pass has finished. The boot pass deletes |
| 58 | // every .carlos-run directory, so until then pass does nothing: a run |
| 59 | // it launched would have its directory swept from under it. |
| 60 | booted atomic.Bool |
| 61 | |
| 62 | mu sync.Mutex |
| 63 | noted map[string]provision.Record // host -> the record reconcile read |
| 64 | active map[string]*runHandle // run id -> the run this agent is executing |
| 65 | unfit map[string]bool // run ids whose request.json was refused, logged once |
| 66 | } |
| 67 | |
| 68 | func newRunsWorker(store release.Store, routes func(string) (*registry.Route, error), launch runLauncher, dataDir, box, boot string) *runsWorker { |
| 69 | w := &runsWorker{store: store, routes: routes, launch: launch, dataDir: dataDir, box: box, boot: boot, |
| 70 | now: time.Now, noted: map[string]provision.Record{}, active: map[string]*runHandle{}, unfit: map[string]bool{}, |
| 71 | start: func(f func()) { go f() }} |
| 72 | w.exec = w.execute |
| 73 | return w |
| 74 | } |
| 75 | |
| 76 | // newBootID is a fresh id for this agent process. |
| 77 | func newBootID() string { |
| 78 | var b [16]byte |
| 79 | if _, err := rand.Read(b[:]); err != nil { |
| 80 | return time.Now().UTC().Format("20060102T150405.000000000") |
| 81 | } |
| 82 | return hex.EncodeToString(b[:]) |
| 83 | } |
| 84 | |
| 85 | // armedRuns is the worker the agent armed; nil until then (and on a box |
| 86 | // with no activator, which cannot run anything). |
| 87 | var armedRuns atomic.Pointer[runsWorker] |
| 88 | |
| 89 | // note is ReconcileConfig.RunSeen: remember, never act, never block. |
| 90 | func (w *runsWorker) note(rec provision.Record) { |
| 91 | w.mu.Lock() |
| 92 | w.noted[rec.Host] = rec |
| 93 | w.mu.Unlock() |
| 94 | } |
| 95 | |
| 96 | // pass acts on every record noted since the last pass, once the boot pass |
| 97 | // has finished; before that the notes wait. |
| 98 | func (w *runsWorker) pass(ctx context.Context) { |
| 99 | if !w.booted.Load() { |
| 100 | return |
| 101 | } |
| 102 | w.mu.Lock() |
| 103 | noted := w.noted |
| 104 | w.noted = map[string]provision.Record{} |
| 105 | w.mu.Unlock() |
| 106 | for _, rec := range noted { |
| 107 | w.consider(ctx, rec) |
| 108 | } |
| 109 | } |
| 110 | |
| 111 | // bootPass runs once at agent start (index item 2: reconcile's first pass |
| 112 | // is a minute away, and the CLI expires a run at 60 s). It deletes every |
| 113 | // .carlos-run directory first (run spec §5.8: no run is open on the box |
| 114 | // yet), then reads every record once and acts on each run. |
| 115 | func (w *runsWorker) bootPass(ctx context.Context) { |
| 116 | // Opens pass, whether or not the records could be read: runs the boot |
| 117 | // pass did not see are the next passes' to pick. |
| 118 | defer w.booted.Store(true) |
| 119 | if err := runs.SweepRunDirs(w.dataDir); err != nil { |
| 120 | log.Printf("agent: runs: boot sweep of .carlos-run: %v", err) |
| 121 | } |
| 122 | recs, err := provision.ListAllRecords(ctx, w.store) |
| 123 | if err != nil { |
| 124 | log.Printf("agent: runs: boot pass: %v", err) |
| 125 | return |
| 126 | } |
| 127 | for _, rec := range recs { |
| 128 | if rec.ActiveRun != nil { |
| 129 | w.consider(ctx, rec) |
| 130 | } |
| 131 | } |
| 132 | } |
| 133 | |
| 134 | // owned is the row this box runs the record's instance from, when it is |
| 135 | // this box's to act on: provisioned for this record's account and app, |
| 136 | // and one the activator manages. |
| 137 | func (w *runsWorker) owned(rec provision.Record) (registry.Route, bool) { |
| 138 | rt, err := w.routes(rec.Host) |
| 139 | if err != nil || rt == nil || !rt.Provisioned || rt.Account != rec.Account || rt.App != rec.App || !activator.ActivatorManaged(*rt) { |
| 140 | return registry.Route{}, false |
| 141 | } |
| 142 | return *rt, true |
| 143 | } |
| 144 | |
| 145 | // consider acts on one record's run (run spec §5.2). |
| 146 | func (w *runsWorker) consider(ctx context.Context, rec provision.Record) { |
| 147 | r := rec.ActiveRun |
| 148 | if r == nil { |
| 149 | return |
| 150 | } |
| 151 | switch r.State { |
| 152 | case provision.RunQueued: |
| 153 | rt, ok := w.owned(rec) |
| 154 | if !ok || w.now().After(r.Deadline) { |
| 155 | return |
| 156 | } |
| 157 | w.pickup(ctx, rec, rt) |
| 158 | case provision.RunPicked, provision.RunCancel: |
| 159 | if r.Box != w.box { |
| 160 | return |
| 161 | } |
| 162 | w.mu.Lock() |
| 163 | h := w.active[r.ID] |
| 164 | w.mu.Unlock() |
| 165 | switch { |
| 166 | case h != nil && r.State == provision.RunCancel: |
| 167 | h.requestCancel() |
| 168 | case h == nil && r.BootID != w.boot: |
| 169 | w.closeInterrupted(ctx, rec) |
| 170 | case h == nil: |
| 171 | w.finishStranded(ctx, rec) |
| 172 | } |
| 173 | } |
| 174 | } |
| 175 | |
| 176 | // finishStranded is the done CAS again for a run this box picked under this |
| 177 | // boot and is no longer executing, whose result landed but whose done CAS |
| 178 | // did not within runFinishBound. Left picked, it would hold the instance |
| 179 | // until the sweep's deadline plus an hour. A run with no result is left |
| 180 | // alone: its result could not be written, and the sweep ends it as lost. |
| 181 | func (w *runsWorker) finishStranded(ctx context.Context, rec provision.Record) { |
| 182 | id := rec.ActiveRun.ID |
| 183 | rctx, cancel := context.WithTimeout(ctx, runReadBound) |
| 184 | defer cancel() |
| 185 | res, err := runs.ReadResult(rctx, w.store, rec.Account, rec.App, id) |
| 186 | if err != nil || res == nil { |
| 187 | return |
| 188 | } |
| 189 | if _, err := provision.SetRecordRun(rctx, w.store, rec.Account, rec.App, rec.Host, runFinishOurs(id, w.box, w.boot)); err != nil { |
| 190 | if !errors.Is(err, provision.ErrRunConflict) && !errors.Is(err, os.ErrNotExist) { |
| 191 | log.Printf("agent: runs: %s %s: done, again: %v", rec.Host, id, err) |
| 192 | } |
| 193 | return |
| 194 | } |
| 195 | log.Printf("agent: runs: %s: run %s had its result but was still picked; marked done", rec.Host, id) |
| 196 | } |
| 197 | |
| 198 | // pickup takes a queued run with one CAS from queued to picked. Only the |
| 199 | // box whose CAS lands launches; any other outcome (another box, a cancel, |
| 200 | // an expiry) leaves the run alone. A request.json that is not the run the |
| 201 | // record names, or whose timeout is out of range, is never picked: the |
| 202 | // expiry ends it. |
| 203 | func (w *runsWorker) pickup(ctx context.Context, rec provision.Record, rt registry.Route) { |
| 204 | id := rec.ActiveRun.ID |
| 205 | req, err := runs.ReadRequest(ctx, w.store, rec.Account, rec.App, id) |
| 206 | if err != nil { |
| 207 | // The console writes request.json before the record names the run; |
| 208 | // without it there is nothing to run, and the expiry ends it. |
| 209 | log.Printf("agent: runs: %s %s: request: %v", rec.Host, id, err) |
| 210 | return |
| 211 | } |
| 212 | if err := requestFits(rec, id, *req); err != nil { |
| 213 | if w.firstUnfit(id) { |
| 214 | log.Printf("agent: runs: %s %s: not picked: %v", rec.Host, id, err) |
| 215 | } |
| 216 | return |
| 217 | } |
| 218 | h := newRunHandle() |
| 219 | w.mu.Lock() |
| 220 | if w.active[id] != nil { |
| 221 | w.mu.Unlock() |
| 222 | return |
| 223 | } |
| 224 | w.active[id] = h |
| 225 | w.mu.Unlock() |
| 226 | got, err := provision.SetRecordRun(ctx, w.store, rec.Account, rec.App, rec.Host, |
| 227 | provision.RunPick(id, w.box, w.boot, w.now().Add(req.Timeout()))) |
| 228 | if err != nil { |
| 229 | if got = w.pickedAnyway(ctx, rec, id); got == nil { |
| 230 | w.forget(id) |
| 231 | if !errors.Is(err, provision.ErrRunConflict) && !errors.Is(err, os.ErrNotExist) { |
| 232 | log.Printf("agent: runs: %s %s: pickup: %v", rec.Host, id, err) |
| 233 | } |
| 234 | return |
| 235 | } |
| 236 | log.Printf("agent: runs: %s %s: pickup answered %v, but the record names the run as this box's", rec.Host, id, err) |
| 237 | } |
| 238 | log.Printf("agent: runs: %s: picked run %s", rec.Host, id) |
| 239 | w.start(func() { w.exec(context.Background(), *got, *req, rt, h) }) |
| 240 | } |
| 241 | |
| 242 | // unfitLogMax caps the ids firstUnfit remembers; past it the set starts |
| 243 | // again, and a refusal still queued is logged once more. |
| 244 | const unfitLogMax = 256 |
| 245 | |
| 246 | // firstUnfit reports whether this is the first refusal of id's request, |
| 247 | // so a run left for the expiry is logged once, not on every pass. |
| 248 | func (w *runsWorker) firstUnfit(id string) bool { |
| 249 | w.mu.Lock() |
| 250 | defer w.mu.Unlock() |
| 251 | if w.unfit[id] { |
| 252 | return false |
| 253 | } |
| 254 | if len(w.unfit) >= unfitLogMax { |
| 255 | clear(w.unfit) |
| 256 | } |
| 257 | w.unfit[id] = true |
| 258 | return true |
| 259 | } |
| 260 | |
| 261 | // requestFits says why request.json is not the run rec names, with a |
| 262 | // timeout the console would have accepted, or nil. |
| 263 | func requestFits(rec provision.Record, id string, req runs.Request) error { |
| 264 | if req.ID != id || req.Account != rec.Account || req.App != rec.App || req.Host != rec.Host { |
| 265 | return fmt.Errorf("request.json names %s/%s/%s %s, not the record's run", req.Account, req.App, req.Host, req.ID) |
| 266 | } |
| 267 | return runs.CheckTimeout(req.Timeout()) |
| 268 | } |
| 269 | |
| 270 | // pickedAnyway reads the record back after a pickup CAS that answered an |
| 271 | // error: a write whose answer was lost, or a conditional write that landed |
| 272 | // and then answered a conflict, may have picked the run all the same. The |
| 273 | // record is what the box launches from when it names the run as picked by |
| 274 | // this box under this boot; nil otherwise. |
| 275 | func (w *runsWorker) pickedAnyway(ctx context.Context, rec provision.Record, id string) *provision.Record { |
| 276 | for range 3 { |
| 277 | rctx, cancel := context.WithTimeout(ctx, runReadBound) |
| 278 | cur, err := provision.GetRecord(rctx, w.store, rec.Account, rec.App, rec.Host) |
| 279 | cancel() |
| 280 | if errors.Is(err, os.ErrNotExist) { |
| 281 | return nil |
| 282 | } |
| 283 | if err == nil { |
| 284 | if r := cur.ActiveRun; r != nil && r.ID == id && r.Box == w.box && r.BootID == w.boot && |
| 285 | (r.State == provision.RunPicked || r.State == provision.RunCancel) { |
| 286 | return cur |
| 287 | } |
| 288 | return nil |
| 289 | } |
| 290 | log.Printf("agent: runs: %s %s: read back the pickup: %v", rec.Host, id, err) |
| 291 | select { |
| 292 | case <-ctx.Done(): |
| 293 | return nil |
| 294 | case <-time.After(runPollRetry): |
| 295 | } |
| 296 | } |
| 297 | return nil |
| 298 | } |
| 299 | |
| 300 | func (w *runsWorker) forget(id string) { |
| 301 | w.mu.Lock() |
| 302 | delete(w.active, id) |
| 303 | w.mu.Unlock() |
| 304 | } |
| 305 | |
| 306 | // closeInterrupted ends a run this box picked under an earlier boot (run |
| 307 | // spec §5.2): its unit went with the edge, so it is never executed again. |
| 308 | // The result names the last chunk the old boot wrote. |
| 309 | func (w *runsWorker) closeInterrupted(ctx context.Context, rec provision.Record) { |
| 310 | id := rec.ActiveRun.ID |
| 311 | last, err := runs.LastChunkSeq(ctx, w.store, rec.Account, rec.App, id) |
| 312 | if err != nil { |
| 313 | log.Printf("agent: runs: %s %s: list chunks: %v", rec.Host, id, err) |
| 314 | return |
| 315 | } |
| 316 | res := runs.Result{Interrupted: runs.InterruptedAgentRestart, Box: w.box, Ended: w.now().UTC(), LastSeq: last} |
| 317 | // A run's log lines are at most once and best effort (amadan #325), like |
| 318 | // the rest of the platform stream; result.json and the audit log are its |
| 319 | // record. So the ending is logged only by a pass that wrote the run's |
| 320 | // result itself (read before, read back after) AND whose own done CAS |
| 321 | // landed. Every pass that still holds a picked snapshot comes back here: |
| 322 | // the create-only write answers a second time as if it had landed, a |
| 323 | // result already there is the boot's that ran the command (which logged |
| 324 | // its own ending), and a stale pass after the run was acknowledged |
| 325 | // recreates a result but its CAS fails. A lost answer anywhere loses the |
| 326 | // line rather than risk a second one. |
| 327 | prior, err := runs.ReadResult(ctx, w.store, rec.Account, rec.App, id) |
| 328 | if err != nil { |
| 329 | log.Printf("agent: runs: %s %s: read the result: %v", rec.Host, id, err) |
| 330 | return |
| 331 | } |
| 332 | ours := false |
| 333 | if prior == nil { |
| 334 | werr := runs.WriteResult(ctx, w.store, rec.Account, rec.App, id, res) |
| 335 | got, err := runs.ReadResult(ctx, w.store, rec.Account, rec.App, id) |
| 336 | if err != nil || got == nil { |
| 337 | log.Printf("agent: runs: %s %s: result: %v", rec.Host, id, errors.Join(werr, err)) |
| 338 | return |
| 339 | } |
| 340 | ours = got.Box == res.Box && got.Ended.Equal(res.Ended) && got.Interrupted == res.Interrupted |
| 341 | } |
| 342 | if _, err := provision.SetRecordRun(ctx, w.store, rec.Account, rec.App, rec.Host, provision.RunFinish(id)); err == nil { |
| 343 | if ours { |
| 344 | w.runEvent(rec, runEndedLine(id, res, true)) |
| 345 | } |
| 346 | } else if !errors.Is(err, provision.ErrRunConflict) && !errors.Is(err, os.ErrNotExist) { |
| 347 | log.Printf("agent: runs: %s %s: done: %v", rec.Host, id, err) |
| 348 | } |
| 349 | if err := runs.RemoveRunDir(provision.InstanceDir(w.dataDir, rec.Host), id); err != nil && !errors.Is(err, os.ErrNotExist) { |
| 350 | log.Printf("agent: runs: %s %s: remove run dir: %v", rec.Host, id, err) |
| 351 | } |
| 352 | log.Printf("agent: runs: %s: run %s was picked before this agent started; ended as interrupted", rec.Host, id) |
| 353 | } |
| 354 | |
| 355 | // armRunsPass arms the runs worker on a box with an activator. Thin, like |
| 356 | // armMovesPass: the behaviour is the worker's, tested there. events must be |
| 357 | // a genuinely nil interface when there is no log store, never a nil |
| 358 | // *applog.Sink (the trap agent.go documents). |
| 359 | func armRunsPass(svc *activator.Service, reg *registry.Registry, dataDir string, events activator.EventSink) { |
| 360 | if svc == nil || reg == nil { |
| 361 | return |
| 362 | } |
| 363 | store, err := openDeployStore(context.Background()) |
| 364 | if err != nil { |
| 365 | log.Printf("agent: runs disabled: %v", err) |
| 366 | return |
| 367 | } |
| 368 | w := newRunsWorker(store, reg.Get, svc, dataDir, BoxLabel(), newBootID()) |
| 369 | w.events = events |
| 370 | armedRuns.Store(w) |
| 371 | log.Printf("agent: runs armed (box %s, boot %s)", w.box, w.boot) |
| 372 | } |
| 373 | |
| 374 | // withRunsHook gives reconcile the worker's note (run spec §5.2). |
| 375 | func withRunsHook(cfg provision.ReconcileConfig) provision.ReconcileConfig { |
| 376 | if w := armedRuns.Load(); w != nil && cfg.RunSeen == nil { |
| 377 | cfg.RunSeen = w.note |
| 378 | } |
| 379 | return cfg |
| 380 | } |
| 381 | |
| 382 | // runRunsPass acts on what the last reconcile pass noted. |
| 383 | func runRunsPass(ctx context.Context) { |
| 384 | if w := armedRuns.Load(); w != nil { |
| 385 | w.pass(ctx) |
| 386 | } |
| 387 | } |
| 388 | |
| 389 | // runRunsBootPass is the worker's one pass at agent start. |
| 390 | func runRunsBootPass(ctx context.Context) { |
| 391 | if w := armedRuns.Load(); w != nil { |
| 392 | w.bootPass(ctx) |
| 393 | } |
| 394 | } |
| 395 | |
amadan