| 1 | // prepare.go is everything a nested role needs before its unit starts |
| 2 | // (isolated-exec design §6.1): the launch file's directories, the merged |
| 3 | // environment, the control listener (run only) and the role's log |
| 4 | // listener. It is the nested counterpart of what startChild/seedChild do |
| 5 | // inline for a flat host. |
| 6 | package activator |
| 7 | |
| 8 | import ( |
| 9 | "context" |
| 10 | "encoding/json" |
| 11 | "errors" |
| 12 | "fmt" |
| 13 | "log" |
| 14 | "net/http" |
| 15 | "os" |
| 16 | "path/filepath" |
| 17 | "strings" |
| 18 | "time" |
| 19 | |
| 20 | "github.com/carlosframework/platform/internal/edge" |
| 21 | "github.com/carlosframework/platform/internal/instancectl" |
| 22 | "github.com/carlosframework/platform/internal/instancefs" |
| 23 | "github.com/carlosframework/platform/internal/instancekey" |
| 24 | "github.com/carlosframework/platform/internal/instanceupdates" |
| 25 | "github.com/carlosframework/platform/internal/provision" |
| 26 | "github.com/carlosframework/platform/internal/registry" |
| 27 | "github.com/carlosframework/platform/internal/release" |
| 28 | "github.com/carlosframework/platform/internal/schedule" |
| 29 | ) |
| 30 | |
| 31 | // prepare is everything a nested role needs before its unit starts |
| 32 | // (design §6.1): the two directories, the admin token and merged env (for |
| 33 | // run and seed; a cmd's env is its generation's, and prepare returns none), the control listener (run only), the role's log listener, |
| 34 | // and stale records cleared. It returns the listeners so the caller can |
| 35 | // hand them to the proc or close them on a failed start. |
| 36 | func (b *ExecBackend) prepare(rt registry.Route, role instancectl.Role) (instancePaths, *edge.ControlListener, *logListener, map[string]string, error) { |
| 37 | return b.prepareAt(rt, role, rt.Socket, 0) |
| 38 | } |
| 39 | |
| 40 | // prepareAt is prepare for a run role whose app socket is sock: rt.Socket, |
| 41 | // or a generation socket (StartGen, zero-downtime spec §7.1). A generation |
| 42 | // socket's own directory, g<n> in the run dir, is made first (rev 19, rule |
| 43 | // 3). Only the file at sock is removed, never a sibling generation's, and |
| 44 | // the control listener binds sock + ".control". The seed and sidecar roles |
| 45 | // ignore sock. |
| 46 | // gen is the generation unit being prepared (0 for the original templates): |
| 47 | // it keys the log socket and the exit records cleared (zero-downtime spec |
| 48 | // §7.6). |
| 49 | func (b *ExecBackend) prepareAt(rt registry.Route, role instancectl.Role, sock string, gen uint64) (instancePaths, *edge.ControlListener, *logListener, map[string]string, error) { |
| 50 | p, ctl, logs, env, _, err := b.prepareKeyedAt(rt, role, sock, gen, nil) |
| 51 | return p, ctl, logs, env, err |
| 52 | } |
| 53 | |
| 54 | // prepareKeyedAt is prepareAt returning, with the env, the layer its |
| 55 | // CARLOS_INSTANCE_KEY came from, decided by the same read (instance-key |
| 56 | // ruling 29). The nested run spawn keeps both on the generation's handle, |
| 57 | // for a run's command against that generation (StartCmd). |
| 58 | // steps, when not nil, times prepare's parts for the caller's journal line. |
| 59 | func (b *ExecBackend) prepareKeyedAt(rt registry.Route, role instancectl.Role, sock string, gen uint64, steps *wakeSteps) (instancePaths, *edge.ControlListener, *logListener, map[string]string, instancekey.Layer, error) { |
| 60 | if steps == nil { |
| 61 | steps = &wakeSteps{} |
| 62 | } else { |
| 63 | steps.begin(time.Now()) |
| 64 | } |
| 65 | p := b.pathsFor(rt) |
| 66 | if err := os.MkdirAll(p.dataDir, 0o700); err != nil { |
| 67 | return p, nil, nil, nil, instancekey.LayerNone, err |
| 68 | } |
| 69 | if err := os.MkdirAll(p.runDir, 0o750|os.ModeSetgid); err != nil { |
| 70 | return p, nil, nil, nil, instancekey.LayerNone, err |
| 71 | } |
| 72 | if err := os.MkdirAll(filepath.Dir(b.logSocketFor(rt.Host, role, gen)), 0o700); err != nil { |
| 73 | return p, nil, nil, nil, instancekey.LayerNone, err |
| 74 | } |
| 75 | steps.mark(time.Now(), "dirs") |
| 76 | var env map[string]string |
| 77 | var keyLayer instancekey.Layer // the layer env's key came from, read with it (ruling 29) |
| 78 | switch role { |
| 79 | case instancectl.RoleRun, instancectl.RoleSeed: |
| 80 | // p.dataDir is the instance directory here, tenant-writable ground |
| 81 | // (design §3.2), so the token file is minted and read through |
| 82 | // instancefs rather than a plain path-based open. |
| 83 | if err := EnsureAdminTokenBeneath(p.dataDir, rt.Host+".env"); err != nil { |
| 84 | return p, nil, nil, nil, instancekey.LayerNone, fmt.Errorf("admin token for %s: %w", rt.Host, err) |
| 85 | } |
| 86 | steps.mark(time.Now(), "admin token") |
| 87 | if _, err := b.ensureInstanceKeyFor(context.Background(), rt, p); err != nil { // prepareAt has no ctx to bound the mint's bucket read; configReadTimeout does |
| 88 | return p, nil, nil, nil, instancekey.LayerNone, fmt.Errorf("instance key for %s: %w", rt.Host, err) |
| 89 | } |
| 90 | steps.mark(time.Now(), "instance key") |
| 91 | var err error |
| 92 | env, keyLayer, err = b.instanceEnvLayer(rt) |
| 93 | if err != nil { |
| 94 | return p, nil, nil, nil, instancekey.LayerNone, err |
| 95 | } |
| 96 | steps.mark(time.Now(), "env") |
| 97 | case instancectl.RoleCmd: |
| 98 | // A run's environment is the one its generation launched with |
| 99 | // (merge ruling M3), which StartCmd has. The delivered files are |
| 100 | // not read: config sync may have written a revision the serving |
| 101 | // generation has not activated. |
| 102 | default: |
| 103 | var err error |
| 104 | env, err = readEnvFileBeneath(p.dataDir, filepath.Base(p.sidecarEnv)) |
| 105 | if err != nil { |
| 106 | return p, nil, nil, nil, instancekey.LayerNone, err |
| 107 | } |
| 108 | keyLayer = sidecarKeyLayer(env) |
| 109 | } |
| 110 | // Stale exit records from earlier invocations of this role: a start |
| 111 | // that reads one would mis-word the very first exit line this |
| 112 | // invocation eventually gets, attributing a previous run's status to |
| 113 | // this one. |
| 114 | if entries, err := os.ReadDir(p.runDir); err == nil { |
| 115 | for _, e := range entries { |
| 116 | if strings.HasPrefix(e.Name(), exitPrefix(role, gen)) { |
| 117 | _ = os.Remove(filepath.Join(p.runDir, e.Name())) |
| 118 | } |
| 119 | } |
| 120 | } |
| 121 | var ctl *edge.ControlListener |
| 122 | if role == instancectl.RoleRun { |
| 123 | // Nothing is running: either this is the first invocation of this |
| 124 | // role for this host, or the unit is not active (Start would not |
| 125 | // have reached prepare otherwise), so any file at rt.Socket is a |
| 126 | // leftover from an earlier invocation, most likely one that was |
| 127 | // SIGKILLed before its own bind was ever established. On the |
| 128 | // nested path the agent itself never binds this socket (the app |
| 129 | // does, inside its own namespace, once the trampoline execs it), |
| 130 | // so unlike startChild's flat spawn there is no "our own live |
| 131 | // process squatting on it" case this could ever race: removing it |
| 132 | // here is safe by construction, and leaving it in place would fail |
| 133 | // EVERY later wake's health check with nothing but a dial timeout, |
| 134 | // no member-facing diagnosis at all, unlike the flat path's #129 |
| 135 | // handling, which at least turns the bind failure into a named |
| 136 | // error (memberSafeWakeReason). |
| 137 | // |
| 138 | // A generation's socket lives in a directory of its own (rev 19, |
| 139 | // rule 1), made here with the run dir's mode. It inherits the run |
| 140 | // dir's setgid carlos-sockets group, and it sits inside the run dir |
| 141 | // the generation unit binds, so the app (uid carlos, as this agent |
| 142 | // is) can bind its socket there. |
| 143 | if dir := filepath.Dir(sock); dir != p.runDir { |
| 144 | if err := os.MkdirAll(dir, 0o750|os.ModeSetgid); err != nil { |
| 145 | return p, nil, nil, nil, instancekey.LayerNone, err |
| 146 | } |
| 147 | } |
| 148 | _ = os.Remove(sock) |
| 149 | ctl = b.listenControl(rt, sock+".control", p.envFile, gen) |
| 150 | steps.mark(time.Now(), "control socket") |
| 151 | } |
| 152 | // A replica's roles are gated like a flat launch (ruling 26): the seed |
| 153 | // and the sidecar have no control socket, the run role has the one just |
| 154 | // bound or none. The sidecar's key, if its own file holds one, is |
| 155 | // checked as the flat sidecar's is (sidecarKeyLayer). A cmd is gated in |
| 156 | // StartCmd, on the environment its generation launched with: prepare |
| 157 | // read none for it. |
| 158 | if role != instancectl.RoleCmd { |
| 159 | if err := b.replicaKeyGate(rt, env, keyLayer, ctl != nil); err != nil { |
| 160 | if ctl != nil { |
| 161 | ctl.Close() |
| 162 | } |
| 163 | return p, nil, nil, nil, instancekey.LayerNone, err |
| 164 | } |
| 165 | } |
| 166 | steps.mark(time.Now(), "key gate") |
| 167 | // The cmd role's output goes only to its own chunks, never to the |
| 168 | // journal or the app log, so it has no app-log listener. |
| 169 | var logs *logListener |
| 170 | if role != instancectl.RoleCmd { |
| 171 | var err error |
| 172 | logs, err = listenLogs(b.logSocketFor(rt.Host, role, gen), b.logEmitter(rt)) |
| 173 | if err != nil { |
| 174 | if ctl != nil { |
| 175 | ctl.Close() |
| 176 | } |
| 177 | return p, nil, nil, nil, instancekey.LayerNone, fmt.Errorf("log listener for %s %s: %w", rt.Host, role, err) |
| 178 | } |
| 179 | } |
| 180 | steps.mark(time.Now(), "logs") |
| 181 | return p, ctl, logs, env, keyLayer, nil |
| 182 | } |
| 183 | |
| 184 | // exitPrefix is the exit records prepare clears before a start: this |
| 185 | // role's for the original templates, this generation's alone otherwise. |
| 186 | // A sibling generation's record is never touched (zero-downtime spec |
| 187 | // §7.6). |
| 188 | func exitPrefix(role instancectl.Role, gen uint64) string { |
| 189 | if gen == 0 { |
| 190 | return "exit." + string(role) + "." |
| 191 | } |
| 192 | return instancectl.GenExitPrefix(gen) |
| 193 | } |
| 194 | |
| 195 | // listenControl binds the control socket exactly as startChild does for an |
| 196 | // exec child (the tokenReader and schedule mount are unchanged, aside from |
| 197 | // reading beneath the instance directory rather than following a plain |
| 198 | // path, design §3.2); a bind failure is logged and the instance runs |
| 199 | // without its control verbs, as today. |
| 200 | // |
| 201 | // gen is the generation the listener serves (0 for the original |
| 202 | // templates): its schedule writes go through that generation's write fence |
| 203 | // (writeFence, final review F8). |
| 204 | func (b *ExecBackend) listenControl(rt registry.Route, control, envPath string, gen uint64) *edge.ControlListener { |
| 205 | tokenReader := func() (string, error) { |
| 206 | env, err := readEnvFileBeneath(filepath.Dir(envPath), filepath.Base(envPath)) |
| 207 | if err != nil { |
| 208 | return "", err |
| 209 | } |
| 210 | return env["CARLOS_ADMIN_TOKEN"], nil |
| 211 | } |
| 212 | var mount func(*http.ServeMux) |
| 213 | if b.Timers != nil || b.UpdatesReader != nil { |
| 214 | mount = func(mux *http.ServeMux) { |
| 215 | if b.Timers != nil { |
| 216 | schedule.MountGated(mux, rt.Host, b.Timers, tokenReader, time.Now, b.writeFence(rt.Host, gen)) |
| 217 | } |
| 218 | if b.UpdatesReader != nil { |
| 219 | instanceupdates.Mount(mux, rt.Host, tokenReader, b.UpdatesReader) |
| 220 | } |
| 221 | } |
| 222 | } |
| 223 | ctl, err := edge.ListenControl(control, rt.Host, b.Cache, mount) |
| 224 | if err != nil { |
| 225 | log.Printf("activator: control listener for %s: %v (continuing without it)", rt.Host, err) |
| 226 | return nil |
| 227 | } |
| 228 | return ctl |
| 229 | } |
| 230 | |
| 231 | // logSocket is logSocketFor for the original templates. |
| 232 | func (b *ExecBackend) logSocket(host string, role instancectl.Role) string { |
| 233 | return b.logSocketFor(host, role, 0) |
| 234 | } |
| 235 | |
| 236 | // logSocketFor is the role's log socket path, keyed by generation for a |
| 237 | // generation unit, honouring the test override of the run directory |
| 238 | // (RunDirOverride). |
| 239 | func (b *ExecBackend) logSocketFor(host string, role instancectl.Role, gen uint64) string { |
| 240 | if gen != 0 { |
| 241 | if b.RunDirOverride != "" { |
| 242 | return filepath.Join(b.RunDirOverride, "logs", instancectl.GenKey(host, gen)+".run.sock") |
| 243 | } |
| 244 | return instancectl.GenLogSocket(host, gen) |
| 245 | } |
| 246 | return b.logSocketNamed(host, string(role)) |
| 247 | } |
| 248 | |
| 249 | // logSocketNamed is an original template's log socket for any listener |
| 250 | // name: the cmd role has two, "cmd" and "cmd-err" (run spec §5.4). |
| 251 | func (b *ExecBackend) logSocketNamed(host, name string) string { |
| 252 | if b.RunDirOverride != "" { |
| 253 | return filepath.Join(b.RunDirOverride, "logs", provision.InstanceID(host)+"."+name+".sock") |
| 254 | } |
| 255 | return provision.LogSocket(host, name) |
| 256 | } |
| 257 | |
| 258 | // logEmitter is the same sink teeStd feeds, or nil when Logs is unset. |
| 259 | func (b *ExecBackend) logEmitter(rt registry.Route) func(string) { |
| 260 | if b.Logs == nil { |
| 261 | return nil |
| 262 | } |
| 263 | return func(line string) { b.Logs.App(release.RouteAccount(rt), rt.App, rt.Host, line) } |
| 264 | } |
| 265 | |
| 266 | // writeLaunch is writeLaunchFor for the original templates. |
| 267 | func (b *ExecBackend) writeLaunch(p instancePaths, role instancectl.Role, host string, argv, env []string) error { |
| 268 | return b.writeLaunchFor(p, role, 0, host, argv, env) |
| 269 | } |
| 270 | |
| 271 | // writeLaunchFor writes the launch file atomically into the instance |
| 272 | // directory (design §6.2): launch.<role>.json, or launch.gen.<n>.json for a |
| 273 | // generation (zero-downtime spec §7.6). The env is childEnv's deduplicated |
| 274 | // list; the trampoline execs it as is. |
| 275 | func (b *ExecBackend) writeLaunchFor(p instancePaths, role instancectl.Role, gen uint64, host string, argv, env []string) error { |
| 276 | raw, err := json.MarshalIndent(instancectl.Launch{Schema: instancectl.LaunchSchema, Host: host, Argv: argv, Env: env}, "", " ") |
| 277 | if err != nil { |
| 278 | return err |
| 279 | } |
| 280 | name := instancectl.LaunchFileName(role) |
| 281 | if gen != 0 { |
| 282 | name = instancectl.GenLaunchFileName(gen) |
| 283 | } |
| 284 | return instancefs.WriteAtomic(p.dataDir, name, append(raw, '\n'), 0o600) |
| 285 | } |
| 286 | |
| 287 | // readEnvFileBeneath is readEnvFile through instancefs: a nested host's env |
| 288 | // files are read without following a symlink (design §3.2). A missing file |
| 289 | // is an empty env, as readEnvFile treats it (the caller's own |
| 290 | // os.IsNotExist check would otherwise have nothing to check against). |
| 291 | func readEnvFileBeneath(dir, name string) (map[string]string, error) { |
| 292 | b, err := instancefs.ReadBeneathLimit(dir, name, 1<<20) |
| 293 | if errors.Is(err, os.ErrNotExist) { |
| 294 | return map[string]string{}, nil |
| 295 | } |
| 296 | if err != nil { |
| 297 | return nil, err |
| 298 | } |
| 299 | return parseEnvBytes(b), nil |
| 300 | } |
| 301 | |
amadan