| 1 | package activator |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "fmt" |
| 8 | "log" |
| 9 | "os" |
| 10 | "path/filepath" |
| 11 | "sync/atomic" |
| 12 | "time" |
| 13 | |
| 14 | "github.com/carlosframework/platform/internal/edge" |
| 15 | "github.com/carlosframework/platform/internal/instancectl" |
| 16 | "github.com/carlosframework/platform/internal/instancefs" |
| 17 | "github.com/carlosframework/platform/internal/instancekey" |
| 18 | "github.com/carlosframework/platform/internal/registry" |
| 19 | ) |
| 20 | |
| 21 | // ErrStillRunning is Stop's answer when ctx expired before termination was |
| 22 | // proven (design §6.3). The activator then keeps the lease and the entry |
| 23 | // in stopping, and the sweeper retries; the lease is never released over a |
| 24 | // writer that may still exist. |
| 25 | var ErrStillRunning = errors.New("activator: stop unproven; processes remain") |
| 26 | |
| 27 | // unitProc is a nested host's Proc (design §6.3): the unit the control |
| 28 | // service started, the listeners the agent holds for it, and the observer |
| 29 | // that proves termination. Named apart from backend_unit.go's svcUnitProc, |
| 30 | // a whole-app tenant's own systemd unit: this one is a nested exec |
| 31 | // instance the activator spawns through the root instance-control channel. |
| 32 | type unitProc struct { |
| 33 | b *ExecBackend |
| 34 | rt registry.Route |
| 35 | role instancectl.Role |
| 36 | unit, slice string |
| 37 | invocation string |
| 38 | runDir string |
| 39 | // gen is the generation of a carlos-instance-gen@ unit, 0 for the |
| 40 | // original templates; dataDir holds its launch file. |
| 41 | gen uint64 |
| 42 | dataDir string |
| 43 | ctl *edge.ControlListener |
| 44 | logs *logListener |
| 45 | exited atomic.Bool |
| 46 | done chan struct{} |
| 47 | closeOnce atomic.Bool |
| 48 | // listenersArmed guards the done-watcher goroutine Stop's ctx.Done |
| 49 | // branch starts: several callers bound their wait on Stop shorter than |
| 50 | // the unit may actually take to die (a sidecar co-stop, a discarded |
| 51 | // sidecar, a failed health wait) and never call Stop again once it |
| 52 | // returns ErrStillRunning, which otherwise leaves ctl and logs open |
| 53 | // for as long as the agent runs. Armed at most once per unitProc. |
| 54 | listenersArmed atomic.Bool |
| 55 | // stopRequested is set before the FIRST Control.Stop this proc's owner |
| 56 | // ever issues for it: once true, the observer stops treating a |
| 57 | // differently-invoked or not-yet-invoked unit as a FOREIGN restart to |
| 58 | // stop on its own. It is instead the proof that our own invocation is |
| 59 | // already over (see observe). |
| 60 | stopRequested atomic.Bool |
| 61 | // genFilesGone is set once a stop of this generation succeeded: the |
| 62 | // control service removed the unit's drop-in and exe layout with it, |
| 63 | // so finaliseGen need not ask again. |
| 64 | genFilesGone atomic.Bool |
| 65 | // launchEnv is the file environment (the config bundle with the |
| 66 | // instance's own file on top) a run role's unit launched with, before |
| 67 | // the agent-owned variables. It is written once, before the handle |
| 68 | // leaves spawnUnitFrom, and never again. A run against this generation |
| 69 | // composes its environment from it (LaunchEnv; merge ruling M3). nil |
| 70 | // for every other role. |
| 71 | launchEnv map[string]string |
| 72 | // launchKeyLayer is the layer launchEnv's CARLOS_INSTANCE_KEY came |
| 73 | // from, decided by the read that composed launchEnv (instance-key |
| 74 | // ruling 29). Written with launchEnv, never again. |
| 75 | launchKeyLayer instancekey.Layer |
| 76 | } |
| 77 | |
| 78 | // launchEnvProc is a generation handle that kept the file environment its |
| 79 | // generation launched with (merge ruling M3). |
| 80 | type launchEnvProc interface { |
| 81 | LaunchEnv() map[string]string |
| 82 | } |
| 83 | |
| 84 | // LaunchEnv is the file environment this generation launched with, or nil |
| 85 | // when it kept none. The map is the handle's own: callers must not write |
| 86 | // it. |
| 87 | func (p *unitProc) LaunchEnv() map[string]string { return p.launchEnv } |
| 88 | |
| 89 | // launchKeyLayerProc is a generation handle that also kept where its |
| 90 | // launch environment's instance key came from. |
| 91 | type launchKeyLayerProc interface { |
| 92 | LaunchKeyLayer() instancekey.Layer |
| 93 | } |
| 94 | |
| 95 | // LaunchKeyLayer is the layer LaunchEnv's CARLOS_INSTANCE_KEY came from. |
| 96 | func (p *unitProc) LaunchKeyLayer() instancekey.Layer { return p.launchKeyLayer } |
| 97 | |
| 98 | func (b *ExecBackend) newUnitProc(rt registry.Route, role instancectl.Role, p instancePaths, inv string, ctl *edge.ControlListener, logs *logListener) *unitProc { |
| 99 | return b.newUnitProcFor(rt, role, 0, p, inv, ctl, logs) |
| 100 | } |
| 101 | |
| 102 | func (b *ExecBackend) newUnitProcFor(rt registry.Route, role instancectl.Role, gen uint64, p instancePaths, inv string, ctl *edge.ControlListener, logs *logListener) *unitProc { |
| 103 | up := &unitProc{b: b, rt: rt, role: role, gen: gen, unit: unitName(role, rt.Host, gen), slice: instancectl.SliceName(rt.Host), |
| 104 | invocation: inv, runDir: p.runDir, dataDir: p.dataDir, ctl: ctl, logs: logs, done: make(chan struct{})} |
| 105 | go b.observe(up) |
| 106 | return up |
| 107 | } |
| 108 | |
| 109 | // unitName is the unit a role of host runs as: the original template's, or |
| 110 | // generation gen's (zero-downtime spec §7.6). |
| 111 | func unitName(role instancectl.Role, host string, gen uint64) string { |
| 112 | if gen == 0 { |
| 113 | return instancectl.UnitName(role, host) |
| 114 | } |
| 115 | return instancectl.GenUnitName(host, gen) |
| 116 | } |
| 117 | |
| 118 | // stopRole stops one role's unit: the original template's through Control, |
| 119 | // a generation's through GenControl. A generation stop never touches a |
| 120 | // sibling generation. |
| 121 | func (b *ExecBackend) stopRole(ctx context.Context, host string, role instancectl.Role, gen uint64) error { |
| 122 | if gen == 0 { |
| 123 | return b.Control.Stop(ctx, host, role) |
| 124 | } |
| 125 | gc, err := b.genControl() |
| 126 | if err != nil { |
| 127 | return err |
| 128 | } |
| 129 | return gc.StopGen(ctx, host, gen, false) |
| 130 | } |
| 131 | |
| 132 | func (p *unitProc) Exited() bool { return p.exited.Load() } |
| 133 | |
| 134 | // Stop asks the control service to stop the unit and waits for termination |
| 135 | // to be proven. The control listener and the log listener are closed only |
| 136 | // after that, preserving childProc's ordering (signal first, close the |
| 137 | // listener after the process is gone). |
| 138 | func (p *unitProc) Stop(ctx context.Context) error { |
| 139 | if !p.Exited() { |
| 140 | // Set before the Control.Stop call, not after: observe polls |
| 141 | // concurrently, and a stop it cannot yet see as "requested" would |
| 142 | // still fire its own foreign-invocation stop for the same unit. |
| 143 | p.stopRequested.Store(true) |
| 144 | sctx, cancel := context.WithTimeout(ctx, instancectl.DeadlineControl) |
| 145 | err := p.stopUnit(sctx) |
| 146 | cancel() |
| 147 | if err != nil { |
| 148 | log.Printf("activator: stop %s %s: %v", p.rt.Host, p.role, err) |
| 149 | } |
| 150 | } |
| 151 | select { |
| 152 | case <-p.done: |
| 153 | p.closeListeners() |
| 154 | return nil |
| 155 | case <-ctx.Done(): |
| 156 | // The caller is giving up on waiting, not on the stop itself: the |
| 157 | // unit may still be mid-drain, so the listeners must stay open while |
| 158 | // a writer could still exist. But several callers never call Stop |
| 159 | // again, and closing them still has to happen once done eventually |
| 160 | // closes, or they leak for the life of the agent. |
| 161 | // One goroutine per unitProc, armed once, does that regardless of |
| 162 | // whether anyone ever retries. |
| 163 | if !p.listenersArmed.Swap(true) { |
| 164 | go func() { |
| 165 | <-p.done |
| 166 | p.closeListeners() |
| 167 | }() |
| 168 | } |
| 169 | return ErrStillRunning |
| 170 | } |
| 171 | } |
| 172 | |
| 173 | // stopUnit asks the control service to stop this proc's unit. For a |
| 174 | // generation, a stop that succeeded also removed the unit's files there. |
| 175 | func (p *unitProc) stopUnit(ctx context.Context) error { |
| 176 | err := p.b.stopRole(ctx, p.rt.Host, p.role, p.gen) |
| 177 | if err == nil && p.gen != 0 { |
| 178 | p.genFilesGone.Store(true) |
| 179 | } |
| 180 | return err |
| 181 | } |
| 182 | |
| 183 | // finaliseGen removes a generation's files once observe has proven it |
| 184 | // gone (generation numbers are never reused, so nothing else would): its |
| 185 | // launch file here, and the control service's drop-in and exe layout |
| 186 | // through a stop of the inactive unit, unless a stop of ours already |
| 187 | // removed them. Never called while termination is unproven. |
| 188 | func (p *unitProc) finaliseGen() { |
| 189 | if p.gen == 0 { |
| 190 | return |
| 191 | } |
| 192 | p.removeLaunch() |
| 193 | if p.genFilesGone.Load() { |
| 194 | return |
| 195 | } |
| 196 | ctx, cancel := context.WithTimeout(context.Background(), instancectl.DeadlineControl) |
| 197 | defer cancel() |
| 198 | if err := p.stopUnit(ctx); err != nil { |
| 199 | log.Printf("activator: %s: remove the files of %s: %v", p.rt.Host, p.unit, err) |
| 200 | } |
| 201 | } |
| 202 | |
| 203 | func (p *unitProc) closeListeners() { |
| 204 | if p.closeOnce.Swap(true) { |
| 205 | return |
| 206 | } |
| 207 | if p.ctl != nil { |
| 208 | p.ctl.Close() |
| 209 | } |
| 210 | if p.logs != nil { |
| 211 | p.logs.Close() |
| 212 | } |
| 213 | } |
| 214 | |
| 215 | // observe is the goroutine of design §6.5. It proves termination from two |
| 216 | // facts the tenant cannot forge: systemd's unit state and cgroup emptiness. |
| 217 | // The exit record decides the wording of the event line and nothing else. |
| 218 | func (b *ExecBackend) observe(p *unitProc) { |
| 219 | poll := b.ObservePoll |
| 220 | if poll <= 0 { |
| 221 | poll = time.Second |
| 222 | } |
| 223 | state := b.UnitState |
| 224 | if state == nil { |
| 225 | state = systemctlUnitState |
| 226 | } |
| 227 | events := b.CgroupEvents |
| 228 | if events == nil { |
| 229 | events = cgroupEventsUnder(b.cgroupRoot()) |
| 230 | } |
| 231 | foreignStopped := false |
| 232 | lastRemain := time.Time{} |
| 233 | lastEventsErr := time.Time{} |
| 234 | for { |
| 235 | ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) |
| 236 | active, inv, err := state(ctx, p.unit) |
| 237 | cancel() |
| 238 | if err != nil { |
| 239 | log.Printf("activator: %s: unit state: %v", p.rt.Host, err) |
| 240 | } |
| 241 | terminal := false |
| 242 | foreign := false |
| 243 | switch { |
| 244 | case err != nil: |
| 245 | case inv != "" && inv != p.invocation: |
| 246 | foreign = true |
| 247 | case active == "inactive" || active == "failed": |
| 248 | terminal = true |
| 249 | } |
| 250 | |
| 251 | // A stop of OUR OWN invocation is outstanding (Stop was already |
| 252 | // called): systemd starts a fresh invocation of the same unit only |
| 253 | // once the previous one's stop job has fully completed, so seeing |
| 254 | // anything that is not provably that same old invocation, a |
| 255 | // different id, or none reported yet while the unit sits in a |
| 256 | // non-terminal ActiveState (mid-cycle to a fresh one), is itself |
| 257 | // proof our invocation is over. No cgroup check is needed here: |
| 258 | // there is no "our process might still be running" left to |
| 259 | // disprove, because systemd's own state machine already disproved |
| 260 | // it by allowing a new invocation to begin. |
| 261 | if p.stopRequested.Load() && err == nil && !terminal && (foreign || inv == "") { |
| 262 | emitEvent(b.Logs, p.rt, "warn", "instance exited (status unknown: replaced after stop)") |
| 263 | p.exited.Store(true) |
| 264 | close(p.done) |
| 265 | return |
| 266 | } |
| 267 | |
| 268 | if foreign && !foreignStopped && !p.stopRequested.Load() { |
| 269 | log.Printf("activator: %s: unit restarted outside the activator (invocation %s, ours %s); stopping it", p.rt.Host, inv, p.invocation) |
| 270 | sctx, scancel := context.WithTimeout(context.Background(), instancectl.DeadlineControl) |
| 271 | if err := p.stopUnit(sctx); err != nil { |
| 272 | log.Printf("activator: %s: stop foreign invocation: %v", p.rt.Host, err) |
| 273 | } |
| 274 | scancel() |
| 275 | foreignStopped = true |
| 276 | } |
| 277 | if terminal || (foreign && foreignStopped) { |
| 278 | exists, populated, eerr := events(p.slice, p.unit) |
| 279 | if eerr != nil { |
| 280 | // A read error is not evidence of anything. Not exited, |
| 281 | // not still running, just unknown. Treating it as "cgroup |
| 282 | // empty" would let a still-running process's exit be |
| 283 | // reported anyway, exactly what this proof scheme exists |
| 284 | // to prevent (design §6.5); rate-limited the same way the |
| 285 | // "processes remain" line below is, since a persistent |
| 286 | // read failure would otherwise flood the log every poll. |
| 287 | if time.Since(lastEventsErr) > 2*time.Second { |
| 288 | log.Printf("activator: %s: cgroup.events: %v", p.rt.Host, eerr) |
| 289 | lastEventsErr = time.Now() |
| 290 | } |
| 291 | time.Sleep(poll) |
| 292 | continue |
| 293 | } |
| 294 | if !exists || !populated { |
| 295 | if terminal { |
| 296 | rec, ok := p.readRecord() |
| 297 | emitEvent(b.Logs, p.rt, exitLevel(rec, ok), exitLine(rec, ok)) |
| 298 | // The record is single-use (design §6.5): consumed |
| 299 | // exactly once into the member-facing line above, then |
| 300 | // cleared so it can never be misread as evidence for a |
| 301 | // LATER invocation of this same role (prepare only |
| 302 | // clears records left over from EARLIER invocations; |
| 303 | // this one is this invocation's own to remove). |
| 304 | p.removeRecord() |
| 305 | } else { |
| 306 | emitEvent(b.Logs, p.rt, "warn", "instance exited (status unknown: restarted outside the activator)") |
| 307 | } |
| 308 | p.finaliseGen() |
| 309 | p.exited.Store(true) |
| 310 | // Stop closes the listeners when it observes done; a crash |
| 311 | // nobody is stopping yet leaves them for the sweeper's |
| 312 | // stand-down, which calls Stop. |
| 313 | close(p.done) |
| 314 | return |
| 315 | } |
| 316 | if time.Since(lastRemain) > 2*time.Second { |
| 317 | log.Printf("activator: %s: processes remain after %s; not releasing", p.rt.Host, active) |
| 318 | lastRemain = time.Now() |
| 319 | } |
| 320 | } |
| 321 | time.Sleep(poll) |
| 322 | } |
| 323 | } |
| 324 | |
| 325 | // exitLine words the event line from the advisory record observe already |
| 326 | // read once and passes to both exitLine and exitLevel, so the two can |
| 327 | // never disagree from reading the file twice with something changing in |
| 328 | // between, the same way childProc does: a non-zero status is a warn line, |
| 329 | // a clean exit is info, and no record is unknown. The record is |
| 330 | // tenant-writable text (design §6.5), so every field that reaches the line |
| 331 | // passes through recordToken first. |
| 332 | func exitLine(rec instancectl.ExitRecord, ok bool) string { |
| 333 | switch { |
| 334 | case !ok: |
| 335 | return "instance exited (status unknown: no exit record)" |
| 336 | case rec.Result == "success" || (rec.Code == "exited" && rec.Status == "0"): |
| 337 | return "instance exited (exit 0)" |
| 338 | case rec.Code == "killed" || rec.Code == "dumped": |
| 339 | return "instance exited: signal " + recordToken(rec.Status) |
| 340 | default: |
| 341 | return fmt.Sprintf("instance exited: exit %s (%s)", recordToken(rec.Status), recordToken(rec.Result)) |
| 342 | } |
| 343 | } |
| 344 | |
| 345 | // recordToken bounds a field of the exit record to what systemd itself |
| 346 | // would write: at most 32 bytes of [A-Za-z0-9_-]. Anything else becomes |
| 347 | // "?" so a tenant cannot put arbitrary text or control characters into |
| 348 | // the platform's event stream. |
| 349 | func recordToken(s string) string { |
| 350 | if len(s) == 0 || len(s) > 32 { |
| 351 | return "?" |
| 352 | } |
| 353 | for _, c := range s { |
| 354 | if !(c >= 'a' && c <= 'z' || c >= 'A' && c <= 'Z' || c >= '0' && c <= '9' || c == '_' || c == '-') { |
| 355 | return "?" |
| 356 | } |
| 357 | } |
| 358 | return s |
| 359 | } |
| 360 | |
| 361 | // exitLevel is exitLine's companion, from the same read (see exitLine). |
| 362 | func exitLevel(rec instancectl.ExitRecord, ok bool) string { |
| 363 | if ok && (rec.Result == "success" || (rec.Code == "exited" && rec.Status == "0")) { |
| 364 | return "info" |
| 365 | } |
| 366 | return "warn" |
| 367 | } |
| 368 | |
| 369 | func (p *unitProc) recordName() string { |
| 370 | if p.gen == 0 { |
| 371 | return instancectl.ExitRecordName(p.role, p.invocation) |
| 372 | } |
| 373 | return instancectl.GenExitRecordName(p.gen, p.invocation) |
| 374 | } |
| 375 | |
| 376 | func (p *unitProc) readRecord() (instancectl.ExitRecord, bool) { |
| 377 | raw, err := instancefs.ReadBeneathLimit(p.runDir, p.recordName(), 4096) |
| 378 | if err != nil { |
| 379 | return instancectl.ExitRecord{}, false |
| 380 | } |
| 381 | var rec instancectl.ExitRecord |
| 382 | if err := json.Unmarshal(raw, &rec); err != nil { |
| 383 | return instancectl.ExitRecord{}, false |
| 384 | } |
| 385 | return rec, true |
| 386 | } |
| 387 | |
| 388 | // removeRecord deletes this invocation's own advisory exit record (design |
| 389 | // §6.5) after observe has consumed it into the member-facing event line. |
| 390 | // Best-effort: a record that is already gone, or a filesystem that refuses |
| 391 | // the remove, leaves nothing worse than what prepare's own stale-record |
| 392 | // sweep would have cleared on the NEXT invocation anyway. |
| 393 | func (p *unitProc) removeRecord() { |
| 394 | _ = os.Remove(filepath.Join(p.runDir, p.recordName())) |
| 395 | } |
| 396 | |
| 397 | // removeLaunch deletes a generation's launch file once the generation is |
| 398 | // proven gone. Generation numbers are never reused, so nothing else would. |
| 399 | // The original templates' launch file is rewritten by every start and is |
| 400 | // left alone. os.Remove never follows the final component. |
| 401 | func (p *unitProc) removeLaunch() { |
| 402 | if p.gen != 0 && p.dataDir != "" { |
| 403 | _ = os.Remove(filepath.Join(p.dataDir, instancectl.GenLaunchFileName(p.gen))) |
| 404 | } |
| 405 | } |
| 406 | |
| 407 | // cgroupRoot is CgroupRoot with the nil default applied. |
| 408 | func (b *ExecBackend) cgroupRoot() string { |
| 409 | if b.CgroupRoot != "" { |
| 410 | return b.CgroupRoot |
| 411 | } |
| 412 | return "/sys/fs/cgroup" |
| 413 | } |
| 414 | |
amadan