| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "fmt" |
| 8 | "io/fs" |
| 9 | "log" |
| 10 | "os" |
| 11 | "path/filepath" |
| 12 | "strconv" |
| 13 | "strings" |
| 14 | "sync" |
| 15 | "time" |
| 16 | |
| 17 | "github.com/carlosframework/platform/internal/alerting" |
| 18 | "github.com/carlosframework/platform/internal/moves" |
| 19 | "github.com/carlosframework/platform/internal/provision" |
| 20 | ) |
| 21 | |
| 22 | // Paging for moves from the box (spec §13; Plan 6 Task 5; preflight L8). The |
| 23 | // box pages, through moveCfg.Page (PageOperator, the agent's alert sinks; |
| 24 | // Plan 5 Task 0), only what it alone sees: |
| 25 | // |
| 26 | // - stuck, an incident, once per entry into stuck (pageStuck): the rule is |
| 27 | // stuck-<epoch>, the record's recovery retry epoch, which only an |
| 28 | // operator's retry advances, so a retried move that sticks again pages |
| 29 | // again and a replay of the same mark does not; |
| 30 | // - rolled_back, a warning, once per move, where this box closes a |
| 31 | // rollback (recoverRun.finalize) or a revert (revertRun.close); |
| 32 | // - a stop evidence refused for a process the move cannot account for |
| 33 | // (StepStray), an incident, once per move, from the cut's stopping: the |
| 34 | // page's body is M-C20, and the agent's log says M-C20 when the box |
| 35 | // detects it. M-C20 is operator copy: the app's members are never told |
| 36 | // (fix round 1, ruling 1); |
| 37 | // - a refused start check (moves.StartCheck), an incident, once per move, |
| 38 | // from the probe owners (movesProbeOwners.startCheck). |
| 39 | // |
| 40 | // The console's move pass (internal/console/movesrecovery.go, Plan 5 Task 7) |
| 41 | // keeps the rules whose facts it holds, and the box does not repeat them: |
| 42 | // rolling_back past release_deadline (an incident, once an hour), preparing |
| 43 | // with no promotable round for 60 minutes (a warning, once, measured from |
| 44 | // moves.PreparingSince, the six-hour abort's own start), an orphaned claim, |
| 45 | // an overdue close and an overdue abort. A move converted to relocating with |
| 46 | // no live box in its region (M-C22) is Plan 5 Task 7a's page. |
| 47 | // |
| 48 | // The page outbox (fix round 2, ruling 1). A page is one file, |
| 49 | // <DataDir>/.moves/.page-outbox/<host>/<move>/<rule>.json, created |
| 50 | // exclusively (outboxCreate, the move log's outbox's own helper) when its |
| 51 | // rule first fires: the file being there is the rule's once-per-entry limit, |
| 52 | // so a rule that fires again never pages again. The entry carries the page |
| 53 | // and its backoff. It is sent when created; one the pager refused is sent |
| 54 | // again by every moves pass (flushPageOutbox, first thing in the pass, before |
| 55 | // any record or fleet read) once its backoff is over (pageRetryFirst, |
| 56 | // doubling, up to pageRetryMax), until it is delivered, whatever the move |
| 57 | // and its journals have done since. A delivered entry stays as a tombstone, |
| 58 | // so a rule that fires again still finds it, until the move is retired on |
| 59 | // this box: no listed record carries it and its journal is gone |
| 60 | // (retirePageTombstones, only after a pass listed every record). The outbox |
| 61 | // depends on nothing in the move's journal. |
| 62 | // |
| 63 | // An index in memory (pageIndex, fix round 3) answers for what the disk |
| 64 | // cannot: a firing looks at the disk entry, then the index, and only a page |
| 65 | // in neither is created; only when that create fails is the page sent at |
| 66 | // once, kept in the index, and, if that fails too, sent again by every pass |
| 67 | // with a loud line in the agent's log. A delivery or a backoff the disk |
| 68 | // cannot record is kept in the index too, and a delivered page's tombstone |
| 69 | // is written again every pass until it lands. Only a restart while the index |
| 70 | // holds what the disk does not loses it. |
| 71 | // |
| 72 | // Unwired (nil Page, as production is until Plan 5 Task 9b wires it), a rule |
| 73 | // is never silent: the agent's log says so loudly, once per rule per move |
| 74 | // for as long as the agent runs, and nothing is refused or kept (fix round 1, |
| 75 | // ruling 2). |
| 76 | |
| 77 | // pagingLogf is the agent's log for the paging code (a variable so tests |
| 78 | // can read it). |
| 79 | var pagingLogf = log.Printf |
| 80 | |
| 81 | // pageFunc is the pager: moveCfg.Page's shape. |
| 82 | type pageFunc = func(ctx context.Context, level, subject, body string) error |
| 83 | |
| 84 | // The rules' names in the outbox. |
| 85 | const ( |
| 86 | pageRuleRolledBack = "rolled_back" |
| 87 | pageRuleStray = "stray" |
| 88 | pageRuleStartCheck = "start_check" |
| 89 | ) |
| 90 | |
| 91 | // pageRuleStuck is the stuck rule of one entry into stuck: its retry epoch. |
| 92 | func pageRuleStuck(m *provision.MoveState) string { |
| 93 | return "stuck-" + strconv.FormatUint(m.RecoveryRetryEpoch, 10) |
| 94 | } |
| 95 | |
| 96 | // pageOutcome is what became of one rule's firing. |
| 97 | type pageOutcome int |
| 98 | |
| 99 | const ( |
| 100 | pageSent pageOutcome = iota // handed to the pager without error |
| 101 | pagePagedBefore // the outbox says it was delivered before |
| 102 | pageUnwired // no pager is wired (logged loudly once) |
| 103 | pageUndelivered // first firing, refused: kept to send again |
| 104 | pagePending // fired before, not delivered yet: the passes send it again |
| 105 | ) |
| 106 | |
| 107 | // String is the outcome as the agent's log says it. |
| 108 | func (o pageOutcome) String() string { |
| 109 | switch o { |
| 110 | case pageSent: |
| 111 | return "paged" |
| 112 | case pagePagedBefore: |
| 113 | return "paged before" |
| 114 | case pageUnwired: |
| 115 | return "not paged: no pager is wired" |
| 116 | case pagePending: |
| 117 | return "not paged yet: it is sent again after a backoff" |
| 118 | } |
| 119 | return "not paged: the page was not delivered, and is sent again after a backoff" |
| 120 | } |
| 121 | |
| 122 | // A refused page is tried again pageRetryFirst after its first attempt, and |
| 123 | // after each further attempt twice as long again, up to pageRetryMax. |
| 124 | const ( |
| 125 | pageRetryFirst = time.Minute |
| 126 | pageRetryMax = 30 * time.Minute |
| 127 | ) |
| 128 | |
| 129 | // pageBackoff is the wait after the tries-th attempt. |
| 130 | func pageBackoff(tries int) time.Duration { |
| 131 | if tries < 1 { |
| 132 | tries = 1 |
| 133 | } |
| 134 | if tries > 6 { |
| 135 | return pageRetryMax |
| 136 | } |
| 137 | if d := pageRetryFirst << (tries - 1); d < pageRetryMax { |
| 138 | return d |
| 139 | } |
| 140 | return pageRetryMax |
| 141 | } |
| 142 | |
| 143 | // pageEntry is one page in the outbox: the page, the attempts so far, when |
| 144 | // the next may start, and when it was delivered. |
| 145 | type pageEntry struct { |
| 146 | V int `json:"v"` |
| 147 | Host string `json:"host"` |
| 148 | Op string `json:"op"` |
| 149 | Rule string `json:"rule"` |
| 150 | Level string `json:"level"` |
| 151 | Subject string `json:"subject"` |
| 152 | Body string `json:"body"` |
| 153 | At time.Time `json:"at"` |
| 154 | Tries int `json:"tries"` |
| 155 | Next time.Time `json:"next"` |
| 156 | Delivered *time.Time `json:"delivered,omitempty"` |
| 157 | } |
| 158 | |
| 159 | // pageOutboxDir is the page outbox's root. |
| 160 | func pageOutboxDir(dataDir string) string { |
| 161 | return filepath.Join(dataDir, ".moves", ".page-outbox") |
| 162 | } |
| 163 | |
| 164 | // pageEntryDir is the directory of one move's pages. |
| 165 | func pageEntryDir(dataDir, host, op string) string { |
| 166 | return filepath.Join(pageOutboxDir(dataDir), host, op) |
| 167 | } |
| 168 | |
| 169 | // pageOutboxMu serializes every change to an outbox entry, and to the page |
| 170 | // index, in this process (the moves pass and the probe owners both fire |
| 171 | // pages); delivery itself is outside it, after the attempt is counted. |
| 172 | var pageOutboxMu sync.Mutex |
| 173 | |
| 174 | // The page index (fix round 3): what this process knows of a page that the |
| 175 | // disk does not say, keyed by (outbox, host, move, rule). It holds a page |
| 176 | // the outbox could not keep (memory), a backoff deadline the disk could not |
| 177 | // record (next), and a delivery the disk does not record yet (delivered, |
| 178 | // whose tombstone every pass writes again until it lands). A firing looks at |
| 179 | // the disk and then the index, so one (host, move, rule) is one page however |
| 180 | // the storage fails; only a restart while the index holds a page the disk |
| 181 | // does not loses it. |
| 182 | type pageKey struct{ dataDir, host, op, rule string } |
| 183 | |
| 184 | type pageIndexItem struct { |
| 185 | e pageEntry |
| 186 | page pageFunc // the pager that fired a memory page |
| 187 | memory bool // the outbox could not keep it |
| 188 | delivered bool |
| 189 | next time.Time |
| 190 | } |
| 191 | |
| 192 | var pageIndex = map[pageKey]*pageIndexItem{} |
| 193 | |
| 194 | func pageKeyOf(dataDir string, e pageEntry) pageKey { |
| 195 | return pageKey{dataDir, e.Host, e.Op, e.Rule} |
| 196 | } |
| 197 | |
| 198 | // readPageEntry reads one entry. |
| 199 | func readPageEntry(path string) (pageEntry, error) { |
| 200 | var e pageEntry |
| 201 | b, err := os.ReadFile(path) |
| 202 | if err != nil { |
| 203 | return e, err |
| 204 | } |
| 205 | if err := json.Unmarshal(b, &e); err != nil || e.V != 1 { |
| 206 | return e, fmt.Errorf("%s is not a page outbox entry", path) |
| 207 | } |
| 208 | return e, nil |
| 209 | } |
| 210 | |
| 211 | // writePageEntry replaces an entry. |
| 212 | func writePageEntry(dataDir string, e pageEntry) error { |
| 213 | b, err := json.Marshal(e) |
| 214 | if err != nil { |
| 215 | return err |
| 216 | } |
| 217 | return outboxReplace(dataDir, pageEntryDir(dataDir, e.Host, e.Op), e.Rule+".json", b) |
| 218 | } |
| 219 | |
| 220 | // pageEntryPath is an entry's file. |
| 221 | func pageEntryPath(dataDir, host, op, rule string) string { |
| 222 | return filepath.Join(pageEntryDir(dataDir, host, op), rule+".json") |
| 223 | } |
| 224 | |
| 225 | // firePage is a rule firing (fix round 3, ruling 1). The disk entry is |
| 226 | // looked at first: there, delivered or not, the rule sends nothing (the |
| 227 | // passes retry a refused one). Then the index: there, it sends nothing. |
| 228 | // Only a (host, move, rule) in neither is created, its attempt counted, and |
| 229 | // sent; and only when that create fails is the page sent directly, and kept |
| 230 | // in the index if that fails too. Unwired, it warns once per rule per move |
| 231 | // (warned is the memory of that) and keeps nothing. |
| 232 | func firePage(ctx context.Context, page pageFunc, warned *sync.Map, dataDir, host, op, rule string, now time.Time, level, subject, body string) pageOutcome { |
| 233 | if page == nil { |
| 234 | if _, seen := warned.LoadOrStore(host+"/"+op+"/"+rule, true); !seen { |
| 235 | pagingLogf("moves: %s: WARNING: no operator pager is wired on this box, so the %s page of move %s is not sent; the move goes on (Plan 5 Task 9b wires the pager)", host, rule, op) |
| 236 | } |
| 237 | return pageUnwired |
| 238 | } |
| 239 | e := pageEntry{V: 1, Host: host, Op: op, Rule: rule, Level: level, Subject: subject, Body: body, At: now.UTC(), Tries: 1, Next: now.UTC().Add(pageBackoff(1))} |
| 240 | if !outboxSegment(host) || !outboxSegment(op) || !outboxSegment(rule) || dataDir == "" { |
| 241 | pagingLogf("moves: %s: WARNING: the %s page of move %s has no usable outbox name, so it is sent now", host, rule, op) |
| 242 | if err := page(ctx, level, subject, body); err != nil { |
| 243 | pagingLogf("moves: %s: WARNING: the %s page of move %s was not delivered: %v", host, rule, op, err) |
| 244 | return pageUndelivered |
| 245 | } |
| 246 | return pageSent |
| 247 | } |
| 248 | key := pageKeyOf(dataDir, e) |
| 249 | pageOutboxMu.Lock() |
| 250 | if kept, err := readPageEntry(pageEntryPath(dataDir, host, op, rule)); err == nil { |
| 251 | pageOutboxMu.Unlock() |
| 252 | if kept.Delivered != nil { |
| 253 | return pagePagedBefore |
| 254 | } |
| 255 | return pagePending |
| 256 | } else if !errors.Is(err, fs.ErrNotExist) { |
| 257 | if _, serr := os.Lstat(pageEntryPath(dataDir, host, op, rule)); serr == nil { |
| 258 | // There, but not readable now: it is the rule's entry all the same. |
| 259 | pageOutboxMu.Unlock() |
| 260 | return pagePending |
| 261 | } |
| 262 | } |
| 263 | if item := pageIndex[key]; item != nil { |
| 264 | delivered := item.delivered |
| 265 | pageOutboxMu.Unlock() |
| 266 | if delivered { |
| 267 | return pagePagedBefore |
| 268 | } |
| 269 | return pagePending |
| 270 | } |
| 271 | b, err := json.Marshal(e) |
| 272 | created := false |
| 273 | if err == nil { |
| 274 | created, err = outboxCreate(dataDir, pageEntryDir(dataDir, host, op), rule+".json", b) |
| 275 | } |
| 276 | if err == nil && !created { |
| 277 | pageOutboxMu.Unlock() |
| 278 | return pagePending |
| 279 | } |
| 280 | if err != nil { |
| 281 | // The outbox cannot keep it: the index does, before the send, so no |
| 282 | // re-fire sends it again. |
| 283 | item := &pageIndexItem{e: e, page: page, memory: true, next: e.Next} |
| 284 | pageIndex[key] = item |
| 285 | pageOutboxMu.Unlock() |
| 286 | pagingLogf("moves: %s: WARNING: the page outbox cannot keep the %s page of move %s, so it is sent now: %v", host, rule, op, err) |
| 287 | if perr := page(ctx, level, subject, body); perr != nil { |
| 288 | pagingLogf("moves: %s: WARNING: the %s page of move %s was not delivered either (%v); it is kept in memory only and sent again every pass, and is lost if the agent restarts first", host, rule, op, perr) |
| 289 | return pageUndelivered |
| 290 | } |
| 291 | pageOutboxMu.Lock() |
| 292 | item.delivered = true |
| 293 | pageOutboxMu.Unlock() |
| 294 | return pageSent |
| 295 | } |
| 296 | pageOutboxMu.Unlock() |
| 297 | return deliverPageEntry(ctx, page, dataDir, e, now) |
| 298 | } |
| 299 | |
| 300 | // deliverPageEntry sends an entry whose attempt is already counted, and marks |
| 301 | // it delivered once the pager took it: on disk, or, when the disk cannot |
| 302 | // record it, in the index, whose tombstone the passes write again until it |
| 303 | // lands. A refused page stays for the passes. |
| 304 | func deliverPageEntry(ctx context.Context, page pageFunc, dataDir string, e pageEntry, now time.Time) pageOutcome { |
| 305 | if err := page(ctx, e.Level, e.Subject, e.Body); err != nil { |
| 306 | pagingLogf("moves: %s: the %s page of move %s was not delivered (try %d), and is sent again after %s: %v", e.Host, e.Rule, e.Op, e.Tries, pageBackoff(e.Tries), err) |
| 307 | return pageUndelivered |
| 308 | } |
| 309 | pageOutboxMu.Lock() |
| 310 | defer pageOutboxMu.Unlock() |
| 311 | if err := markPageDelivered(dataDir, e, now); err != nil { |
| 312 | pageIndex[pageKeyOf(dataDir, e)] = &pageIndexItem{e: e, delivered: true} |
| 313 | pagingLogf("moves: %s: the %s page of move %s was sent; the outbox cannot say so yet, so this agent remembers it and writes it again every pass: %v", e.Host, e.Rule, e.Op, err) |
| 314 | } |
| 315 | return pageSent |
| 316 | } |
| 317 | |
| 318 | // markPageDelivered writes e's entry as delivered: the entry on disk marked, |
| 319 | // or, when there is none, a tombstone created. The caller holds |
| 320 | // pageOutboxMu. |
| 321 | func markPageDelivered(dataDir string, e pageEntry, now time.Time) error { |
| 322 | at := now.UTC() |
| 323 | kept, err := readPageEntry(pageEntryPath(dataDir, e.Host, e.Op, e.Rule)) |
| 324 | switch { |
| 325 | case err == nil: |
| 326 | if kept.Delivered != nil { |
| 327 | // Marked before, perhaps by a write whose sync failed: it |
| 328 | // counts once it is durable (final review I2). |
| 329 | return outboxSettle(dataDir, pageEntryDir(dataDir, e.Host, e.Op), pageEntryPath(dataDir, e.Host, e.Op, e.Rule)) |
| 330 | } |
| 331 | kept.Delivered = &at |
| 332 | return writePageEntry(dataDir, kept) |
| 333 | case errors.Is(err, fs.ErrNotExist): |
| 334 | t := e |
| 335 | t.Delivered = &at |
| 336 | b, err := json.Marshal(t) |
| 337 | if err != nil { |
| 338 | return err |
| 339 | } |
| 340 | _, err = outboxCreate(dataDir, pageEntryDir(dataDir, e.Host, e.Op), e.Rule+".json", b) |
| 341 | return err |
| 342 | } |
| 343 | return err |
| 344 | } |
| 345 | |
| 346 | // flushPageOutbox is every moves pass's delivery half of paging (fix round 3, |
| 347 | // ruling 4: it runs first, before any record or fleet read, so a bucket that |
| 348 | // cannot be read never holds a page back). The index's pages are sent again |
| 349 | // once their backoff is over and its delivered pages' tombstones written |
| 350 | // again; then each disk entry not yet delivered is sent again once its |
| 351 | // backoff (the later of the disk's and the index's) is over, its attempt |
| 352 | // counted and the next put off before the send, so two senders in one |
| 353 | // window send it once. A deadline the disk cannot record is kept in the |
| 354 | // index (fix round 3, ruling 2), with its attempt count, until a write of |
| 355 | // the entry lands (re-review 3). Unwired, the disk's pages wait. |
| 356 | func (w *movesWorker) flushPageOutbox(ctx context.Context) { |
| 357 | page, dd := w.cfg.Page, w.cfg.DataDir |
| 358 | now := w.now() |
| 359 | flushPageIndex(ctx, now) |
| 360 | if page == nil || dd == "" { |
| 361 | return |
| 362 | } |
| 363 | root := pageOutboxDir(dd) |
| 364 | err := walkPageOutbox(root, func(p string) error { |
| 365 | if err := ctx.Err(); err != nil { |
| 366 | return err |
| 367 | } |
| 368 | w.flushPageEntry(ctx, page, dd, p, now) |
| 369 | return nil |
| 370 | }) |
| 371 | if err != nil { |
| 372 | pagingLogf("moves: the page outbox: %v", err) |
| 373 | } |
| 374 | } |
| 375 | |
| 376 | // walkPageOutbox calls fn with each entry file under root. |
| 377 | func walkPageOutbox(root string, fn func(p string) error) error { |
| 378 | return filepath.WalkDir(root, func(p string, d fs.DirEntry, err error) error { |
| 379 | if errors.Is(err, fs.ErrNotExist) && p == root { |
| 380 | return fs.SkipAll |
| 381 | } |
| 382 | if err != nil { |
| 383 | return err |
| 384 | } |
| 385 | if d.IsDir() || !strings.HasSuffix(d.Name(), ".json") || strings.HasPrefix(d.Name(), ".") { |
| 386 | return nil |
| 387 | } |
| 388 | return fn(p) |
| 389 | }) |
| 390 | } |
| 391 | |
| 392 | // flushPageEntry is one disk entry of flushPageOutbox. |
| 393 | func (w *movesWorker) flushPageEntry(ctx context.Context, page pageFunc, dd, p string, now time.Time) { |
| 394 | pageOutboxMu.Lock() |
| 395 | e, err := readPageEntry(p) |
| 396 | if err != nil || e.Rule+".json" != filepath.Base(p) || e.Op != filepath.Base(filepath.Dir(p)) || e.Host != filepath.Base(filepath.Dir(filepath.Dir(p))) { |
| 397 | pageOutboxMu.Unlock() |
| 398 | pagingLogf("moves: the page outbox: %s is not an entry, and is left for an operator: %v", p, err) |
| 399 | return |
| 400 | } |
| 401 | key := pageKeyOf(dd, e) |
| 402 | item := pageIndex[key] |
| 403 | if e.Delivered != nil { |
| 404 | if item != nil && !item.memory { |
| 405 | delete(pageIndex, key) // the disk says it all now |
| 406 | } |
| 407 | pageOutboxMu.Unlock() |
| 408 | return |
| 409 | } |
| 410 | if item != nil && (item.delivered || now.Before(item.next)) { |
| 411 | pageOutboxMu.Unlock() |
| 412 | return // delivered (its mark is the index's to write) or backing off |
| 413 | } |
| 414 | if item == nil && now.Before(e.Next) { |
| 415 | pageOutboxMu.Unlock() |
| 416 | return |
| 417 | } |
| 418 | // The index's attempts and deadline, when the disk could not record |
| 419 | // them, stand over the disk's until a write lands (re-review 3). |
| 420 | if item != nil && item.e.Tries > e.Tries { |
| 421 | e.Tries = item.e.Tries |
| 422 | } |
| 423 | e.Tries++ |
| 424 | e.Next = now.UTC().Add(pageBackoff(e.Tries)) |
| 425 | if err := writePageEntry(dd, e); err != nil { |
| 426 | pageIndex[key] = &pageIndexItem{e: e, next: e.Next} |
| 427 | pagingLogf("moves: %s: the %s page of move %s is sent with its backoff kept in memory: the outbox cannot record it: %v", e.Host, e.Rule, e.Op, err) |
| 428 | } else if item != nil { |
| 429 | delete(pageIndex, key) |
| 430 | } |
| 431 | pageOutboxMu.Unlock() |
| 432 | deliverPageEntry(ctx, page, dd, e, now) |
| 433 | } |
| 434 | |
| 435 | // flushPageIndex sends each memory page whose backoff is over (a delivered |
| 436 | // one is marked delivered, a refused one waits for its next attempt), and |
| 437 | // writes each delivered page's tombstone, dropping it from the index once |
| 438 | // it lands (fix round 3, ruling 3). |
| 439 | func flushPageIndex(ctx context.Context, now time.Time) { |
| 440 | pageOutboxMu.Lock() |
| 441 | var due []*pageIndexItem |
| 442 | for key, item := range pageIndex { |
| 443 | switch { |
| 444 | case item.delivered: |
| 445 | if err := markPageDelivered(key.dataDir, item.e, now); err == nil { |
| 446 | delete(pageIndex, key) |
| 447 | } |
| 448 | case item.memory && !now.Before(item.next): |
| 449 | item.e.Tries++ |
| 450 | item.next = now.UTC().Add(pageBackoff(item.e.Tries)) |
| 451 | due = append(due, item) |
| 452 | } |
| 453 | } |
| 454 | pageOutboxMu.Unlock() |
| 455 | for _, item := range due { |
| 456 | e := item.e |
| 457 | if err := item.page(ctx, e.Level, e.Subject, e.Body); err != nil { |
| 458 | pagingLogf("moves: %s: WARNING: the %s page of move %s, kept in memory only, was not delivered (try %d): %v", e.Host, e.Rule, e.Op, e.Tries, err) |
| 459 | continue |
| 460 | } |
| 461 | pagingLogf("moves: %s: the %s page of move %s, kept in memory, is delivered", e.Host, e.Rule, e.Op) |
| 462 | pageOutboxMu.Lock() |
| 463 | item.delivered = true |
| 464 | pageOutboxMu.Unlock() |
| 465 | } |
| 466 | } |
| 467 | |
| 468 | // retirePageTombstones removes each delivered disk entry whose move is |
| 469 | // retired on this box: no record in recs carries it and its journal is gone. |
| 470 | // Only a pass that listed every record calls it. |
| 471 | func (w *movesWorker) retirePageTombstones(recs []provision.Record) { |
| 472 | dd := w.cfg.DataDir |
| 473 | if dd == "" { |
| 474 | return |
| 475 | } |
| 476 | live := map[string]bool{} |
| 477 | for _, rec := range recs { |
| 478 | if rec.Move != nil && rec.Move.ID != "" { |
| 479 | live[rec.Host+"/"+rec.Move.ID] = true |
| 480 | } |
| 481 | } |
| 482 | root := pageOutboxDir(dd) |
| 483 | err := walkPageOutbox(root, func(p string) error { |
| 484 | pageOutboxMu.Lock() |
| 485 | defer pageOutboxMu.Unlock() |
| 486 | e, err := readPageEntry(p) |
| 487 | if err != nil || e.Delivered == nil || live[e.Host+"/"+e.Op] || moveJournalPresent(dd, e.Host, e.Op) { |
| 488 | return nil |
| 489 | } |
| 490 | if _, held := pageIndex[pageKeyOf(dd, e)]; held { |
| 491 | return nil |
| 492 | } |
| 493 | if err := os.Remove(p); err != nil { |
| 494 | pagingLogf("moves: the page outbox: %v", err) |
| 495 | } |
| 496 | return nil |
| 497 | }) |
| 498 | if err != nil { |
| 499 | pagingLogf("moves: the page outbox: %v", err) |
| 500 | } |
| 501 | pruneEmptyDirs(root) |
| 502 | } |
| 503 | |
| 504 | // firePage is the worker's firePage: its pager, outbox and clock. |
| 505 | func (w *movesWorker) firePage(ctx context.Context, host, op, rule, level, subject, body string) pageOutcome { |
| 506 | return firePage(ctx, w.cfg.Page, &w.pageWarned, w.cfg.DataDir, host, op, rule, w.now(), level, subject, body) |
| 507 | } |
| 508 | |
| 509 | // pageStuck pages an operator once a MoveMarkStuck has landed (moves spec |
| 510 | // §13: stuck is an incident; K11), once per entry into stuck. role is this |
| 511 | // box's part in the move, "source" or "target", and reason the recorded |
| 512 | // stuck sentence. The page names the host, the move, the box and the |
| 513 | // reason, and nothing else: no path, no error text, no secret. |
| 514 | func (w *movesWorker) pageStuck(ctx context.Context, rec provision.Record, role, reason string) { |
| 515 | m := rec.Move |
| 516 | if m == nil || m.ID == "" { |
| 517 | return |
| 518 | } |
| 519 | subject, body := stuckPage(rec.Host, m.ID, w.cfg.Box, role, reason) |
| 520 | w.firePage(ctx, rec.Host, m.ID, pageRuleStuck(m), alerting.LevelIncident, subject, body) |
| 521 | } |
| 522 | |
| 523 | // pageRolledBack pages a warning once this box has closed a move rolled |
| 524 | // back (§13): rec is the record as the close found it. |
| 525 | func (w *movesWorker) pageRolledBack(ctx context.Context, rec provision.Record) { |
| 526 | if rec.Move == nil || rec.Move.ID == "" { |
| 527 | return |
| 528 | } |
| 529 | subject, body := rolledBackPage(rec.Host, rec.Move.ID, w.cfg.Box) |
| 530 | w.firePage(ctx, rec.Host, rec.Move.ID, pageRuleRolledBack, alerting.LevelWarning, subject, body) |
| 531 | } |
| 532 | |
| 533 | // strayRefused reports whether a stop evidence report was refused for a |
| 534 | // process the move cannot account for: its StepStray ran and failed. A step |
| 535 | // that never ran is not a refusal. |
| 536 | func strayRefused(rep moves.StoppedReport) bool { |
| 537 | c, ok := rep.Steps[moves.StepStray] |
| 538 | return ok && !c.OK |
| 539 | } |
| 540 | |
| 541 | // pageStray pages an incident once per move for a stop evidence refused by a |
| 542 | // stray process (§13). The page's body is M-C20, and the agent's log says |
| 543 | // M-C20, with the move, when the box first detects it (the firing that keeps |
| 544 | // the page), whether or not the page is delivered then (fix round 2, |
| 545 | // ruling 1). rec is the cut's record. |
| 546 | func (w *movesWorker) pageStray(ctx context.Context, rec provision.Record) { |
| 547 | if rec.Move == nil || rec.Move.ID == "" { |
| 548 | return |
| 549 | } |
| 550 | subject, body := strayPage(rec.Host, w.cfg.Box) |
| 551 | switch w.firePage(ctx, rec.Host, rec.Move.ID, pageRuleStray, alerting.LevelIncident, subject, body) { |
| 552 | case pageSent, pageUndelivered: |
| 553 | pagingLogf("moves: %s: move %s: %s", rec.Host, rec.Move.ID, strayLine(w.cfg.Box, rec.Host)) |
| 554 | } |
| 555 | } |
| 556 | |
| 557 | // pageStartRefused pages an incident once per move for a refused §5.1 start |
| 558 | // check on this box (§13), and returns what became of the page. Its outbox |
| 559 | // is the box's, which the moves pass retries. |
| 560 | func (m *movesProbeOwners) pageStartRefused(ctx context.Context, host, op string) pageOutcome { |
| 561 | now := time.Now() |
| 562 | if m.Now != nil { |
| 563 | now = m.Now() |
| 564 | } |
| 565 | subject, body := startRefusedPage(host, op, m.Box) |
| 566 | return firePage(ctx, m.Page, &m.pageWarned, m.DataDir, host, op, pageRuleStartCheck, now, alerting.LevelIncident, subject, body) |
| 567 | } |
| 568 | |
| 569 | // rolledBackPage is the rolled_back page: the subject (MV-N385) and the |
| 570 | // body (MV-N386). |
| 571 | func rolledBackPage(host, id, box string) (subject, body string) { |
| 572 | subject = fmt.Sprintf("carlos move rolled back: %s", host) |
| 573 | body = fmt.Sprintf("The move %s of %s rolled back, and %s is serving on %s again.", id, host, host, box) |
| 574 | return subject, body |
| 575 | } |
| 576 | |
| 577 | // strayPage is the stray process's page: the subject (MV-N387) and the |
| 578 | // body, M-C20. |
| 579 | func strayPage(host, box string) (subject, body string) { |
| 580 | subject = fmt.Sprintf("carlos move stray process: %s", host) |
| 581 | return subject, strayLine(box, host) |
| 582 | } |
| 583 | |
| 584 | // strayLine is M-C20, the operators' sentence for the stray process: the |
| 585 | // stray page's body and the agent's log line. |
| 586 | func strayLine(box, host string) string { |
| 587 | return fmt.Sprintf("%s is running a process for %s that this move cannot account for. An operator has been paged.", box, host) |
| 588 | } |
| 589 | |
| 590 | // startRefusedPage is the refused start check's page: the subject |
| 591 | // (MV-N389) and the body (MV-N390). |
| 592 | func startRefusedPage(host, id, box string) (subject, body string) { |
| 593 | subject = fmt.Sprintf("carlos move start refused: %s", host) |
| 594 | body = fmt.Sprintf("%s refused to start %s for the move %s: what it holds of %s is not the generation the host claim names. Nothing was started.", |
| 595 | box, host, id, host) |
| 596 | return subject, body |
| 597 | } |
| 598 | |
amadan