Ask the backup seats before the first tier (issue 1065)
This commit is contained in:
@@ -0,0 +1,294 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"slices"
|
||||
"sort"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/novox/mesh-controller/internal/catalogue"
|
||||
"github.com/novox/mesh-controller/internal/inventory"
|
||||
"github.com/novox/mesh-controller/internal/link"
|
||||
)
|
||||
|
||||
// **A backup before the send** (novox/hq ADR 0404 decisions 2 and 3, design 47, issue 1065). Once a delivery plan
|
||||
// has started and before it asks its first tier, the controller asks the node-backup holder of every machine that
|
||||
// runs a moved module with an item protected by a backup to back that module up now, and asks nothing of a builder
|
||||
// and sends nothing until every one of them answered done. The plan keeps the record (inventory.PlanBackups).
|
||||
|
||||
// backupsWithin is how long a backup is given from its ask to say it is done, and backupsPollEvery how soon the plan
|
||||
// is looked at again while it waits: variables so a test waits for neither.
|
||||
var (
|
||||
backupsWithin = 10 * time.Minute
|
||||
backupsPollEvery = 10 * time.Second
|
||||
)
|
||||
|
||||
// night is one module's last night as a holder's `backed-up` says it (lastNight).
|
||||
type night struct {
|
||||
OK bool `json:"ok"`
|
||||
At time.Time `json:"at"`
|
||||
Error string `json:"error,omitempty"`
|
||||
}
|
||||
|
||||
// askBackupNow asks a machine's backup holder to back a module up now; the holder answers when it has started. A
|
||||
// variable so a test needs no bus. An error is the holder's own words when it refused, or why it could not be asked.
|
||||
var askBackupNow = func(ctx context.Context, machine, module string) error {
|
||||
return onTheBus(func(conn *nats.Conn) error {
|
||||
answer, err := link.AskSeatTool(ctx, conn, catalogue.BackupSeat, "now", machine,
|
||||
map[string]any{"module": module}, 30*time.Second)
|
||||
if err != nil {
|
||||
return &backupNotAsked{fmt.Errorf("%s's backup holder was not asked: %w", machine, err)}
|
||||
}
|
||||
if answer.Error != "" {
|
||||
return fmt.Errorf("%s", answer.Error)
|
||||
}
|
||||
return nil
|
||||
})
|
||||
}
|
||||
|
||||
// backupNotAsked is an ask that could not be made, as opposed to one the holder refused: the step is tried again.
|
||||
type backupNotAsked struct{ error }
|
||||
|
||||
func (e *backupNotAsked) Unwrap() error { return e.error }
|
||||
|
||||
// readLastNights reads a machine's holder's `backed-up` by module: how each module's last backup went. A variable
|
||||
// so a test needs no bus.
|
||||
var readLastNights = func(ctx context.Context, machine string) (map[string]*night, error) {
|
||||
var out map[string]*night
|
||||
err := onTheBus(func(conn *nats.Conn) error {
|
||||
answer, err := link.AskSeatTool(ctx, conn, catalogue.BackupSeat, "backed-up", machine, map[string]any{}, 10*time.Second)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if answer.Error != "" {
|
||||
return fmt.Errorf("%s", answer.Error)
|
||||
}
|
||||
var read []struct {
|
||||
Module string `json:"module"`
|
||||
LastNight *night `json:"lastNight"`
|
||||
}
|
||||
if err := json.Unmarshal(answer.Result, &read); err != nil {
|
||||
return fmt.Errorf("its answer is not readable: %w", err)
|
||||
}
|
||||
out = map[string]*night{}
|
||||
for _, m := range read {
|
||||
out[m.Module] = m.LastNight
|
||||
}
|
||||
return nil
|
||||
})
|
||||
return out, err
|
||||
}
|
||||
|
||||
// backupFacts are what the step needs of the mesh: the modules with an item protected by a backup, where each
|
||||
// runs, and which machines hold node-backup.
|
||||
type backupFacts struct {
|
||||
backedUp map[string][]string // module → machines running it
|
||||
holder map[string]bool
|
||||
}
|
||||
|
||||
// backupFactsFrom reads the facts from the catalogue's entries and the seat holdings. A module is backed up when
|
||||
// one of its own items is (catalogue.DataItem.BackedUp: never a cache, `backup: none` or redundancy alone). A
|
||||
// module whose policy records (ADR 0377) is decided per machine after the build, so every machine running it is
|
||||
// asked: a superset costs time and never leaves a send without its restore point.
|
||||
func backupFactsFrom(entries []inventory.Entry, held []catalogue.Held) backupFacts {
|
||||
f := backupFacts{backedUp: map[string][]string{}, holder: map[string]bool{}}
|
||||
for _, e := range entries {
|
||||
for _, it := range e.Manifest.DataItems() {
|
||||
if it.BackedUp() {
|
||||
f.backedUp[e.Manifest.Module] = slices.Clone(e.On)
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
for _, h := range held {
|
||||
if s, known := catalogue.SeatNamed(h.Claim); known && s.Name == catalogue.BackupSeat {
|
||||
f.holder[h.Node] = true
|
||||
}
|
||||
}
|
||||
return f
|
||||
}
|
||||
|
||||
// loadBackupFacts reads the facts from the inventory.
|
||||
func loadBackupFacts(ctx context.Context, inv *inventory.Inventory) (backupFacts, error) {
|
||||
entries, err := inv.Catalogued(ctx)
|
||||
if err != nil {
|
||||
return backupFacts{}, err
|
||||
}
|
||||
held, err := inv.Holdings(ctx)
|
||||
if err != nil {
|
||||
return backupFacts{}, err
|
||||
}
|
||||
return backupFactsFrom(entries, held), nil
|
||||
}
|
||||
|
||||
// stepBackups takes one step of a plan's backups and says whether the plan may go on to ask its first tier. It
|
||||
// asks what is not asked, reads how the asked ones went, and records it all in p.Backups; it waits (false, nil)
|
||||
// while any is pending, and stays stopped (false, nil) once one failed, which part B holds and says. A plan whose
|
||||
// moved modules have no backed-up item keeps no record and goes on at once. An error is an ask that could not be
|
||||
// made or a read the store refused: the step is tried again, the asks already made kept.
|
||||
func stepBackups(ctx context.Context, p *inventory.Plan, facts backupFacts, now time.Time) (bool, error) {
|
||||
if p.Backups == nil {
|
||||
b := wantedBackups(*p, facts)
|
||||
if b == nil {
|
||||
return true, nil
|
||||
}
|
||||
p.Backups = b
|
||||
}
|
||||
b := p.Backups
|
||||
switch b.State {
|
||||
case inventory.BackupDone:
|
||||
return true, nil
|
||||
case inventory.BackupFailed:
|
||||
return false, nil
|
||||
}
|
||||
nights := map[string]map[string]*night{}
|
||||
for i := range b.Asked {
|
||||
a := &b.Asked[i]
|
||||
if a.State != inventory.BackupPending {
|
||||
continue
|
||||
}
|
||||
if a.AskedAt == nil {
|
||||
asked := now
|
||||
if err := askBackupNow(ctx, a.Machine, a.Module); err != nil {
|
||||
var unasked *backupNotAsked
|
||||
if errors.As(err, &unasked) {
|
||||
return false, err
|
||||
}
|
||||
a.State, a.Words = inventory.BackupFailed, err.Error()
|
||||
a.AskedAt = &asked
|
||||
continue
|
||||
}
|
||||
a.AskedAt = &asked
|
||||
continue
|
||||
}
|
||||
if _, read := nights[a.Machine]; !read {
|
||||
n, err := readLastNights(ctx, a.Machine)
|
||||
if err != nil {
|
||||
n = nil // not read now: the bound below still ends the wait
|
||||
}
|
||||
nights[a.Machine] = n
|
||||
}
|
||||
if n := nights[a.Machine][a.Module]; n != nil && n.At.After(*a.AskedAt) {
|
||||
if n.OK {
|
||||
a.State, a.Words = inventory.BackupDone, ""
|
||||
} else {
|
||||
a.State, a.Words = inventory.BackupFailed, n.Error
|
||||
if a.Words == "" {
|
||||
a.Words = "the backup failed and did not say why"
|
||||
}
|
||||
}
|
||||
continue
|
||||
}
|
||||
if now.Sub(*a.AskedAt) >= backupsWithin {
|
||||
a.State = inventory.BackupFailed
|
||||
a.Words = "did not answer in " + wordsOfBound(backupsWithin)
|
||||
}
|
||||
}
|
||||
b.State = backupsStateOf(b.Asked)
|
||||
switch b.State {
|
||||
case inventory.BackupDone:
|
||||
return true, nil
|
||||
case inventory.BackupPending:
|
||||
lookAgainForBackups(p.ID)
|
||||
}
|
||||
return false, nil
|
||||
}
|
||||
|
||||
// wantedBackups is the record of backups a plan has to take, or nil for none: one entry per machine with a holder
|
||||
// running a moved module that has a backed-up item, and the machines with none.
|
||||
func wantedBackups(p inventory.Plan, facts backupFacts) *inventory.PlanBackups {
|
||||
var asked []inventory.PlanBackup
|
||||
noHolder := map[string]bool{}
|
||||
for _, tier := range p.Tiers {
|
||||
for _, module := range tier {
|
||||
for _, machine := range facts.backedUp[module] {
|
||||
if !facts.holder[machine] {
|
||||
noHolder[machine] = true
|
||||
continue
|
||||
}
|
||||
asked = append(asked, inventory.PlanBackup{Machine: machine, Module: module, State: inventory.BackupPending})
|
||||
}
|
||||
}
|
||||
}
|
||||
if len(asked) == 0 && len(noHolder) == 0 {
|
||||
return nil
|
||||
}
|
||||
sort.SliceStable(asked, func(i, j int) bool {
|
||||
if asked[i].Machine != asked[j].Machine {
|
||||
return asked[i].Machine < asked[j].Machine
|
||||
}
|
||||
return asked[i].Module < asked[j].Module
|
||||
})
|
||||
b := &inventory.PlanBackups{Asked: asked}
|
||||
for m := range noHolder {
|
||||
b.NoHolder = append(b.NoHolder, m)
|
||||
}
|
||||
sort.Strings(b.NoHolder)
|
||||
b.State = backupsStateOf(asked)
|
||||
return b
|
||||
}
|
||||
|
||||
// backupsStateOf is the whole of the backups' states: failed if any failed, done if all are (or none are asked).
|
||||
func backupsStateOf(asked []inventory.PlanBackup) string {
|
||||
state := inventory.BackupDone
|
||||
for _, a := range asked {
|
||||
switch a.State {
|
||||
case inventory.BackupFailed:
|
||||
return inventory.BackupFailed
|
||||
case inventory.BackupPending:
|
||||
state = inventory.BackupPending
|
||||
}
|
||||
}
|
||||
return state
|
||||
}
|
||||
|
||||
func wordsOfBound(d time.Duration) string {
|
||||
if d == 10*time.Minute {
|
||||
return "ten minutes"
|
||||
}
|
||||
if d%time.Minute == 0 {
|
||||
return fmt.Sprintf("%d minutes", int(d/time.Minute))
|
||||
}
|
||||
return d.String()
|
||||
}
|
||||
|
||||
var (
|
||||
backupLookMu sync.Mutex
|
||||
backupLooks = map[string]bool{}
|
||||
)
|
||||
|
||||
// lookAgainForBackups has the serving controller look at the plans again soon, once per waiting plan: a backup ending
|
||||
// is no event the controller hears, so the wait is looked at by a timer, never by blocking the hold on the plans.
|
||||
func lookAgainForBackups(plan string) {
|
||||
backupLookMu.Lock()
|
||||
if backupLooks[plan] {
|
||||
backupLookMu.Unlock()
|
||||
return
|
||||
}
|
||||
backupLooks[plan] = true
|
||||
backupLookMu.Unlock()
|
||||
time.AfterFunc(backupsPollEvery, func() {
|
||||
backupLookMu.Lock()
|
||||
delete(backupLooks, plan)
|
||||
backupLookMu.Unlock()
|
||||
askToLook()
|
||||
})
|
||||
}
|
||||
|
||||
// backupsFirst is the step advanceOnce takes before it asks a walk's first tier: whether the plan may go on.
|
||||
func backupsFirst(ctx context.Context, inv *inventory.Inventory, p *inventory.Plan) (bool, error) {
|
||||
facts := backupFacts{}
|
||||
if p.Backups == nil {
|
||||
f, err := loadBackupFacts(ctx, inv)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
facts = f
|
||||
}
|
||||
return stepBackups(ctx, p, facts, time.Now().UTC())
|
||||
}
|
||||
Reference in New Issue
Block a user