| 1 | package layout |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "fmt" |
| 8 | "os" |
| 9 | "path/filepath" |
| 10 | "sort" |
| 11 | "strings" |
| 12 | "sync" |
| 13 | "time" |
| 14 | |
| 15 | "github.com/carlosframework/platform/internal/activator" |
| 16 | "github.com/carlosframework/platform/internal/controlname" |
| 17 | "github.com/carlosframework/platform/internal/instancefs" |
| 18 | "github.com/carlosframework/platform/internal/provision" |
| 19 | "github.com/carlosframework/platform/internal/registry" |
| 20 | ) |
| 21 | |
| 22 | // Fence is what the pass needs from the activator (design §8.2 step 2): |
| 23 | // stop the host and refuse wakes for it, but only when no other operation |
| 24 | // already holds the host's one fence slot (FenceAndStopIfUnfenced); let it |
| 25 | // wake again (LiftFence, at exactly the generation this pass installed); |
| 26 | // and the two reads that decide whether an unproven stop can be trusted. |
| 27 | type Fence interface { |
| 28 | FenceAndStopIfUnfenced(ctx context.Context, host string, generation uint64) (*activator.StopEvidence, error) |
| 29 | LiftFence(host string, generation uint64) bool |
| 30 | FenceGeneration(host string) uint64 |
| 31 | Quiesced(host string) bool |
| 32 | StopUncertain(host string) bool |
| 33 | } |
| 34 | |
| 35 | // IntentUnreadable is the intent Recover is given when the agent could not |
| 36 | // read the fleet's layout intent at all, as distinct from "" (no intent was |
| 37 | // ever recorded). A revert completed under it is held fenced; see |
| 38 | // recoverHost. |
| 39 | const IntentUnreadable Mode = "unreadable" |
| 40 | |
| 41 | // Pass is the layout pass for one box. It runs one Converge or Recover at |
| 42 | // a time; the agent's single-flight is what guarantees that. |
| 43 | type Pass struct { |
| 44 | Registry *registry.Registry |
| 45 | // Fence is the activator. Nil means there is none (tests): the pass |
| 46 | // still journals and moves. |
| 47 | Fence Fence |
| 48 | DataDir string |
| 49 | SocketDir string // "" means provision.SocketDir |
| 50 | SocketsGID int // the carlos-sockets group; -1 leaves the group alone |
| 51 | // Reload asks routing to pick up a flipped row now rather than on the |
| 52 | // next 2s poll. Nil relies on the poll. |
| 53 | Reload func() |
| 54 | Now func() time.Time |
| 55 | Logf func(string, ...any) |
| 56 | Generation func() uint64 // monotonic fence generations; default time.Now().UnixNano() |
| 57 | // SweptAtStart reports that the agent's startup unit sweep succeeded, |
| 58 | // so no instance from an earlier agent can still be running. Without |
| 59 | // it an unproven stop is never trusted. |
| 60 | SweptAtStart bool |
| 61 | // NestedOK is the isolation probe's gate (design §8.5): a box that |
| 62 | // cannot establish the namespace does not start nested tenants, so it |
| 63 | // does not make any either. Nil means ungated. |
| 64 | NestedOK func() error |
| 65 | // Serialize, when set, runs the span from journal write to journal |
| 66 | // removal under the lock config sync holds, so sync never renders a |
| 67 | // host's bundle into the directory the pass is moving it out of. |
| 68 | Serialize func(func()) |
| 69 | // DeferAwake leaves a host that is awake for a later pass, before any |
| 70 | // fence is asked for, rather than stopping it to move it. The design |
| 71 | // migrates a row that becomes eligible "when it is asleep", and the |
| 72 | // agent's serial-move leg runs while tenants are serving; recovery is |
| 73 | // not affected, because a host with a journal cannot be woken. |
| 74 | // |
| 75 | // The deferral is bounded by AwakeGrace. A flat tenant that never |
| 76 | // idles would otherwise defer its own migration for as long as it |
| 77 | // liked, and while it runs flat it has the agent's uid and so writes |
| 78 | // to every nested sibling's raw directory under /data/.instances: |
| 79 | // the box is not isolated until no flat tenant remains on it. Once a |
| 80 | // host has been seen awake for the grace period, the pass stops it |
| 81 | // the way it stops any host (the fence, §8.2 step 2) and moves it; it |
| 82 | // wakes nested on its next request. |
| 83 | DeferAwake bool |
| 84 | // AwakeGrace is how long an awake host is deferred before the pass |
| 85 | // stops it anyway; zero means DefaultAwakeGrace. |
| 86 | AwakeGrace time.Duration |
| 87 | // awakeSince is when each deferred host was first seen awake under |
| 88 | // the current direction. In memory only: an agent restart stops every |
| 89 | // tenant, so afterwards they are asleep and migrate on the next pass. |
| 90 | awakeMu sync.Mutex |
| 91 | awakeSince map[string]time.Time |
| 92 | |
| 93 | // kept is the generation of each fence this pass kept installed after |
| 94 | // a failed handover, so a later pass knows that fence as its own |
| 95 | // rather than another operation's. Cleared when the fence is lifted. |
| 96 | keptMu sync.Mutex |
| 97 | kept map[string]uint64 |
| 98 | // held is the reason each host whose nested move was refused after |
| 99 | // it had been stopped is kept fenced and down (see migrateHost). It |
| 100 | // is cleared when the move goes through, or when the intent turns |
| 101 | // flat and the fence is lifted. |
| 102 | held map[string]string |
| 103 | // notes is the last thing worth telling an operator about each host |
| 104 | // that was not a refusal: what a move quarantined or kept aside. It |
| 105 | // is an event for status, in memory only; the log carries it too. |
| 106 | notes map[string]string |
| 107 | |
| 108 | // abortAt is a test seam: the step name after which a move returns as |
| 109 | // if the agent had died there, leaving its journal for Recover. |
| 110 | abortAt string |
| 111 | } |
| 112 | |
| 113 | // Result is what the pass says about one host. Outcome is one of |
| 114 | // migrated, reverted, recovered, unchanged, skipped, deferred, refused or |
| 115 | // held (refused after the host was stopped for a nested move, and kept |
| 116 | // fenced and down; see migrateHost). A Result with an empty Host is about |
| 117 | // the whole pass. |
| 118 | type Result struct { |
| 119 | Host string |
| 120 | Outcome string |
| 121 | Reason string |
| 122 | } |
| 123 | |
| 124 | // quiesceTimeout bounds one host's stop. |
| 125 | const quiesceTimeout = 60 * time.Second |
| 126 | |
| 127 | // DefaultAwakeGrace is how long an awake host may defer its own move. An |
| 128 | // hour is longer than any idle window the activator hibernates on, so a |
| 129 | // tenant that is ever quiet in that time moves while asleep as the design |
| 130 | // prefers, and one that is never quiet is stopped once, briefly, rather |
| 131 | // than left flat beside nested siblings for good. |
| 132 | const DefaultAwakeGrace = time.Hour |
| 133 | |
| 134 | // deferAwake reports whether host, found awake, is still within its |
| 135 | // grace, and how much of the grace is left. |
| 136 | func (p *Pass) deferAwake(host string) (bool, time.Duration) { |
| 137 | grace := p.AwakeGrace |
| 138 | if grace <= 0 { |
| 139 | grace = DefaultAwakeGrace |
| 140 | } |
| 141 | p.awakeMu.Lock() |
| 142 | defer p.awakeMu.Unlock() |
| 143 | if p.awakeSince == nil { |
| 144 | p.awakeSince = map[string]time.Time{} |
| 145 | } |
| 146 | now := p.now() |
| 147 | since, seen := p.awakeSince[host] |
| 148 | if !seen { |
| 149 | p.awakeSince[host] = now |
| 150 | return true, grace |
| 151 | } |
| 152 | if left := grace - now.Sub(since); left > 0 { |
| 153 | return true, left |
| 154 | } |
| 155 | return false, 0 |
| 156 | } |
| 157 | |
| 158 | // forgetAwake clears a host's grace clock once the pass has dealt with it |
| 159 | // in some other way than deferring it. |
| 160 | func (p *Pass) forgetAwake(host string) { |
| 161 | p.awakeMu.Lock() |
| 162 | defer p.awakeMu.Unlock() |
| 163 | delete(p.awakeSince, host) |
| 164 | } |
| 165 | |
| 166 | func (p *Pass) now() time.Time { |
| 167 | if p.Now != nil { |
| 168 | return p.Now() |
| 169 | } |
| 170 | return time.Now() |
| 171 | } |
| 172 | |
| 173 | func (p *Pass) logf(format string, a ...any) { |
| 174 | if p.Logf != nil { |
| 175 | p.Logf(format, a...) |
| 176 | } |
| 177 | } |
| 178 | |
| 179 | func (p *Pass) generation() uint64 { |
| 180 | if p.Generation != nil { |
| 181 | return p.Generation() |
| 182 | } |
| 183 | return uint64(time.Now().UnixNano()) |
| 184 | } |
| 185 | |
| 186 | func (p *Pass) socketDir() string { |
| 187 | if p.SocketDir != "" { |
| 188 | return p.SocketDir |
| 189 | } |
| 190 | return provision.SocketDir |
| 191 | } |
| 192 | |
| 193 | func (p *Pass) reload() { |
| 194 | if p.Reload != nil { |
| 195 | p.Reload() |
| 196 | } |
| 197 | } |
| 198 | |
| 199 | func (p *Pass) serialize(f func()) { |
| 200 | if p.Serialize != nil { |
| 201 | p.Serialize(f) |
| 202 | return |
| 203 | } |
| 204 | f() |
| 205 | } |
| 206 | |
| 207 | func (p *Pass) layoutInputs(routes []registry.Route) activator.LayoutInputs { |
| 208 | return activator.LayoutInputs{ |
| 209 | DataDir: p.DataDir, |
| 210 | SocketDir: p.socketDir(), |
| 211 | Journal: func(host string) bool { return JournalPresent(p.DataDir, host) }, |
| 212 | Peers: func() ([]registry.Route, error) { return routes, nil }, |
| 213 | } |
| 214 | } |
| 215 | |
| 216 | // Classify is the box's one classification, shared with the status |
| 217 | // snapshot so both say the same thing about a row. |
| 218 | func (p *Pass) Classify(rt registry.Route, peers []registry.Route) (activator.Layout, string) { |
| 219 | return activator.Classify(rt, p.layoutInputs(peers)) |
| 220 | } |
| 221 | |
| 222 | func deadline(ctx context.Context) (context.Context, context.CancelFunc) { |
| 223 | return context.WithTimeout(ctx, quiesceTimeout) |
| 224 | } |
| 225 | |
| 226 | // ready is the status a row must have for the pass to move it. Any other |
| 227 | // status belongs to an operation in flight on the row, the residence |
| 228 | // move's "moving" among them, and a row about to leave the box is not |
| 229 | // worth moving. |
| 230 | func ready(rt registry.Route) bool { return rt.Status == "" || rt.Status == "ready" } |
| 231 | |
| 232 | // Converge runs Recover, then moves every managed row that is not already |
| 233 | // in mode (design §8.2 steps 1 to 9). Rows are visited in host order so a |
| 234 | // log reads the same way twice. A host Recover reported is not visited |
| 235 | // again: recovery already converged it, or refused it. |
| 236 | // |
| 237 | // Recovery runs first and ungated, so a failing isolation probe never |
| 238 | // leaves a crashed move fenced; its legs toward nested ask the gate. The |
| 239 | // fresh moves to nested wait for the gate as a whole. The gate is asked at |
| 240 | // most once per pass, and both share the answer. |
| 241 | func (p *Pass) Converge(ctx context.Context, mode Mode) []Result { |
| 242 | if mode != ModeNested && mode != ModeFlat { |
| 243 | return []Result{{Outcome: "refused", Reason: fmt.Sprintf("layout mode %q is not nested or flat", mode)}} |
| 244 | } |
| 245 | gate := p.nestedGate() |
| 246 | out := p.recover(ctx, mode, gate) |
| 247 | if mode == ModeNested { |
| 248 | if err := gate(); err != nil { |
| 249 | return append(out, Result{Outcome: "refused", Reason: "isolation not ready: " + err.Error()}) |
| 250 | } |
| 251 | } |
| 252 | seen := map[string]bool{} |
| 253 | for _, r := range out { |
| 254 | seen[r.Host] = true |
| 255 | } |
| 256 | routes, err := p.Registry.List() |
| 257 | if err != nil { |
| 258 | p.logf("layout: converge: list routes: %v", err) |
| 259 | return append(out, Result{Outcome: "refused", Reason: "list routes: " + err.Error()}) |
| 260 | } |
| 261 | sort.Slice(routes, func(i, j int) bool { return routes[i].Host < routes[j].Host }) |
| 262 | migrated := false |
| 263 | for _, rt := range routes { |
| 264 | if !activator.ActivatorManaged(rt) || seen[rt.Host] { |
| 265 | continue |
| 266 | } |
| 267 | if ctx.Err() != nil { |
| 268 | break |
| 269 | } |
| 270 | r := p.convergeHost(ctx, rt, routes, mode) |
| 271 | migrated = migrated || r.Outcome == "migrated" |
| 272 | out = append(out, r) |
| 273 | } |
| 274 | if migrated { |
| 275 | // What no host's move could take (extras.go): a flat app with cwd |
| 276 | // / wrote relative paths straight into the data directory, and a |
| 277 | // name no row owns cannot be given to one host. Said once per pass |
| 278 | // that moved something, for the operator who has to place them. |
| 279 | if left := UnattributedEntries(p.DataDir, hostsOf(routes)); len(left) > 0 { |
| 280 | p.logf("layout: left in %s, owned by no host on this box: %s. a nested instance cannot see them; move what an app wrote into that app's instance directory by hand", p.DataDir, strings.Join(left, ", ")) |
| 281 | } |
| 282 | } |
| 283 | return out |
| 284 | } |
| 285 | |
| 286 | func (p *Pass) convergeHost(ctx context.Context, rt registry.Route, routes []registry.Route, mode Mode) Result { |
| 287 | if !ready(rt) { |
| 288 | return Result{rt.Host, "deferred", "status " + rt.Status} |
| 289 | } |
| 290 | layout, reason := p.Classify(rt, routes) |
| 291 | switch layout { |
| 292 | case activator.LayoutFenced, activator.LayoutConflict: |
| 293 | return Result{rt.Host, "refused", reason} |
| 294 | case activator.LayoutForeign: |
| 295 | return Result{rt.Host, "skipped", "foreign: " + reason} |
| 296 | case activator.LayoutNested: |
| 297 | if mode == ModeNested { |
| 298 | p.forgetAwake(rt.Host) |
| 299 | // A host moved before the pass took the app's own entries |
| 300 | // with it (#256) has them still in the data directory, |
| 301 | // where its namespace cannot see them. Nothing moves here: |
| 302 | // the host is running, and what the app has minted since |
| 303 | // may be the copy it wants. Said on every pass, so an |
| 304 | // operator reading status or the log sees it while it lasts. |
| 305 | if left := FlatLeftovers(p.DataDir, rt.Host, hostsOf(routes)); len(left) > 0 { |
| 306 | text := fmt.Sprintf("still in %s from before its move, where the instance cannot see them: %s. move what the app needs into %s by hand", p.DataDir, strings.Join(left, ", "), rawInstanceDir(p.DataDir, rt.Host)) |
| 307 | p.logf("layout: %s: %s", rt.Host, text) |
| 308 | p.note(rt.Host, text) |
| 309 | } |
| 310 | return Result{rt.Host, "unchanged", ""} |
| 311 | } |
| 312 | case activator.LayoutFlat: |
| 313 | if mode == ModeFlat { |
| 314 | p.forgetAwake(rt.Host) |
| 315 | p.releaseHeld(rt.Host) |
| 316 | return Result{rt.Host, "unchanged", ""} |
| 317 | } |
| 318 | } |
| 319 | if p.DeferAwake && p.Fence != nil && !p.Fence.Quiesced(rt.Host) { |
| 320 | if within, left := p.deferAwake(rt.Host); within { |
| 321 | return Result{rt.Host, "deferred", fmt.Sprintf("awake (stopped for the move in %s unless it sleeps first)", left.Round(time.Second))} |
| 322 | } |
| 323 | p.logf("layout: %s: awake past its %s grace -> stopping it for the move", rt.Host, p.grace()) |
| 324 | } |
| 325 | p.forgetAwake(rt.Host) |
| 326 | return p.migrateHost(ctx, rt.Host, routes, mode) |
| 327 | } |
| 328 | |
| 329 | func (p *Pass) grace() time.Duration { |
| 330 | if p.AwakeGrace > 0 { |
| 331 | return p.AwakeGrace |
| 332 | } |
| 333 | return DefaultAwakeGrace |
| 334 | } |
| 335 | |
| 336 | // migrateHost is steps 2 to 9 for one host. It checks what it can before |
| 337 | // stopping anything, fences and stops the host, and releases it only after |
| 338 | // the move is committed and routing has reloaded, or has been refused. |
| 339 | func (p *Pass) migrateHost(ctx context.Context, host string, routes []registry.Route, mode Mode) Result { |
| 340 | plan, err := PlanFor(p.DataDir, p.socketDir(), host, mode) |
| 341 | if err != nil { |
| 342 | return Result{host, "refused", err.Error()} |
| 343 | } |
| 344 | plan.Logf = p.logf |
| 345 | if err := Preflight(plan, routes); err != nil { |
| 346 | return Result{host, "refused", err.Error()} |
| 347 | } |
| 348 | // A host whose database is not on this box cannot move. Finding that |
| 349 | // out must not cost it a stop, and the pass directories and a |
| 350 | // destination check, on every pass. The same check under the fence |
| 351 | // stays the authority. |
| 352 | if db := plan.Artifacts[0]; db.Required { |
| 353 | if ok, err := lstatPresent(db.Src); err != nil || !ok { |
| 354 | return Result{host, "refused", fmt.Sprintf("db is missing at %s", db.Src)} |
| 355 | } |
| 356 | } |
| 357 | r, fence, gen := p.quiesce(ctx, host, p.generation()) |
| 358 | if fence != fenceOurs { |
| 359 | return r |
| 360 | } |
| 361 | var hold bool |
| 362 | r, _, hold = p.shift(host, mode, nil) |
| 363 | // Fail closed on the way to nested. A flat host that has been stopped |
| 364 | // for its move does not run flat again because a tenant kept the move |
| 365 | // refused: a second name for its database, a file in its own future |
| 366 | // directory, a symlink at it, each one syscall from inside, and a host |
| 367 | // released here would wake flat on its next request and keep its |
| 368 | // raw-path reach over its nested siblings for as long as it kept the |
| 369 | // refusal alive. So the fence stays, the host stays down, the reason |
| 370 | // is kept for status, and every later pass retries from the kept |
| 371 | // fence: when the reason is gone the host moves and is released; when |
| 372 | // the fleet's intent turns flat, convergeHost lifts the fence and the |
| 373 | // host runs flat as asked. Refusals that come from the row (an |
| 374 | // operation in flight, an unflag) are not a tenant's doing and lift as |
| 375 | // before, so a residence move is never deadlocked on this fence. A |
| 376 | // reverse move is never held: a nested host refused its revert is |
| 377 | // already isolated, and leaving it running is the safe side. |
| 378 | if hold && mode == ModeNested { |
| 379 | p.keep(host, gen) |
| 380 | p.hold(host, r.Reason) |
| 381 | p.logf("layout: %s: held -> %s", host, r.Reason) |
| 382 | return Result{host, "held", r.Reason + "; held fenced and down until the reason is cleared or the layout intent is flat"} |
| 383 | } |
| 384 | p.release(host, gen) |
| 385 | return r |
| 386 | } |
| 387 | |
| 388 | // hold records why host is kept fenced and down after a refused nested |
| 389 | // move; Held is how status reads it. The reason is also written to a hold |
| 390 | // marker in the layout directory, so an agent restart (every carlos |
| 391 | // system update) can restore the hold before anything wakes |
| 392 | // (RestoreHolds) instead of handing the tenant a fresh grace. |
| 393 | // |
| 394 | // The marker is uid carlos's file like everything under /data, so a flat |
| 395 | // tenant can unlink it. That buys the tenant nothing durable: the marker |
| 396 | // is a hint to re-check, not the authority. At start the pass re-derives |
| 397 | // the hold from the files themselves (the planted link or symlink is the |
| 398 | // reason, and it is the thing the tenant would have to remove to be rid |
| 399 | // of the hold, which is the outcome wanted), so a deleted marker costs |
| 400 | // one awake grace after one restart, after which the next pass meets the |
| 401 | // same refusal and holds again. A marker whose reason is gone is dropped |
| 402 | // at start rather than honoured, for the same reason in reverse. |
| 403 | func (p *Pass) hold(host, reason string) { |
| 404 | p.keptMu.Lock() |
| 405 | if p.held == nil { |
| 406 | p.held = map[string]string{} |
| 407 | } |
| 408 | p.held[host] = reason |
| 409 | p.keptMu.Unlock() |
| 410 | if err := writeHoldMarker(p.DataDir, host, reason, p.now()); err != nil { |
| 411 | p.logf("layout: %s: hold marker: %v (the hold stands in memory until the next restart)", host, err) |
| 412 | } |
| 413 | } |
| 414 | |
| 415 | // note records an event for status; Note reads it. |
| 416 | func (p *Pass) note(host, text string) { |
| 417 | p.keptMu.Lock() |
| 418 | defer p.keptMu.Unlock() |
| 419 | if p.notes == nil { |
| 420 | p.notes = map[string]string{} |
| 421 | } |
| 422 | p.notes[host] = text |
| 423 | } |
| 424 | |
| 425 | // Note is the last event the pass recorded for host that was not a |
| 426 | // refusal: what a move quarantined or kept aside, and when. |
| 427 | func (p *Pass) Note(host string) string { |
| 428 | p.keptMu.Lock() |
| 429 | defer p.keptMu.Unlock() |
| 430 | return p.notes[host] |
| 431 | } |
| 432 | |
| 433 | // holdMarker is the on-disk record of a hold: <data>/layout/<host>.hold. |
| 434 | // The suffix is not .json, so JournalHosts and the downgrade guard, which |
| 435 | // list journals by that suffix, never take a marker for a journal. |
| 436 | type holdMarker struct { |
| 437 | Host string `json:"host"` |
| 438 | Reason string `json:"reason"` |
| 439 | Since time.Time `json:"since"` |
| 440 | } |
| 441 | |
| 442 | func holdMarkerName(host string) string { return host + ".hold" } |
| 443 | |
| 444 | func writeHoldMarker(dataDir, host, reason string, at time.Time) error { |
| 445 | if _, err := instancefs.EnsureDirBeneath(dataDir, filepath.Base(journalDir(dataDir)), 0o700); err != nil { |
| 446 | return err |
| 447 | } |
| 448 | raw, err := json.Marshal(holdMarker{Host: host, Reason: reason, Since: at.UTC()}) |
| 449 | if err != nil { |
| 450 | return err |
| 451 | } |
| 452 | return instancefs.WriteAtomic(journalDir(dataDir), holdMarkerName(host), raw, 0o600) |
| 453 | } |
| 454 | |
| 455 | func removeHoldMarker(dataDir, host string) { |
| 456 | if err := instancefs.RemoveBeneath(journalDir(dataDir), holdMarkerName(host)); err != nil && !errors.Is(err, os.ErrNotExist) { |
| 457 | // Best effort: a marker that outlives its hold is dropped at the |
| 458 | // next start when its reason does not reproduce. |
| 459 | _ = err |
| 460 | } |
| 461 | } |
| 462 | |
| 463 | // HoldMarkerPresent reports whether host has a hold marker, without reading |
| 464 | // it. Lstat, like JournalPresent, and an error other than "not exist" |
| 465 | // counts as present: a marker this process cannot stat is not proof there |
| 466 | // is none. |
| 467 | func HoldMarkerPresent(dataDir, host string) bool { |
| 468 | _, err := os.Lstat(filepath.Join(journalDir(dataDir), holdMarkerName(host))) |
| 469 | return !errors.Is(err, os.ErrNotExist) |
| 470 | } |
| 471 | |
| 472 | // HoldMarkers lists the hosts with a hold marker in the layout directory. |
| 473 | func HoldMarkers(dataDir string) ([]string, error) { |
| 474 | entries, err := os.ReadDir(journalDir(dataDir)) |
| 475 | if errors.Is(err, os.ErrNotExist) { |
| 476 | return nil, nil |
| 477 | } |
| 478 | if err != nil { |
| 479 | return nil, err |
| 480 | } |
| 481 | var hosts []string |
| 482 | for _, e := range entries { |
| 483 | if n := e.Name(); !strings.HasPrefix(n, ".") && strings.HasSuffix(n, ".hold") { |
| 484 | hosts = append(hosts, strings.TrimSuffix(n, ".hold")) |
| 485 | } |
| 486 | } |
| 487 | return hosts, nil |
| 488 | } |
| 489 | |
| 490 | // RestoreHolds runs at agent start, after journal recovery and before the |
| 491 | // activator can wake anything. For each hold marker it re-derives the |
| 492 | // hold from the files: a flat, managed host whose pre-journal checks |
| 493 | // still refuse for a reason a tenant can plant is fenced again and held, |
| 494 | // so a restart does not hand it a fresh grace to run flat in. A marker |
| 495 | // whose reason is gone is removed, and under a flat intent, or none, every |
| 496 | // marker is removed: the fleet does not want the host nested. An intent |
| 497 | // that could not be read is treated as nested, the fail-closed side. |
| 498 | func (p *Pass) RestoreHolds(ctx context.Context, intent Mode) []Result { |
| 499 | hosts, err := HoldMarkers(p.DataDir) |
| 500 | if err != nil { |
| 501 | p.logf("layout: restore holds: %v", err) |
| 502 | return []Result{{Outcome: "refused", Reason: "read hold markers: " + err.Error()}} |
| 503 | } |
| 504 | if len(hosts) == 0 { |
| 505 | return nil |
| 506 | } |
| 507 | sort.Strings(hosts) |
| 508 | var out []Result |
| 509 | for _, host := range hosts { |
| 510 | if intent != ModeNested && intent != IntentUnreadable { |
| 511 | removeHoldMarker(p.DataDir, host) |
| 512 | out = append(out, Result{host, "released", "hold marker dropped: the layout intent is not nested"}) |
| 513 | continue |
| 514 | } |
| 515 | out = append(out, p.restoreHold(ctx, host)) |
| 516 | } |
| 517 | return out |
| 518 | } |
| 519 | |
| 520 | func (p *Pass) restoreHold(ctx context.Context, host string) Result { |
| 521 | rt, err := p.Registry.Get(host) |
| 522 | if err != nil { |
| 523 | return Result{host, "refused", "row: " + err.Error()} |
| 524 | } |
| 525 | routes, err := p.Registry.List() |
| 526 | if err != nil { |
| 527 | return Result{host, "refused", "list routes: " + err.Error()} |
| 528 | } |
| 529 | if rt == nil || !activator.ActivatorManaged(*rt) { |
| 530 | removeHoldMarker(p.DataDir, host) |
| 531 | return Result{host, "released", "hold marker dropped: the row is gone or no longer managed"} |
| 532 | } |
| 533 | if layout, _ := p.Classify(*rt, routes); layout != activator.LayoutFlat { |
| 534 | removeHoldMarker(p.DataDir, host) |
| 535 | return Result{host, "released", "hold marker dropped: the row is " + string(layout)} |
| 536 | } |
| 537 | reason, plantable := p.recheck(host, routes) |
| 538 | if !plantable { |
| 539 | removeHoldMarker(p.DataDir, host) |
| 540 | if reason == "" { |
| 541 | return Result{host, "released", "hold marker dropped: its reason is gone; the next pass moves the host"} |
| 542 | } |
| 543 | return Result{host, "released", "hold marker dropped: " + reason} |
| 544 | } |
| 545 | _, fence, gen := p.quiesce(ctx, host, p.generation()) |
| 546 | if fence != fenceOurs { |
| 547 | return Result{host, "deferred", "hold not restored: " + reason} |
| 548 | } |
| 549 | p.keep(host, gen) |
| 550 | p.hold(host, reason) |
| 551 | p.logf("layout: %s: held -> %s (restored at start)", host, reason) |
| 552 | return Result{host, "held", reason + "; restored at start"} |
| 553 | } |
| 554 | |
| 555 | // recheck runs the pre-journal checks a nested move would run for host |
| 556 | // and reports the first refusal and whether it is one a tenant can plant |
| 557 | // (the same classes shift holds on). An empty reason means the move |
| 558 | // would go ahead. |
| 559 | func (p *Pass) recheck(host string, routes []registry.Route) (reason string, plantable bool) { |
| 560 | plan, err := PlanFor(p.DataDir, p.socketDir(), host, ModeNested) |
| 561 | if err != nil { |
| 562 | return err.Error(), false |
| 563 | } |
| 564 | plan.Logf = p.logf |
| 565 | |
| 566 | if err := Preflight(plan, routes); err != nil { |
| 567 | return err.Error(), false |
| 568 | } |
| 569 | if err := EnsureDirs(p.DataDir, p.socketDir(), host, p.SocketsGID); err != nil { |
| 570 | return err.Error(), true |
| 571 | } |
| 572 | if _, err := presence(plan, hostsOf(routes)); err != nil { |
| 573 | return err.Error(), true |
| 574 | } |
| 575 | return "", false |
| 576 | } |
| 577 | |
| 578 | // hostsOf is every host with a row on the box, whatever its kind: the set |
| 579 | // an entry's owner is chosen from (extras.go). |
| 580 | func hostsOf(routes []registry.Route) []string { |
| 581 | hosts := make([]string, 0, len(routes)) |
| 582 | for _, rt := range routes { |
| 583 | hosts = append(hosts, rt.Host) |
| 584 | } |
| 585 | return hosts |
| 586 | } |
| 587 | |
| 588 | // Held reports whether the pass keeps host fenced and down after a |
| 589 | // refused nested move, and why. The fence must still be the pass's own: |
| 590 | // a record the activator no longer matches is dropped. |
| 591 | func (p *Pass) Held(host string) (string, bool) { |
| 592 | if _, ok := p.keptFence(host); !ok { |
| 593 | p.keptMu.Lock() |
| 594 | delete(p.held, host) |
| 595 | p.keptMu.Unlock() |
| 596 | return "", false |
| 597 | } |
| 598 | p.keptMu.Lock() |
| 599 | defer p.keptMu.Unlock() |
| 600 | reason, ok := p.held[host] |
| 601 | return reason, ok |
| 602 | } |
| 603 | |
| 604 | // releaseHeld lifts a fence the pass kept for host after a refused nested |
| 605 | // move, because the fleet now wants the host flat. |
| 606 | func (p *Pass) releaseHeld(host string) { |
| 607 | if gen, ok := p.keptFence(host); ok { |
| 608 | p.logf("layout: %s: released -> the layout intent is flat; the held fence is lifted", host) |
| 609 | p.release(host, gen) |
| 610 | } |
| 611 | } |
| 612 | |
| 613 | // fenceState is who holds a host's fence once quiesce returns. |
| 614 | type fenceState int |
| 615 | |
| 616 | const ( |
| 617 | // fenceOurs: this pass installed the fence and the stop is proven or |
| 618 | // trusted, or it is the fence an earlier pass kept after a failed |
| 619 | // handover, which has held the host stopped ever since. |
| 620 | fenceOurs fenceState = iota |
| 621 | // fenceHeld: another operation's fence holds the host; the pass |
| 622 | // installed nothing and must lift nothing. |
| 623 | fenceHeld |
| 624 | // fenceNone: the pass installed nothing, or lifted what it installed, |
| 625 | // and the host is not to be touched. |
| 626 | fenceNone |
| 627 | ) |
| 628 | |
| 629 | // quiesce installs this pass's fence, unless another operation's fence |
| 630 | // already holds the host, and decides whether the host is proven stopped. |
| 631 | // The residence move shares the activator's one fence slot per host, so |
| 632 | // installing over its fence and later lifting ours would reopen admission |
| 633 | // the move closed; the exclusive install refuses instead, atomically. An unproven stop is trusted only when nothing can be |
| 634 | // running: no live entry, no stop on record that failed to finish, and a |
| 635 | // startup sweep that stopped anything an earlier agent left. After an |
| 636 | // agent restart no asleep host has a stop on record, which is why the |
| 637 | // entry-less case needs this rather than a proof. Every refusal lifts the |
| 638 | // fence it installed: a fence left behind would keep a host down with |
| 639 | // nothing on its way to release it. |
| 640 | // |
| 641 | // It returns the generation the host is fenced at when the fence is the |
| 642 | // pass's: gen for a fence it just installed, or the generation an earlier |
| 643 | // pass kept after a failed handover, which the caller lifts in the end. |
| 644 | func (p *Pass) quiesce(ctx context.Context, host string, gen uint64) (Result, fenceState, uint64) { |
| 645 | if p.Fence == nil { |
| 646 | return Result{}, fenceOurs, gen |
| 647 | } |
| 648 | fctx, cancel := deadline(ctx) |
| 649 | defer cancel() |
| 650 | _, err := p.Fence.FenceAndStopIfUnfenced(fctx, host, gen) |
| 651 | if err == nil { |
| 652 | return Result{}, fenceOurs, gen |
| 653 | } |
| 654 | if errors.Is(err, activator.ErrFenceHeld) { |
| 655 | if kept, ok := p.keptFence(host); ok { |
| 656 | return Result{}, fenceOurs, kept |
| 657 | } |
| 658 | p.logf("layout: %s: deferred -> fenced by another operation", host) |
| 659 | return Result{host, "deferred", "fenced by another operation"}, fenceHeld, 0 |
| 660 | } |
| 661 | unproven := errors.Is(err, activator.ErrStopUnproven) |
| 662 | if unproven && p.SweptAtStart && p.Fence.Quiesced(host) && !p.Fence.StopUncertain(host) { |
| 663 | return Result{}, fenceOurs, gen |
| 664 | } |
| 665 | p.Fence.LiftFence(host, gen) |
| 666 | reason := "stop unproven" |
| 667 | if !unproven { |
| 668 | reason += ": " + err.Error() |
| 669 | } |
| 670 | p.logf("layout: %s: deferred -> %s", host, reason) |
| 671 | return Result{host, "deferred", reason}, fenceNone, 0 |
| 672 | } |
| 673 | |
| 674 | // keep records a fence this pass leaves installed after a failed handover. |
| 675 | func (p *Pass) keep(host string, gen uint64) { |
| 676 | p.keptMu.Lock() |
| 677 | defer p.keptMu.Unlock() |
| 678 | if p.kept == nil { |
| 679 | p.kept = map[string]uint64{} |
| 680 | } |
| 681 | p.kept[host] = gen |
| 682 | } |
| 683 | |
| 684 | // keptFence reports the generation of a fence this pass kept for host, |
| 685 | // provided the fence installed now is still exactly that one. A record |
| 686 | // the activator no longer matches is dropped, so a later fence of another |
| 687 | // operation is never taken for the pass's own. |
| 688 | func (p *Pass) keptFence(host string) (uint64, bool) { |
| 689 | p.keptMu.Lock() |
| 690 | defer p.keptMu.Unlock() |
| 691 | gen, ok := p.kept[host] |
| 692 | if !ok { |
| 693 | return 0, false |
| 694 | } |
| 695 | if p.Fence == nil || p.Fence.FenceGeneration(host) != gen { |
| 696 | delete(p.kept, host) |
| 697 | return 0, false |
| 698 | } |
| 699 | return gen, true |
| 700 | } |
| 701 | |
| 702 | func (p *Pass) release(host string, gen uint64) { |
| 703 | if p.Fence != nil { |
| 704 | p.Fence.LiftFence(host, gen) |
| 705 | } |
| 706 | p.keptMu.Lock() |
| 707 | defer p.keptMu.Unlock() |
| 708 | if p.kept[host] == gen { |
| 709 | delete(p.kept, host) |
| 710 | if _, held := p.held[host]; held { |
| 711 | delete(p.held, host) |
| 712 | removeHoldMarker(p.DataDir, host) |
| 713 | } |
| 714 | } |
| 715 | } |
| 716 | |
| 717 | // shift is steps 3 to 9 for a host this pass already holds fenced and |
| 718 | // stopped. It re-reads the row under the fence, because the copy the pass |
| 719 | // classified can be stale: a move cut can have started, or the row been |
| 720 | // unflagged, re-added or migrated. Every return before the journal is |
| 721 | // written leaves nothing changed. |
| 722 | // |
| 723 | // prior is a completed journal recovery is still holding the host with, |
| 724 | // or nil. That journal is this pass's own, so it does not count as the |
| 725 | // host being fenced here. It is replaced by the new journal in one rename |
| 726 | // (ReplaceJournal), so the host holds one journal or the other at every |
| 727 | // instant, and a refusal before the handover leaves it held by prior. The |
| 728 | // handover result says which of those happened. |
| 729 | // |
| 730 | // hold is set on a refusal a tenant could have brought about itself from |
| 731 | // inside the box, with a syscall on files the agent's uid owns: a planted |
| 732 | // symlink or entry at the instance directory, a planted file at a |
| 733 | // destination, a second name for a file about to move. A refusal that |
| 734 | // comes from the row (an operation in flight, an unflag, a route-name |
| 735 | // collision) or from the pass's own seams is not one, and is not held. |
| 736 | func (p *Pass) shift(host string, mode Mode, prior *Journal) (res Result, h handover, hold bool) { |
| 737 | fresh, err := p.Registry.Get(host) |
| 738 | if err != nil { |
| 739 | return Result{host, "refused", "row: " + err.Error()}, handoverNone, false |
| 740 | } |
| 741 | if fresh == nil { |
| 742 | return Result{host, "refused", "row vanished under the fence"}, handoverNone, false |
| 743 | } |
| 744 | if !ready(*fresh) { |
| 745 | return Result{host, "refused", "status " + fresh.Status + " under the fence"}, handoverNone, false |
| 746 | } |
| 747 | if !activator.ActivatorManaged(*fresh) { |
| 748 | return Result{host, "refused", "no longer activator-managed under the fence"}, handoverNone, false |
| 749 | } |
| 750 | routes, err := p.Registry.List() |
| 751 | if err != nil { |
| 752 | return Result{host, "refused", "list routes: " + err.Error()}, handoverNone, false |
| 753 | } |
| 754 | want, from := activator.LayoutNested, activator.LayoutFlat |
| 755 | if mode == ModeFlat { |
| 756 | want, from = activator.LayoutFlat, activator.LayoutNested |
| 757 | } |
| 758 | in := p.layoutInputs(routes) |
| 759 | if prior != nil { |
| 760 | in.Journal = func(h string) bool { return h != host && JournalPresent(p.DataDir, h) } |
| 761 | } |
| 762 | switch layout, reason := activator.Classify(*fresh, in); layout { |
| 763 | case want: |
| 764 | return Result{host, "unchanged", ""}, handoverNone, false |
| 765 | case from: |
| 766 | default: |
| 767 | return Result{host, "refused", fmt.Sprintf("%s under the fence: %s", layout, reason)}, handoverNone, false |
| 768 | } |
| 769 | plan, err := PlanFor(p.DataDir, p.socketDir(), host, mode) |
| 770 | if err != nil { |
| 771 | return Result{host, "refused", err.Error()}, handoverNone, false |
| 772 | } |
| 773 | plan.Logf = p.logf |
| 774 | |
| 775 | if err := Preflight(plan, routes); err != nil { |
| 776 | return Result{host, "refused", err.Error()}, handoverNone, false |
| 777 | } |
| 778 | // Both directions: a revert renames out of the raw directory, so it |
| 779 | // must be a real directory too, and it is cheaper to find a planted |
| 780 | // link here than after the journal has fenced the host. |
| 781 | if err := EnsureDirs(p.DataDir, p.socketDir(), host, p.SocketsGID); err != nil { |
| 782 | return Result{host, "refused", err.Error()}, handoverNone, true |
| 783 | } |
| 784 | aside, err := DestinationCheck(plan, p.now) |
| 785 | if err != nil { |
| 786 | return Result{host, "refused", err.Error()}, handoverNone, true |
| 787 | } |
| 788 | // The presence scan runs inside the section config sync is locked out |
| 789 | // of, with the journal write: a bundle rendered between a scan outside |
| 790 | // it and the journal would be recorded absent and left at the old path. |
| 791 | p.serialize(func() { |
| 792 | present, err := presence(plan, hostsOf(routes)) |
| 793 | if err != nil { |
| 794 | res, h, hold = Result{host, "refused", err.Error()}, handoverNone, true |
| 795 | return |
| 796 | } |
| 797 | res, h = p.commit(plan, plan.Journal(present), aside, prior) |
| 798 | }) |
| 799 | if res.Outcome == "migrated" || res.Outcome == "reverted" { |
| 800 | p.reload() |
| 801 | } |
| 802 | return res, h, hold |
| 803 | } |
| 804 | |
| 805 | // presence records which artifacts are at their sources, in artifactNames |
| 806 | // order, and refuses a plan whose database is not. It then adds to the |
| 807 | // plan every entry the app made for the host (ScanExtras; hosts is the |
| 808 | // box's rows, which decide ownership), so the journal written from the |
| 809 | // plan names them and the move carries them. |
| 810 | func presence(plan *Plan, hosts []string) ([]bool, error) { |
| 811 | present := make([]bool, len(plan.Artifacts)) |
| 812 | for i, a := range plan.Artifacts { |
| 813 | // Nothing is ever renamed for the socket: the listener is gone |
| 814 | // with the stopped host, and the next wake binds the new path. |
| 815 | if a.Name == "socket" { |
| 816 | continue |
| 817 | } |
| 818 | ok, err := lstatPresent(a.Src) |
| 819 | if err != nil { |
| 820 | return nil, err |
| 821 | } |
| 822 | if a.Required && !ok { |
| 823 | return nil, fmt.Errorf("%s is missing at %s", a.Name, a.Src) |
| 824 | } |
| 825 | present[i] = ok |
| 826 | plan.Artifacts[i].Present = ok |
| 827 | } |
| 828 | extras, err := ScanExtras(plan, hosts) |
| 829 | if err != nil { |
| 830 | return nil, err |
| 831 | } |
| 832 | if err := plan.AddExtras(extras); err != nil { |
| 833 | return nil, err |
| 834 | } |
| 835 | // With presence known, every file about to move must have one name. |
| 836 | // This runs under the same lock as the journal write, so nothing the |
| 837 | // agent itself does can add a name between the check and the journal. |
| 838 | if err := LinkCheck(plan, func(a Artifact) string { return a.Src }); err != nil { |
| 839 | return nil, err |
| 840 | } |
| 841 | return present, nil |
| 842 | } |
| 843 | |
| 844 | // applyJournal gives a plan rebuilt from the host name what the journal |
| 845 | // recorded about the move: which fixed artifacts were present, and the |
| 846 | // extra entries by name. Validate has already checked the names' shape; |
| 847 | // this checks their ownership against the box's rows as they stand now, |
| 848 | // because a journal is not an instruction (design §3.3): a forged journal |
| 849 | // for H naming H-x, where H-x is a host on the box, would otherwise move |
| 850 | // that host's own directory into H's tree on recovery. |
| 851 | func applyJournal(plan *Plan, j *Journal, hosts []string) error { |
| 852 | for i := range plan.Artifacts { |
| 853 | if i < len(j.Artifacts) { |
| 854 | plan.Artifacts[i].Present = j.Artifacts[i].Present |
| 855 | } |
| 856 | } |
| 857 | all := append([]string{plan.Host}, hosts...) |
| 858 | for _, name := range j.extraNames() { |
| 859 | if owner := ownerOf(name, all); owner != plan.Host { |
| 860 | return fmt.Errorf("layout: %s: journal names extra %q, which belongs to %s", plan.Host, name, owner) |
| 861 | } |
| 862 | } |
| 863 | return plan.AddExtras(j.extraNames()) |
| 864 | } |
| 865 | |
| 866 | // commit is steps 6 to 9 up to the journal's removal. The caller reloads |
| 867 | // routing afterwards, outside Serialize's lock and still under the fence. |
| 868 | // Every return after the journal is written leaves it for Recover, and the |
| 869 | // journal keeps the host fenced until then. |
| 870 | func (p *Pass) commit(plan *Plan, j *Journal, aside []string, prior *Journal) (Result, handover) { |
| 871 | host := plan.Host |
| 872 | if prior == nil { |
| 873 | if err := WriteJournal(p.DataDir, j); err != nil { |
| 874 | return Result{host, "refused", err.Error()}, handoverNone |
| 875 | } |
| 876 | return p.commitWritten(plan, j, aside), handoverNone |
| 877 | } |
| 878 | if err := ReplaceJournal(p.DataDir, prior, j); err != nil { |
| 879 | // Whatever is on disk now, the pass cannot vouch that it fences |
| 880 | // the host, so the caller keeps its own fence instead. |
| 881 | return Result{host, "refused", err.Error()}, handoverFailed |
| 882 | } |
| 883 | return p.commitWritten(plan, j, aside), handoverDone |
| 884 | } |
| 885 | |
| 886 | // handover is what became of the journal recovery held a host with when |
| 887 | // the converge leg ran. |
| 888 | type handover int |
| 889 | |
| 890 | const ( |
| 891 | // handoverNone: the leg stopped before the handover; the held journal |
| 892 | // is untouched and still fences the host. |
| 893 | handoverNone handover = iota |
| 894 | // handoverDone: the new journal replaced it; the host is fenced by the |
| 895 | // new journal until that move completes. |
| 896 | handoverDone |
| 897 | // handoverFailed: the replacement itself failed, and what is on disk |
| 898 | // is not known to fence the host. The in-memory fence must stay. |
| 899 | handoverFailed |
| 900 | ) |
| 901 | |
| 902 | // commitWritten is commit after the journal is on disk. |
| 903 | func (p *Pass) commitWritten(plan *Plan, j *Journal, aside []string) Result { |
| 904 | host := plan.Host |
| 905 | if p.abortAt == "after-journal" { |
| 906 | return Result{host, "refused", "aborted"} |
| 907 | } |
| 908 | if err := p.moveFiles(plan); err != nil { |
| 909 | return Result{host, "refused", err.Error()} |
| 910 | } |
| 911 | moved, err := VerifyMoved(plan) |
| 912 | aside = append(aside, moved...) |
| 913 | if err != nil { |
| 914 | return Result{host, "refused", notActing(err)} |
| 915 | } |
| 916 | if p.abortAt == "after-files" { |
| 917 | return Result{host, "refused", "aborted"} |
| 918 | } |
| 919 | if err := UpdatePhase(p.DataDir, j, PhaseRow); err != nil { |
| 920 | return Result{host, "refused", err.Error()} |
| 921 | } |
| 922 | if err := p.Registry.SetInstancePaths(host, plan.From.DB, plan.From.Socket, plan.To.DB, plan.To.Socket); err != nil { |
| 923 | return Result{host, "refused", "row: " + err.Error()} |
| 924 | } |
| 925 | if p.abortAt == "after-cas" { |
| 926 | return Result{host, "refused", "aborted"} |
| 927 | } |
| 928 | if err := UpdatePhase(p.DataDir, j, PhaseDone); err != nil { |
| 929 | return Result{host, "refused", err.Error()} |
| 930 | } |
| 931 | if p.abortAt == "before-remove" { |
| 932 | return Result{host, "refused", "aborted"} |
| 933 | } |
| 934 | if err := RemoveJournal(p.DataDir, host); err != nil { |
| 935 | return Result{host, "refused", err.Error()} |
| 936 | } |
| 937 | outcome := "migrated" |
| 938 | if plan.Mode == ModeFlat { |
| 939 | outcome = "reverted" |
| 940 | } |
| 941 | var events []string |
| 942 | if extras := plan.ExtraNames(); len(extras) > 0 { |
| 943 | events = append(events, "moved with it: "+strings.Join(extras, ", ")) |
| 944 | } |
| 945 | if len(aside) > 0 { |
| 946 | events = append(events, "kept aside: "+strings.Join(aside, ", ")) |
| 947 | } |
| 948 | if len(events) > 0 { |
| 949 | p.logf("layout: %s: %s -> %s (%s)", host, plan.From.DB, plan.To.DB, strings.Join(events, "; ")) |
| 950 | p.note(host, fmt.Sprintf("%s at %s: %s", outcome, p.now().UTC().Format(time.RFC3339), strings.Join(events, "; "))) |
| 951 | } else { |
| 952 | p.logf("layout: %s: %s -> %s", host, plan.From.DB, plan.To.DB) |
| 953 | } |
| 954 | return Result{host, outcome, ""} |
| 955 | } |
| 956 | |
| 957 | var errAborted = errors.New("layout: aborted by test seam") |
| 958 | |
| 959 | // moveFiles renames every artifact the plan records present (from the |
| 960 | // scan on a first attempt, from the journal on recovery), the app's own |
| 961 | // entries included. One already at its destination is done; one still at |
| 962 | // its source is renamed; one at neither is lost, and one at both was not |
| 963 | // left that way by a crash (the destination check cleared every |
| 964 | // destination before the journal was written, and a rename leaves a file |
| 965 | // in exactly one place). Either of the last two is an error that keeps |
| 966 | // the journal and so the fence. |
| 967 | func (p *Pass) moveFiles(plan *Plan) error { |
| 968 | for _, a := range plan.Artifacts { |
| 969 | if a.Name == "socket" || !a.Present { |
| 970 | continue |
| 971 | } |
| 972 | atDst, err := lstatPresent(a.Dst) |
| 973 | if err != nil { |
| 974 | return err |
| 975 | } |
| 976 | atSrc, err := lstatPresent(a.Src) |
| 977 | if err != nil { |
| 978 | return err |
| 979 | } |
| 980 | switch { |
| 981 | case atDst && atSrc: |
| 982 | return fmt.Errorf("layout: %s: %s is at both %s and %s; refusing to choose, not acting", plan.Host, a.Name, a.Src, a.Dst) |
| 983 | case atDst: |
| 984 | continue |
| 985 | case !atSrc: |
| 986 | return fmt.Errorf("layout: %s: %s is at neither %s nor %s; not acting", plan.Host, a.Name, a.Src, a.Dst) |
| 987 | } |
| 988 | if err := instancefs.RenameBetween(filepath.Dir(a.Src), filepath.Base(a.Src), filepath.Dir(a.Dst), filepath.Base(a.Dst)); err != nil { |
| 989 | return fmt.Errorf("layout: %s: move %s: %w", plan.Host, a.Name, err) |
| 990 | } |
| 991 | if p.abortAt == "mid-files" { |
| 992 | return errAborted |
| 993 | } |
| 994 | } |
| 995 | return nil |
| 996 | } |
| 997 | |
| 998 | // Recover completes or refuses every journal in <data>/layout (design |
| 999 | // §8.2, "Recovery is the journal"). It runs at agent start and at the top |
| 1000 | // of every Converge. intent is the fleet's recorded layout: "" when none |
| 1001 | // was ever recorded, IntentUnreadable when the agent could not read it. |
| 1002 | // Completing a journal is never gated on the isolation probe, since that |
| 1003 | // is how a host gets out of the fenced state; a leg after it toward a |
| 1004 | // nested intent is (see recoverHost). |
| 1005 | func (p *Pass) Recover(ctx context.Context, intent Mode) []Result { |
| 1006 | return p.recover(ctx, intent, p.nestedGate()) |
| 1007 | } |
| 1008 | |
| 1009 | func (p *Pass) recover(ctx context.Context, intent Mode, gate func() error) []Result { |
| 1010 | names, err := JournalHosts(p.DataDir) |
| 1011 | if err != nil { |
| 1012 | p.logf("layout: recover: read journals: %v", err) |
| 1013 | return []Result{{Outcome: "refused", Reason: "read journals: " + err.Error()}} |
| 1014 | } |
| 1015 | if len(names) == 0 { |
| 1016 | return nil |
| 1017 | } |
| 1018 | routes, err := p.Registry.List() |
| 1019 | if err != nil { |
| 1020 | p.logf("layout: recover: list routes: %v", err) |
| 1021 | return []Result{{Outcome: "refused", Reason: "list routes: " + err.Error()}} |
| 1022 | } |
| 1023 | byHost := map[string]registry.Route{} |
| 1024 | for _, rt := range routes { |
| 1025 | byHost[rt.Host] = rt |
| 1026 | } |
| 1027 | var out []Result |
| 1028 | for _, host := range names { |
| 1029 | if ctx.Err() != nil { |
| 1030 | break |
| 1031 | } |
| 1032 | r := p.recoverHost(ctx, host, byHost, intent, gate) |
| 1033 | if r.Outcome == "refused" { |
| 1034 | p.logf("layout: %s: recovery refused -> %s", host, r.Reason) |
| 1035 | } |
| 1036 | out = append(out, r) |
| 1037 | } |
| 1038 | return out |
| 1039 | } |
| 1040 | |
| 1041 | // nestedGate asks NestedOK at most once, on first use, and remembers the |
| 1042 | // answer for the rest of one pass. |
| 1043 | func (p *Pass) nestedGate() func() error { |
| 1044 | asked := false |
| 1045 | var answer error |
| 1046 | return func() error { |
| 1047 | if p.NestedOK == nil { |
| 1048 | return nil |
| 1049 | } |
| 1050 | if !asked { |
| 1051 | asked, answer = true, p.NestedOK() |
| 1052 | } |
| 1053 | return answer |
| 1054 | } |
| 1055 | } |
| 1056 | |
| 1057 | // notActing makes a recovery refusal say, in the words design §8.2 uses, |
| 1058 | // that the pass left the host alone and fenced. |
| 1059 | func notActing(err error) string { |
| 1060 | if s := err.Error(); strings.HasSuffix(s, "not acting") { |
| 1061 | return s |
| 1062 | } |
| 1063 | return err.Error() + "; not acting" |
| 1064 | } |
| 1065 | |
| 1066 | // recoverHost checks one journal against the row and both derived |
| 1067 | // endpoints, completes it to its own end under a fence, and then, still |
| 1068 | // under that fence, converges the host to intent. The host is released |
| 1069 | // only once it stands at the intent (design §3.3): while the converge leg |
| 1070 | // runs, the completed journal stays on disk at done and keeps the host |
| 1071 | // fenced, and if the leg refuses, for any reason, it stays there. A flat |
| 1072 | // tenant can forge a valid revert journal for a nested sibling; completing |
| 1073 | // it costs that sibling a round trip, and if the way back is refused the |
| 1074 | // sibling waits fenced for an operator rather than running flat. |
| 1075 | // |
| 1076 | // Completing a journal is never gated on the isolation probe; the leg |
| 1077 | // after it toward a nested intent is, exactly as Converge(ModeNested) is. |
| 1078 | // A gate that refuses the leg holds the host like any other refused leg. |
| 1079 | // |
| 1080 | // When another operation's fence already holds the host (the residence |
| 1081 | // move's cut), the pass completes the journal under that fence, installing |
| 1082 | // and lifting nothing, provided the host has no live entry: the move's own |
| 1083 | // fence proved or is proving the stop, and the move refuses a host while a |
| 1084 | // journal exists, so the journal has to be finished for the move to go on. |
| 1085 | // Without a fence of its own the pass cannot converge the host, so unless |
| 1086 | // the completed endpoint is already the intent it keeps the journal at |
| 1087 | // done. The journal then keeps the host fenced if the move aborts and |
| 1088 | // lifts its fence, and the next recovery converges it. |
| 1089 | func (p *Pass) recoverHost(ctx context.Context, host string, byHost map[string]registry.Route, intent Mode, gate func() error) Result { |
| 1090 | if !controlname.ValidHostShape(host) { |
| 1091 | return Result{host, "refused", "journal file name is not a host; not acting"} |
| 1092 | } |
| 1093 | j, found, err := ReadJournal(p.DataDir, host) |
| 1094 | if err != nil { |
| 1095 | return Result{host, "refused", notActing(err)} |
| 1096 | } |
| 1097 | if !found { |
| 1098 | return Result{host, "unchanged", ""} |
| 1099 | } |
| 1100 | rt, ok := byHost[host] |
| 1101 | if !ok { |
| 1102 | return Result{host, "refused", "journal names a host with no row; not acting"} |
| 1103 | } |
| 1104 | // Checked before the stop as well as after it, so a journal that |
| 1105 | // cannot be honoured never costs its host a stop. |
| 1106 | if _, err := Validate(j, p.DataDir, p.socketDir(), rt); err != nil { |
| 1107 | return Result{host, "refused", notActing(err)} |
| 1108 | } |
| 1109 | plan, err := PlanFor(p.DataDir, p.socketDir(), host, j.Mode) |
| 1110 | if err != nil { |
| 1111 | return Result{host, "refused", notActing(err)} |
| 1112 | } |
| 1113 | plan.Logf = p.logf |
| 1114 | |
| 1115 | hosts := make([]string, 0, len(byHost)) |
| 1116 | for h := range byHost { |
| 1117 | hosts = append(hosts, h) |
| 1118 | } |
| 1119 | if err := applyJournal(plan, j, hosts); err != nil { |
| 1120 | return Result{host, "refused", notActing(err)} |
| 1121 | } |
| 1122 | keepFence := false |
| 1123 | r, fence, gen := p.quiesce(ctx, host, p.generation()) |
| 1124 | switch fence { |
| 1125 | case fenceNone: |
| 1126 | return r |
| 1127 | case fenceHeld: |
| 1128 | if !p.Fence.Quiesced(host) { |
| 1129 | return Result{host, "deferred", "fenced by another operation and not yet stopped"} |
| 1130 | } |
| 1131 | case fenceOurs: |
| 1132 | defer func() { |
| 1133 | if !keepFence { |
| 1134 | p.release(host, gen) |
| 1135 | } |
| 1136 | }() |
| 1137 | } |
| 1138 | // The row was read before the host stopped; check the journal against |
| 1139 | // the row as it stands now that nothing else can wake it. |
| 1140 | fresh, err := p.Registry.Get(host) |
| 1141 | if err != nil { |
| 1142 | return Result{host, "refused", notActing(fmt.Errorf("row: %w", err))} |
| 1143 | } |
| 1144 | if fresh == nil { |
| 1145 | return Result{host, "refused", "row vanished under the fence; not acting"} |
| 1146 | } |
| 1147 | atFrom, err := Validate(j, p.DataDir, p.socketDir(), *fresh) |
| 1148 | if err != nil { |
| 1149 | return Result{host, "refused", notActing(err)} |
| 1150 | } |
| 1151 | converge := fence == fenceOurs && (intent == ModeNested || intent == ModeFlat) && intent != j.Mode |
| 1152 | var holdReason string |
| 1153 | switch { |
| 1154 | case fence == fenceHeld && intent != j.Mode: |
| 1155 | // Only an instance move shares the fence slot, and its cut waits |
| 1156 | // for this journal to go while the pass waits for the move's |
| 1157 | // fence to lift. Neither goes on alone, so the reason names both |
| 1158 | // ways out for whoever reads the agent log. |
| 1159 | holdReason = fmt.Sprintf("completed to %s under an instance move's fence; journal kept at done. the move waits for this journal and the journal waits for the move. abort the move, or remove %s now that the row stands at its destination, and the move can finish", j.Mode, provision.LayoutJournalPath(p.DataDir, host)) |
| 1160 | case intent == IntentUnreadable && j.Mode == ModeFlat: |
| 1161 | // Releasing a completed revert flat before the intent is known is |
| 1162 | // the forged-revert window again. |
| 1163 | holdReason = "completed to flat; journal kept at done until the layout intent can be read" |
| 1164 | } |
| 1165 | remove := holdReason == "" && !converge |
| 1166 | var res Result |
| 1167 | p.serialize(func() { res = p.complete(plan, j, atFrom, remove) }) |
| 1168 | if res.Outcome != "recovered" { |
| 1169 | return res |
| 1170 | } |
| 1171 | // The row flipped (or was found flipped). A host still held by its |
| 1172 | // journal is refused wakes by that, but routing should say where it is. |
| 1173 | p.reload() |
| 1174 | if holdReason != "" { |
| 1175 | p.logf("layout: %s: held -> %s", host, holdReason) |
| 1176 | return Result{host, "deferred", holdReason} |
| 1177 | } |
| 1178 | if !converge { |
| 1179 | return res |
| 1180 | } |
| 1181 | if intent == ModeNested { |
| 1182 | if gerr := gate(); gerr != nil { |
| 1183 | // Held, not released flat: there is no degraded mode (design |
| 1184 | // §8.5), and a host released at a forged revert's endpoint is |
| 1185 | // a nested tenant running flat. The journal at done keeps it |
| 1186 | // fenced, so the in-memory fence is lifted as usual, and the |
| 1187 | // first recovery after the probe passes converges it. |
| 1188 | reason := fmt.Sprintf("completed to %s; journal kept at done: not converging to nested: isolation not ready: %v", j.Mode, gerr) |
| 1189 | p.logf("layout: %s: held -> %s", host, reason) |
| 1190 | return Result{host, "deferred", reason} |
| 1191 | } |
| 1192 | } |
| 1193 | next, h, _ := p.shift(host, intent, j) |
| 1194 | switch { |
| 1195 | case next.Outcome == "migrated" || next.Outcome == "reverted": |
| 1196 | return Result{host, "recovered", fmt.Sprintf("completed to %s, then converged to %s", j.Mode, intent)} |
| 1197 | case h == handoverFailed: |
| 1198 | // Nothing on disk is known to fence the host, so the pass's own |
| 1199 | // fence stays and the host stays down. Releasing it here would run |
| 1200 | // it at the completed endpoint, which after a forged revert is a |
| 1201 | // nested host running flat. The generation is recorded so the next |
| 1202 | // pass knows the fence as its own and runs the leg again. |
| 1203 | keepFence = true |
| 1204 | p.keep(host, gen) |
| 1205 | reason := fmt.Sprintf("completed to %s; handing over to the move to %s failed: %s. the fence is held by this pass after a failed handover; the next pass retries it", j.Mode, intent, next.Reason) |
| 1206 | p.logf("layout: %s: refused -> %s", host, reason) |
| 1207 | return Result{host, "refused", reason} |
| 1208 | case h == handoverNone: |
| 1209 | reason := fmt.Sprintf("completed to %s; journal kept at done: converging to %s: %s %s", j.Mode, intent, next.Outcome, next.Reason) |
| 1210 | p.logf("layout: %s: held -> %s", host, reason) |
| 1211 | return Result{host, "deferred", reason} |
| 1212 | } |
| 1213 | return Result{host, next.Outcome, fmt.Sprintf("completed to %s; converging to %s: %s", j.Mode, intent, next.Reason)} |
| 1214 | } |
| 1215 | |
| 1216 | // complete runs the journal's remaining phases (design §8.2's recovery |
| 1217 | // list), and removes the journal only when remove is set; otherwise it |
| 1218 | // stops at done and the journal keeps holding the host. |
| 1219 | // |
| 1220 | // Two states a crash cannot produce are refused rather than acted on. An |
| 1221 | // artifact the journal records absent that exists at either end: the host |
| 1222 | // has been fenced since the plan was made, so nothing recorded absent can |
| 1223 | // have appeared, and moving the rest without it would leave it behind (a |
| 1224 | // forged journal can use exactly that to strand a database file). And a |
| 1225 | // done journal whose row is still at from: done is written only after the |
| 1226 | // row's swap committed, and releasing it would leave a row that names the |
| 1227 | // files' old place. A journal at row is also checked for every artifact |
| 1228 | // it records present being at its destination before the row is swapped. |
| 1229 | func (p *Pass) complete(plan *Plan, j *Journal, atFrom, remove bool) Result { |
| 1230 | host, phase := plan.Host, j.Phase |
| 1231 | for _, a := range plan.Artifacts { |
| 1232 | if a.Name == "socket" || a.Present { |
| 1233 | continue |
| 1234 | } |
| 1235 | for _, path := range []string{a.Src, a.Dst} { |
| 1236 | present, err := lstatPresent(path) |
| 1237 | if err != nil { |
| 1238 | return Result{host, "refused", notActing(err)} |
| 1239 | } |
| 1240 | if present { |
| 1241 | return Result{host, "refused", fmt.Sprintf("journal records %s absent, but it exists at %s; not acting", a.Name, path)} |
| 1242 | } |
| 1243 | } |
| 1244 | } |
| 1245 | switch j.Phase { |
| 1246 | case PhaseFiles: |
| 1247 | if err := p.moveFiles(plan); err != nil { |
| 1248 | return Result{host, "refused", notActing(err)} |
| 1249 | } |
| 1250 | // The same check a first attempt makes before its row flips; the |
| 1251 | // plan carries the journal's presence (applyJournal) so the check |
| 1252 | // knows which files it moved. |
| 1253 | moved, err := VerifyMoved(plan) |
| 1254 | if len(moved) > 0 { |
| 1255 | p.logf("layout: %s: quarantined on recovery -> %s", host, strings.Join(moved, ", ")) |
| 1256 | p.note(host, "quarantined "+strings.Join(moved, ", ")) |
| 1257 | } |
| 1258 | if err != nil { |
| 1259 | return Result{host, "refused", notActing(err)} |
| 1260 | } |
| 1261 | if err := UpdatePhase(p.DataDir, j, PhaseRow); err != nil { |
| 1262 | return Result{host, "refused", err.Error()} |
| 1263 | } |
| 1264 | fallthrough |
| 1265 | case PhaseRow: |
| 1266 | // The row is about to name the destinations, so every file the |
| 1267 | // journal records present must be there: a row pointed at a |
| 1268 | // missing database has the next wake restore a second one. |
| 1269 | for _, a := range plan.Artifacts { |
| 1270 | if a.Name == "socket" || !a.Present { |
| 1271 | continue |
| 1272 | } |
| 1273 | present, err := lstatPresent(a.Dst) |
| 1274 | if err != nil { |
| 1275 | return Result{host, "refused", notActing(err)} |
| 1276 | } |
| 1277 | if !present { |
| 1278 | return Result{host, "refused", fmt.Sprintf("journal records %s present, but it is missing at %s; not acting", a.Name, a.Dst)} |
| 1279 | } |
| 1280 | } |
| 1281 | if atFrom { |
| 1282 | if err := p.Registry.SetInstancePaths(host, j.From.DB, j.From.Socket, j.To.DB, j.To.Socket); err != nil { |
| 1283 | return Result{host, "refused", notActing(fmt.Errorf("row: %w", err))} |
| 1284 | } |
| 1285 | } |
| 1286 | if err := UpdatePhase(p.DataDir, j, PhaseDone); err != nil { |
| 1287 | return Result{host, "refused", err.Error()} |
| 1288 | } |
| 1289 | case PhaseDone: |
| 1290 | if atFrom { |
| 1291 | return Result{host, "refused", "journal is done but the row is still at its source; not acting"} |
| 1292 | } |
| 1293 | default: |
| 1294 | return Result{host, "refused", fmt.Sprintf("journal phase %q unknown; not acting", j.Phase)} |
| 1295 | } |
| 1296 | if remove { |
| 1297 | if err := RemoveJournal(p.DataDir, host); err != nil { |
| 1298 | return Result{host, "refused", err.Error()} |
| 1299 | } |
| 1300 | } |
| 1301 | p.logf("layout: %s: recovered -> %s (journal phase was %s)", host, j.Mode, phase) |
| 1302 | return Result{host, "recovered", string(j.Mode)} |
| 1303 | } |
| 1304 | |
amadan