| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "container/list" |
| 5 | "context" |
| 6 | "fmt" |
| 7 | "log" |
| 8 | "math" |
| 9 | "strconv" |
| 10 | "sync" |
| 11 | |
| 12 | "github.com/carlosframework/platform/internal/activator" |
| 13 | "github.com/carlosframework/platform/internal/hostclaim" |
| 14 | "github.com/carlosframework/platform/internal/moves" |
| 15 | "github.com/carlosframework/platform/internal/provision" |
| 16 | "github.com/carlosframework/platform/internal/release" |
| 17 | ) |
| 18 | |
| 19 | // The move log and the member stream (spec §13, Plan 6 Task 3). |
| 20 | // |
| 21 | // Every record write a box makes for a move goes through observedMove, which |
| 22 | // keeps the record as the write found it; logTransition names what the write |
| 23 | // did (moves.TransitionOf, which the console's move pass shares): one event, |
| 24 | // one identity. Transitions another writer made (the console's request, |
| 25 | // manual cut, abort and retry; a declared loss) are seen by the box's own |
| 26 | // pass (observeMove) and named the same way. A recovery probe's starts are |
| 27 | // the attempt journal's (movesProbeAttempts.RecoveryAttempt, |
| 28 | // recoveryAttemptLogger). The identity is the log object's key, written |
| 29 | // create-only behind moveCfg.LogEvent (moveLogWriter in production), and |
| 30 | // only the writer that created the object tells the app's members (preflight |
| 31 | // L9, L27). |
| 32 | // |
| 33 | // A write that fails is kept, with the entry's exact bytes, in the worker's |
| 34 | // outbox on disk (moveslogoutbox.go), and every pass retries the outbox |
| 35 | // whatever the record now says, until the entry is created or found there; |
| 36 | // so neither a phase that moves on nor a close loses it (fix round 2, |
| 37 | // ruling 1). |
| 38 | // |
| 39 | // A worker remembers the identities it has handed off (written, found, or |
| 40 | // kept in the outbox), so a pass that sees the same transition again does |
| 41 | // not even try. The memory is bounded (moveLogMemoCap, oldest dropped), and |
| 42 | // a move is forgotten once its terminal outcome is logged or a pass sees its |
| 43 | // record with no move or another move (retireMoves): create-only storage |
| 44 | // still refuses a replay. |
| 45 | |
| 46 | // observedMove wraps a record mutation so *pre keeps the record as the write |
| 47 | // found it: the transition's before, for the log. |
| 48 | func observedMove(pre *provision.Record, mut func(*provision.Record) error) func(*provision.Record) error { |
| 49 | return func(rec *provision.Record) error { |
| 50 | *pre = *rec |
| 51 | pre.Move = copyMove(rec.Move) |
| 52 | return mut(rec) |
| 53 | } |
| 54 | } |
| 55 | |
| 56 | // copyMove is a deep enough copy of a move for moves.TransitionOf: the fields |
| 57 | // it compares and reads, safe from a mutation that edits the journal in |
| 58 | // place. |
| 59 | func copyMove(m *provision.MoveState) *provision.MoveState { |
| 60 | if m == nil { |
| 61 | return nil |
| 62 | } |
| 63 | c := *m |
| 64 | if m.Rollback != nil { |
| 65 | rb := *m.Rollback |
| 66 | c.Rollback = &rb |
| 67 | } |
| 68 | if m.Landing != nil { |
| 69 | l := *m.Landing |
| 70 | c.Landing = &l |
| 71 | } |
| 72 | if m.Source != nil { |
| 73 | src := *m.Source |
| 74 | c.Source = &src |
| 75 | } |
| 76 | return &c |
| 77 | } |
| 78 | |
| 79 | // logTransition logs the transition one write made. |
| 80 | func (w *movesWorker) logTransition(ctx context.Context, s release.Store, before, after provision.Record) { |
| 81 | if e, ok := moves.TransitionOf(before, after); ok { |
| 82 | w.record(ctx, s, after, e) |
| 83 | } |
| 84 | } |
| 85 | |
| 86 | // observeMove logs the transitions another writer made that this box's pass |
| 87 | // sees: the console's request, manual cut, abort and retry, and a declared |
| 88 | // loss. Each is named as if it were this box's write from a record with no |
| 89 | // move (or, for a retry, by its epoch), so it has the same identity whoever |
| 90 | // logs it. |
| 91 | func (w *movesWorker) observeMove(ctx context.Context, s release.Store, rec provision.Record) { |
| 92 | w.retireMoves(rec) |
| 93 | m := rec.Move |
| 94 | if m == nil || m.ID == "" { |
| 95 | return |
| 96 | } |
| 97 | switch { |
| 98 | case m.State == provision.MovePreparing, m.State == provision.MoveAborting, |
| 99 | m.State == provision.MoveCutting && m.CutPhase == provision.CutRequested: |
| 100 | w.logTransition(ctx, s, provision.Record{}, rec) |
| 101 | case m.State == provision.MoveRollingBack && m.Rollback != nil && m.Rollback.TargetPhase == provision.RbTargetLost: |
| 102 | // The console that performed the declaration logs it first (final |
| 103 | // review I4); this is the second producer of the same identity. |
| 104 | if e, ok := moves.DeclaredLost(m); ok { |
| 105 | w.record(ctx, s, rec, e) |
| 106 | } |
| 107 | } |
| 108 | if m.RecoveryRetryEpoch > 0 && m.State != provision.MoveStuck { |
| 109 | w.record(ctx, s, rec, moves.Transition{Op: m.ID, Event: "retry", Identity: "retry-" + strconv.FormatUint(m.RecoveryRetryEpoch, 10), |
| 110 | Fields: map[string]string{"from": m.From, "to": m.To}}) |
| 111 | } |
| 112 | } |
| 113 | |
| 114 | // logRevPhase logs a revert's step, which is the claim's rev_phase, not the |
| 115 | // record's: reverting-<phase>, event rev_<phase>. |
| 116 | func (w *movesWorker) logRevPhase(ctx context.Context, s release.Store, rec provision.Record, before, after hostclaim.Observed) { |
| 117 | phase := func(o hostclaim.Observed) string { |
| 118 | if o.Claim == nil { |
| 119 | return "" |
| 120 | } |
| 121 | return o.Claim.Op.Journal[hostclaim.JournalRevPhase] |
| 122 | } |
| 123 | p := phase(after) |
| 124 | if p == "" || p == phase(before) || rec.Move == nil || rec.Move.ID == "" { |
| 125 | return |
| 126 | } |
| 127 | w.record(ctx, s, rec, moves.Transition{Op: rec.Move.ID, Event: "rev_" + p, Identity: "reverting-" + p, |
| 128 | Fields: map[string]string{"from": rec.Move.From, "to": rec.Move.To}}) |
| 129 | } |
| 130 | |
| 131 | // recoveryAttemptLogger is the attempt journal's RecoveryAttempt hook for |
| 132 | // this worker (set by probeAttempts): each recovery probe start the journal |
| 133 | // counts is probe_attempt-<epoch>-<n> (preflight L11), the starts of one wake |
| 134 | // included. s is the store the log is written to. A write that fails goes to |
| 135 | // the outbox like any other. |
| 136 | func (w *movesWorker) recoveryAttemptLogger(s release.Store) func(ctx context.Context, rec provision.Record, epoch uint64, n int) { |
| 137 | return func(ctx context.Context, rec provision.Record, epoch uint64, n int) { |
| 138 | if rec.Move == nil || rec.Move.ID == "" { |
| 139 | return |
| 140 | } |
| 141 | w.record(ctx, s, rec, moves.Transition{Op: rec.Move.ID, Event: "probe_attempt", |
| 142 | Identity: "probe_attempt-" + strconv.FormatUint(epoch, 10) + "-" + strconv.Itoa(n), |
| 143 | Fields: map[string]string{"from": rec.Move.From, "to": rec.Move.To}}) |
| 144 | } |
| 145 | } |
| 146 | |
| 147 | // terminalEvents are the events that end a move's log. |
| 148 | var terminalEvents = map[string]bool{"done": true, "rolled_back": true, "aborted": true} |
| 149 | |
| 150 | // record hands one entry to the LogEvent hook unless this worker has |
| 151 | // already handed it off. An entry whose write fails is kept in the outbox, |
| 152 | // for every later pass to retry; only an entry written, found there or kept |
| 153 | // is remembered. With no hook (a test's worker) the event is a log line. |
| 154 | func (w *movesWorker) record(ctx context.Context, s release.Store, rec provision.Record, e moves.Transition) { |
| 155 | key := rec.Host + "/" + e.Op + "/" + e.Identity |
| 156 | if w.logged.seen(key) { |
| 157 | return |
| 158 | } |
| 159 | ev := moves.LogEvent{V: 1, Op: e.Op, Host: rec.Host, Event: e.Event, Identity: e.Identity, Actor: w.cfg.Box, Fields: e.Fields} |
| 160 | if w.cfg.Now != nil { |
| 161 | ev.At = w.cfg.Now().UTC() |
| 162 | } |
| 163 | if f := w.cfg.LogEvent; f != nil { |
| 164 | if err := f(ctx, s, rec, ev); err != nil { |
| 165 | if qerr := w.queueLog(rec, ev); qerr != nil { |
| 166 | log.Printf("moves: %s: the log of %s at %s waits for the next sighting: %v; the outbox: %v", rec.Host, e.Op, e.Identity, err, qerr) |
| 167 | return |
| 168 | } |
| 169 | log.Printf("moves: %s: the log of %s at %s is kept for the next pass: %v", rec.Host, e.Op, e.Identity, err) |
| 170 | } |
| 171 | } else { |
| 172 | log.Printf("moves: %s: %s %s (%s)", rec.Host, e.Op, e.Event, e.Identity) |
| 173 | } |
| 174 | if terminalEvents[e.Event] { |
| 175 | w.logged.forget(rec.Host, e.Op) |
| 176 | return |
| 177 | } |
| 178 | w.logged.add(rec.Host, e.Op, key) |
| 179 | } |
| 180 | |
| 181 | // retireMoves forgets the moves of rec's host other than the one its record |
| 182 | // names now: a move closed, or replaced, by another writer. |
| 183 | func (w *movesWorker) retireMoves(rec provision.Record) { |
| 184 | keep := "" |
| 185 | if rec.Move != nil { |
| 186 | keep = rec.Move.ID |
| 187 | } |
| 188 | w.logged.retire(rec.Host, keep) |
| 189 | } |
| 190 | |
| 191 | // moveLogMemoCap bounds a worker's memory of handed-off identities. |
| 192 | const moveLogMemoCap = 4096 |
| 193 | |
| 194 | // moveLogMemo is the identities a worker has handed off: at most |
| 195 | // moveLogMemoCap, the oldest dropped first, indexed by host and move so a |
| 196 | // move is forgotten whole. The zero value is ready. |
| 197 | type moveLogMemo struct { |
| 198 | mu sync.Mutex |
| 199 | order *list.List // memoEntry, oldest first |
| 200 | at map[string]*list.Element // key to its element |
| 201 | moves map[string]map[string][]*list.Element // host to move to its elements |
| 202 | } |
| 203 | |
| 204 | type memoEntry struct{ host, op, key string } |
| 205 | |
| 206 | func (m *moveLogMemo) seen(key string) bool { |
| 207 | m.mu.Lock() |
| 208 | defer m.mu.Unlock() |
| 209 | _, ok := m.at[key] |
| 210 | return ok |
| 211 | } |
| 212 | |
| 213 | func (m *moveLogMemo) len() int { |
| 214 | m.mu.Lock() |
| 215 | defer m.mu.Unlock() |
| 216 | return len(m.at) |
| 217 | } |
| 218 | |
| 219 | func (m *moveLogMemo) add(host, op, key string) { |
| 220 | m.mu.Lock() |
| 221 | defer m.mu.Unlock() |
| 222 | if m.order == nil { |
| 223 | m.order, m.at, m.moves = list.New(), map[string]*list.Element{}, map[string]map[string][]*list.Element{} |
| 224 | } |
| 225 | if _, ok := m.at[key]; ok { |
| 226 | return |
| 227 | } |
| 228 | for len(m.at) >= moveLogMemoCap { |
| 229 | m.drop(m.order.Front()) |
| 230 | } |
| 231 | el := m.order.PushBack(memoEntry{host, op, key}) |
| 232 | m.at[key] = el |
| 233 | if m.moves[host] == nil { |
| 234 | m.moves[host] = map[string][]*list.Element{} |
| 235 | } |
| 236 | m.moves[host][op] = append(m.moves[host][op], el) |
| 237 | } |
| 238 | |
| 239 | // drop removes one element from every index. |
| 240 | func (m *moveLogMemo) drop(el *list.Element) { |
| 241 | e := el.Value.(memoEntry) |
| 242 | m.order.Remove(el) |
| 243 | delete(m.at, e.key) |
| 244 | els := m.moves[e.host][e.op] |
| 245 | for i, x := range els { |
| 246 | if x == el { |
| 247 | els = append(els[:i], els[i+1:]...) |
| 248 | break |
| 249 | } |
| 250 | } |
| 251 | if len(els) > 0 { |
| 252 | m.moves[e.host][e.op] = els |
| 253 | return |
| 254 | } |
| 255 | delete(m.moves[e.host], e.op) |
| 256 | if len(m.moves[e.host]) == 0 { |
| 257 | delete(m.moves, e.host) |
| 258 | } |
| 259 | } |
| 260 | |
| 261 | // forget drops one move's identities. |
| 262 | func (m *moveLogMemo) forget(host, op string) { |
| 263 | m.mu.Lock() |
| 264 | defer m.mu.Unlock() |
| 265 | for _, el := range append([]*list.Element(nil), m.moves[host][op]...) { |
| 266 | m.drop(el) |
| 267 | } |
| 268 | } |
| 269 | |
| 270 | // retire drops the identities of host's moves other than keep. |
| 271 | func (m *moveLogMemo) retire(host, keep string) { |
| 272 | m.mu.Lock() |
| 273 | defer m.mu.Unlock() |
| 274 | for op, els := range m.moves[host] { |
| 275 | if op == keep { |
| 276 | continue |
| 277 | } |
| 278 | for _, el := range append([]*list.Element(nil), els...) { |
| 279 | m.drop(el) |
| 280 | } |
| 281 | } |
| 282 | } |
| 283 | |
| 284 | // moveLogWriter is the production LogEvent: the object, create-only, under |
| 285 | // the move's report prefix in the deployment bucket, and the member event |
| 286 | // when this writer created it. A storage failure is returned, so the worker |
| 287 | // tries the transition again at its next sighting. |
| 288 | func moveLogWriter(sink activator.EventSink) func(context.Context, release.Store, provision.Record, moves.LogEvent) error { |
| 289 | return func(ctx context.Context, s release.Store, rec provision.Record, ev moves.LogEvent) error { |
| 290 | if s == nil { |
| 291 | return fmt.Errorf("moves: no store for the log of %s", ev.Op) |
| 292 | } |
| 293 | created, err := moves.WriteLog(ctx, s, moves.ReportPrefix(rec.Account, rec.App, rec.Host, ev.Op), ev) |
| 294 | if err != nil { |
| 295 | return err |
| 296 | } |
| 297 | if !created || sink == nil { |
| 298 | return nil |
| 299 | } |
| 300 | if line, ok := memberLine(ev, rec.Host); ok { |
| 301 | sink.Event(rec.Account, rec.App, rec.Host, memberLevel(ev.Event), line) |
| 302 | } |
| 303 | return nil |
| 304 | } |
| 305 | } |
| 306 | |
| 307 | // memberLine is what the app's members read for an event (preflight L9: M-C1, |
| 308 | // M-C5, M-C6, M-C7, M-C8, M-C9, and for stuck MV-N380, a line with no |
| 309 | // operator command in it), from the shared templates (moves/operatorcopy.go, |
| 310 | // L12), built from the entry alone, so an entry the outbox writes later says |
| 311 | // the same. A region the record names as the home's "" is |
| 312 | // moves.HomeRegionText. ok is false for an event members are not told of. |
| 313 | // Its level is memberLevel's. |
| 314 | func memberLine(ev moves.LogEvent, host string) (line string, ok bool) { |
| 315 | from, to := moves.RegionText(ev.Fields["from"]), moves.RegionText(ev.Fields["to"]) |
| 316 | switch ev.Event { |
| 317 | case "requested": |
| 318 | return moves.MoveStartingLine(host, from, to), true |
| 319 | case "admitted": |
| 320 | budget, err := strconv.Atoi(ev.Fields["cut_budget_s"]) |
| 321 | if err != nil || budget <= 0 { |
| 322 | return "", false |
| 323 | } |
| 324 | return moves.CutStartingLine(to, host, moves.CutBudgetText(budget)), true |
| 325 | case "flipped": |
| 326 | return moves.FlippedLine(from, to), true |
| 327 | case "done": |
| 328 | offline, err := strconv.ParseFloat(ev.Fields["offline_s"], 64) |
| 329 | if err != nil { |
| 330 | return "", false |
| 331 | } |
| 332 | return moves.MovedLine(host, to, int(math.Round(offline))), true |
| 333 | case "rolling_back": |
| 334 | reason := ev.Fields["reason"] |
| 335 | if reason == "" { |
| 336 | return "", false |
| 337 | } |
| 338 | return moves.MovingBackLine(to, host, reason, from), true |
| 339 | case "rolled_back": |
| 340 | return moves.MovedBackLine(host, from), true |
| 341 | case "stuck": |
| 342 | return moves.StuckMemberLine(host, to, from), true |
| 343 | } |
| 344 | return "", false |
| 345 | } |
| 346 | |
| 347 | // memberLevel is the level of an event's member line. |
| 348 | func memberLevel(event string) string { |
| 349 | switch event { |
| 350 | case "rolling_back", "rolled_back": |
| 351 | return "warn" |
| 352 | case "stuck": |
| 353 | return "error" |
| 354 | } |
| 355 | return "info" |
| 356 | } |
| 357 | |
amadan