1050 lines
36 KiB
Go
1050 lines
36 KiB
Go
// Package view derives the operator's rows from the delivery buckets.
|
|
package view
|
|
|
|
import (
|
|
"encoding/json"
|
|
"fmt"
|
|
"slices"
|
|
"sort"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/asks"
|
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
|
)
|
|
|
|
// Entry is one change of a bucket's key, as the watch delivers it.
|
|
type Entry struct {
|
|
Bucket string
|
|
Key string
|
|
Value []byte
|
|
At time.Time
|
|
Deleted bool
|
|
// Revision is the bucket revision of the change, when the watch gives one (the source revision is the gate's source key's).
|
|
Revision uint64
|
|
}
|
|
|
|
// Row is one module on one node.
|
|
type Row struct {
|
|
Node string `json:"node"`
|
|
Module string `json:"module"`
|
|
Status string `json:"status"`
|
|
Assigned string `json:"assigned,omitempty"`
|
|
Have string `json:"have,omitempty"`
|
|
Phase string `json:"phase,omitempty"`
|
|
Health string `json:"health,omitempty"`
|
|
Since time.Time `json:"since,omitzero"`
|
|
Detail string `json:"detail,omitempty"`
|
|
Waits string `json:"waits,omitempty"`
|
|
// RequiredBy is what pulled the module onto the node, for an assignment that is pulled in.
|
|
RequiredBy string `json:"required-by,omitempty"`
|
|
Error string `json:"error,omitempty"`
|
|
}
|
|
|
|
// NodeRow is a node's own facts as counts.
|
|
type NodeRow struct {
|
|
Node string `json:"node"`
|
|
Windows int `json:"windows"`
|
|
Busy int `json:"busy"`
|
|
Strays int `json:"strays"`
|
|
Since time.Time `json:"since,omitzero"`
|
|
Error string `json:"error,omitempty"`
|
|
}
|
|
|
|
// Snapshot is everything the view shows.
|
|
type Snapshot struct {
|
|
Rows []Row `json:"rows"`
|
|
Nodes []NodeRow `json:"nodes"`
|
|
// Offline is every node and part instance whose presence is offline (the watchman's word).
|
|
Offline []OfflineRow `json:"offline"`
|
|
// Stuck is every key that stopped moving and has not moved again.
|
|
Stuck []StuckRow `json:"stuck"`
|
|
// Unreadable is every watchman value the view could not read, named by its key: never dropped silently.
|
|
Unreadable []UnreadableRow `json:"unreadable"`
|
|
// Groups is every candidate group the indexer holds: the pull requests that share a branch, with the modules each touches.
|
|
Groups []GroupRow `json:"groups"`
|
|
// Fingerprints is every module the indexer fingerprinted at a trunk merge, beside its newest successful trunk build's.
|
|
Fingerprints []FingerprintRow `json:"fingerprints"`
|
|
// Asks is every ask the gate holds that has no answer yet: what waits for the operator's Approve.
|
|
Asks []AskRow `json:"asks"`
|
|
// Wants is every want of the assigner beside the have of its node and where it stands (novox/hq ADR 0539 §8, issue 1906 D7).
|
|
Wants []WantRow `json:"wants"`
|
|
// Source is the state of the mesh's source revision, derived from the wants and the haves (ADR 0429 §1): recorded by no part.
|
|
// Nil when the gate holds no source yet.
|
|
Source *SourceRow `json:"source,omitempty"`
|
|
// Backlog is, for each language and kind of build ask with one waiting or held, the asks waiting for a node, the asks a node
|
|
// holds and the oldest waiting ask's age (issue 1887 A8a).
|
|
Backlog []BacklogRow `json:"backlog"`
|
|
}
|
|
|
|
// The states of a want against the have of its node.
|
|
const (
|
|
WantApplied = "applied" // the node runs the want's fingerprint, running and healthy
|
|
WantPending = "pending" // the node has not yet run the want's fingerprint, or not healthy yet
|
|
WantFailed = "failed" // the node runs the want's fingerprint and its health check failed
|
|
WantHeldBack = "held back" // the assigner holds the node back from the newest fingerprint (Rule), or set it back
|
|
WantRemoved = "removed" // the node's want is removed: the module is no longer assigned to it
|
|
)
|
|
|
|
// WantRow is a want of the assigner beside the have of its node: the fingerprint wanted, the one the node runs with its phase and
|
|
// health, the state of the two, and for a want held back the newest fingerprint, the rule and the assigner's words.
|
|
type WantRow struct {
|
|
Node string `json:"node"`
|
|
Module string `json:"module"`
|
|
Want string `json:"want,omitempty"`
|
|
Have string `json:"have,omitempty"`
|
|
Phase string `json:"phase,omitempty"`
|
|
Health string `json:"health,omitempty"`
|
|
State string `json:"state"`
|
|
Newest string `json:"newest,omitempty"`
|
|
Rule int `json:"rule,omitempty"`
|
|
Words string `json:"words,omitempty"`
|
|
// Judgement is the controller's judgement of the move to the composition the want names (recreates, moves-data, neither or
|
|
// cannot-judge), and Judged its words; empty when the controller wrote none (ADR 0539 §14).
|
|
Judgement string `json:"judgement,omitempty"`
|
|
Judged string `json:"judged,omitempty"`
|
|
}
|
|
|
|
// The states of a source revision (ADR 0429 §1).
|
|
const (
|
|
SourceDelivered = "delivered"
|
|
SourceNotDelivered = "not delivered"
|
|
)
|
|
|
|
// SourceRow is the mesh's source revision (the revision of the gate's source key) and whether it is delivered: every want equals its
|
|
// module's fingerprint at the indexer and every have equals its want, running and healthy, and every node a module is assigned to
|
|
// has a want. Waiting names each module on a node that is not there yet, and why.
|
|
type SourceRow struct {
|
|
Revision uint64 `json:"revision"`
|
|
State string `json:"state"`
|
|
Waiting []string `json:"waiting,omitempty"`
|
|
// Wants is how many wants of modules at the indexer were looked at; a revision with none is delivered of nothing.
|
|
Wants int `json:"wants"`
|
|
}
|
|
|
|
// AskRow is an open ask (novox/hq ADR 0437 §10): the gate's ask with no answer under its key in the teller's answers.
|
|
type AskRow struct {
|
|
Key string `json:"key"`
|
|
Group string `json:"group"`
|
|
Headline string `json:"headline"`
|
|
Words string `json:"words"`
|
|
Pulls []string `json:"pulls"`
|
|
Expires time.Time `json:"expires,omitzero"`
|
|
How string `json:"how"`
|
|
// Options is what the view offers to answer with, for an ask that needs no proof of who answered (the assigner's hold and
|
|
// restore, ADR 0528 clause 6); Chosen is the option the view wrote that the teller has not filled yet.
|
|
Options []string `json:"options,omitempty"`
|
|
Chosen string `json:"chosen,omitempty"`
|
|
}
|
|
|
|
// AnswerOnThePhone is what an open ask says about answering it: the view only shows, and the Approve is answered on the
|
|
// operator's phone (ADR 0437 §10: answering in the view is release 5's).
|
|
const AnswerOnThePhone = "answer on the phone"
|
|
|
|
// AnswerInTheView is what an open ask that needs no proof of who answered says (ADR 0528 clause 6): the operator answers it here.
|
|
const AnswerInTheView = "answer in the view"
|
|
|
|
// GroupRow is a candidate group (novox/hq ADR 0437 §5): the branch its pull requests share and the pull requests.
|
|
type GroupRow struct {
|
|
Group string `json:"group"`
|
|
Pulls []GroupPull `json:"pulls"`
|
|
}
|
|
|
|
// GroupPull is one pull request of a group, its head shortened to 12 digits, and the modules it touches.
|
|
type GroupPull struct {
|
|
Repository string `json:"repository"`
|
|
Number int `json:"number"`
|
|
Head string `json:"head"`
|
|
Modules []string `json:"modules"`
|
|
}
|
|
|
|
// The states of a module's fingerprint against its newest successful trunk build's.
|
|
const (
|
|
FPSame = "same"
|
|
FPDiffers = "differs"
|
|
FPNone = "none"
|
|
FPNoBuild = "no build"
|
|
// The states of a fingerprint of the new form against the builder's key of it (I5, issue 1851): built (the key is done),
|
|
// build failed (Reason is its error), building (queued, claimed or building; Reason says what it waits for) and no key.
|
|
FPBuilt = "built"
|
|
FPBuildFailed = "build failed"
|
|
FPBuilding = "building"
|
|
FPNoKey = "no key"
|
|
)
|
|
|
|
// FingerprintRow compares the indexer's fingerprint of a module at a trunk merge with the source fingerprint its newest
|
|
// successful trunk build recorded. State is same, differs (both are given), none (the indexer could not fingerprint it:
|
|
// Reason says why) or no build (the indexer's alone is given, no successful trunk build recorded one).
|
|
type FingerprintRow struct {
|
|
Module string `json:"module"`
|
|
Repository string `json:"repository,omitempty"`
|
|
Commit string `json:"commit,omitempty"`
|
|
State string `json:"state"`
|
|
Indexer string `json:"indexer,omitempty"`
|
|
Build string `json:"build,omitempty"`
|
|
Reason string `json:"reason,omitempty"`
|
|
}
|
|
|
|
// OfflineRow is a node ("node") or a part's instance ("part") that is offline, since the presence entry's time.
|
|
type OfflineRow struct {
|
|
Kind string `json:"kind"`
|
|
Node string `json:"node,omitempty"`
|
|
Part string `json:"part,omitempty"`
|
|
Instance string `json:"instance,omitempty"`
|
|
Since time.Time `json:"since,omitzero"`
|
|
}
|
|
|
|
// StuckRow is a key that stopped moving: its bucket and key, what it waited for and since when.
|
|
type StuckRow struct {
|
|
Bucket string `json:"bucket"`
|
|
Key string `json:"key"`
|
|
Waited string `json:"waited,omitempty"`
|
|
Since time.Time `json:"since,omitzero"`
|
|
}
|
|
|
|
// UnreadableRow is a value of the watchman's buckets that did not decode, with the reader's error.
|
|
type UnreadableRow struct {
|
|
Bucket string `json:"bucket"`
|
|
Key string `json:"key"`
|
|
Error string `json:"error"`
|
|
}
|
|
|
|
// Model holds the rows' inputs in memory and nothing else: a restart fills it again from the buckets' last values.
|
|
type Model struct {
|
|
mu sync.Mutex
|
|
assigns map[string]assignment
|
|
builds map[string]buildEntry // by <module>.build.<id>
|
|
groups map[string]groupEntry // the indexer's candidates, by group
|
|
modules map[string]moduleEntry // the indexer's modules, by module
|
|
keys map[string]keyEntry // the builder's build keys, by <module>.<fingerprint>
|
|
asks map[string]askRecord // the gate's asks, by key
|
|
wants map[string]wantEntry // the assigner's wants, by <node>.<module>
|
|
source *sourceEntry // the gate's source key
|
|
deliveredEvent func(DeliveredEvent)
|
|
said uint64 // the source revision whose delivery was said
|
|
composed map[string]composedEntry // the controller's digests and judgements, by <node>.<module>.<hex>
|
|
aasks map[string]assignerAskRecord // the assigner's asks (hold, restore), by key
|
|
chosen map[string]delivery.ViewAnswer // the answers the view wrote, by ask key
|
|
writer Writer
|
|
backlog *Backlog
|
|
answers map[string]string // by key: the router id of the asking the teller's answer is for ("" when unreadable)
|
|
newest map[string]string // module -> id
|
|
haves map[string]haveEntry
|
|
nodes map[string]nodeEntry
|
|
// the watchman's buckets, by key
|
|
presence map[string]presenceEntry
|
|
stuck map[string]stuckEntry
|
|
}
|
|
|
|
type presenceEntry struct {
|
|
value delivery.Presence
|
|
err string
|
|
at time.Time
|
|
}
|
|
|
|
type stuckEntry struct {
|
|
value delivery.Stuck
|
|
err string
|
|
}
|
|
|
|
// assignment holds only the fields of the controller's Assignment the view reads.
|
|
type assignment struct {
|
|
Node string `json:"node"`
|
|
Module string `json:"module"`
|
|
Build string `json:"build"`
|
|
Commit string `json:"commit"`
|
|
Since time.Time `json:"since"`
|
|
Ended time.Time `json:"ended"`
|
|
State string `json:"state"`
|
|
PulledIn bool `json:"pulled-in"`
|
|
RequiredBy string `json:"required-by"`
|
|
}
|
|
|
|
// open says whether the value is an assignment that stands for the module on its node, as the controller's
|
|
// Assignment.Stands does. A module pulled in by another stands too.
|
|
func (a assignment) open() bool { return a.Ended.IsZero() && a.State == "" }
|
|
|
|
// build holds only the fields of the controller's Build record the view reads. A record with no state and no failure
|
|
// is a done build.
|
|
type build struct {
|
|
ID string `json:"id"`
|
|
Module string `json:"module"`
|
|
State string `json:"state"`
|
|
Commit string `json:"commit"`
|
|
Failed string `json:"failed"`
|
|
}
|
|
|
|
func (b build) done() bool { return b.State == "" && b.Failed == "" }
|
|
|
|
// buildEntry is a build as the view reads it: its own fields (state, failure) and the SDK's record (the source
|
|
// fingerprint, the ref, the time), decoded as the controller writes them.
|
|
type buildEntry struct {
|
|
build
|
|
rec delivery.BuildRecord
|
|
}
|
|
|
|
type groupEntry struct {
|
|
value delivery.Candidate
|
|
err string
|
|
}
|
|
|
|
type keyEntry struct {
|
|
value delivery.BuilderBuild
|
|
err string
|
|
}
|
|
|
|
type wantEntry struct {
|
|
value delivery.Want
|
|
err string
|
|
}
|
|
|
|
type composedEntry struct {
|
|
value delivery.ComposedDigest
|
|
err string
|
|
}
|
|
|
|
type sourceEntry struct {
|
|
revision uint64
|
|
group string
|
|
heads map[string]string
|
|
err string
|
|
}
|
|
|
|
type assignerAskRecord struct {
|
|
ask asks.Ask
|
|
err string
|
|
}
|
|
|
|
type moduleEntry struct {
|
|
value delivery.Module
|
|
err string
|
|
}
|
|
|
|
type askRecord struct {
|
|
value delivery.Ask
|
|
err string
|
|
}
|
|
|
|
type haveEntry struct {
|
|
have delivery.Have
|
|
err string
|
|
at time.Time
|
|
}
|
|
|
|
type nodeEntry struct {
|
|
node delivery.HaveNode
|
|
err string
|
|
at time.Time
|
|
}
|
|
|
|
// New is an empty model.
|
|
func New() *Model {
|
|
return &Model{assigns: map[string]assignment{}, builds: map[string]buildEntry{}, groups: map[string]groupEntry{}, modules: map[string]moduleEntry{}, keys: map[string]keyEntry{}, asks: map[string]askRecord{}, wants: map[string]wantEntry{}, composed: map[string]composedEntry{}, aasks: map[string]assignerAskRecord{}, chosen: map[string]delivery.ViewAnswer{}, answers: map[string]string{}, newest: map[string]string{},
|
|
haves: map[string]haveEntry{}, nodes: map[string]nodeEntry{},
|
|
presence: map[string]presenceEntry{}, stuck: map[string]stuckEntry{}}
|
|
}
|
|
|
|
// Apply takes one change of a key. A value it cannot read is kept as an error on its row, never guessed. When the change makes
|
|
// the source revision delivered, the event of it is said once (SetDeliveredEvent), after the model's lock is released.
|
|
func (m *Model) Apply(e Entry) {
|
|
m.mu.Lock()
|
|
m.apply(e)
|
|
event := m.deliveredNow()
|
|
say := m.deliveredEvent
|
|
m.mu.Unlock()
|
|
if event != nil && say != nil {
|
|
say(*event)
|
|
}
|
|
}
|
|
|
|
// DeliveredEvent is what the view says when the source revision is delivered (event `delivered` of the view, issue 1906 D9, ADR
|
|
// 0539 §17): the revision of the gate's source key, the group that moved it and every repository's merged head at it. A
|
|
// consumer takes every earlier revision as covered by it, and takes the same revision twice as once (after a restart of the view).
|
|
type DeliveredEvent struct {
|
|
Revision uint64 `json:"revision"`
|
|
Group string `json:"group"`
|
|
Heads map[string]string `json:"heads"`
|
|
}
|
|
|
|
// SetDeliveredEvent sets where the event of a delivered source revision goes; nothing is said until it is set.
|
|
func (m *Model) SetDeliveredEvent(say func(DeliveredEvent)) {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
m.deliveredEvent = say
|
|
}
|
|
|
|
// deliveredNow is the event of the source revision when it is delivered now and has not been said, or nil. A revision that went
|
|
// bad after it was delivered and recovered is not said again.
|
|
func (m *Model) deliveredNow() *DeliveredEvent {
|
|
if m.deliveredEvent == nil || m.source == nil || m.source.err != "" || m.said == m.source.revision {
|
|
return nil
|
|
}
|
|
row := m.sourceRow()
|
|
if row == nil || row.State != SourceDelivered || row.Wants == 0 {
|
|
return nil
|
|
}
|
|
m.said = m.source.revision
|
|
return &DeliveredEvent{Revision: m.source.revision, Group: m.source.group, Heads: m.source.heads}
|
|
}
|
|
|
|
func (m *Model) apply(e Entry) {
|
|
switch e.Bucket {
|
|
case delivery.HaveBucket:
|
|
node, module, isNode, err := delivery.ParseHaveKey(e.Key)
|
|
if err != nil {
|
|
return
|
|
}
|
|
if isNode {
|
|
if e.Deleted {
|
|
delete(m.nodes, node)
|
|
return
|
|
}
|
|
n, err := delivery.DecodeHaveNode(e.Value)
|
|
m.nodes[node] = nodeEntry{node: n, err: errText(err), at: e.At}
|
|
return
|
|
}
|
|
key := node + "." + module
|
|
if e.Deleted {
|
|
delete(m.haves, key)
|
|
return
|
|
}
|
|
h, err := delivery.DecodeHave(e.Value)
|
|
m.haves[key] = haveEntry{have: h, err: errText(err), at: e.At}
|
|
case delivery.PresenceBucket:
|
|
if e.Deleted {
|
|
delete(m.presence, e.Key)
|
|
return
|
|
}
|
|
p, err := delivery.DecodePresence(e.Value)
|
|
m.presence[e.Key] = presenceEntry{value: p, err: errText(err), at: e.At}
|
|
case delivery.StuckBucket:
|
|
if e.Deleted {
|
|
delete(m.stuck, e.Key)
|
|
return
|
|
}
|
|
st, err := delivery.DecodeStuck(e.Value)
|
|
m.stuck[e.Key] = stuckEntry{value: st, err: errText(err)}
|
|
case AssignmentsBucket:
|
|
if e.Deleted {
|
|
delete(m.assigns, e.Key)
|
|
return
|
|
}
|
|
var a assignment
|
|
if json.Unmarshal(e.Value, &a) != nil || a.Node == "" || a.Module == "" {
|
|
return
|
|
}
|
|
m.assigns[a.Node+"."+a.Module] = a
|
|
case delivery.CandidatesBucket:
|
|
if e.Deleted {
|
|
delete(m.groups, e.Key)
|
|
return
|
|
}
|
|
c, err := delivery.DecodeCandidate(e.Value)
|
|
m.groups[e.Key] = groupEntry{value: c, err: errText(err)}
|
|
case delivery.ModulesBucket:
|
|
if e.Deleted {
|
|
delete(m.modules, e.Key)
|
|
return
|
|
}
|
|
mod, err := delivery.DecodeModule(e.Value)
|
|
m.modules[e.Key] = moduleEntry{value: mod, err: errText(err)}
|
|
case delivery.AsksBucket:
|
|
if e.Deleted {
|
|
delete(m.asks, e.Key)
|
|
return
|
|
}
|
|
a, err := delivery.DecodeAsk(e.Value)
|
|
m.asks[e.Key] = askRecord{value: a, err: errText(err)}
|
|
case delivery.AssignerAsksBucket:
|
|
if e.Deleted {
|
|
delete(m.aasks, e.Key)
|
|
return
|
|
}
|
|
a, err := decodeAssignerAsk(e.Key, e.Value)
|
|
m.aasks[e.Key] = assignerAskRecord{ask: a, err: errText(err)}
|
|
case delivery.ViewAnswersBucket:
|
|
askKey, err := delivery.ParseViewAnswerKey(e.Key)
|
|
if err != nil {
|
|
return
|
|
}
|
|
if e.Deleted {
|
|
delete(m.chosen, askKey)
|
|
return
|
|
}
|
|
if a, err := delivery.DecodeViewAnswer(e.Value); err == nil {
|
|
m.chosen[askKey] = a
|
|
}
|
|
case delivery.AssignerWantsBucket:
|
|
node, module, err := delivery.ParseWantKey(e.Key)
|
|
if err != nil {
|
|
return // a hold, and the node's own key, are not a module's want
|
|
}
|
|
key := node + "." + module
|
|
if e.Deleted {
|
|
delete(m.wants, key)
|
|
return
|
|
}
|
|
w, err := delivery.DecodeWant(e.Value)
|
|
m.wants[key] = wantEntry{value: w, err: errText(err)}
|
|
case delivery.ControllerComposedBucket:
|
|
if e.Deleted {
|
|
delete(m.composed, e.Key)
|
|
return
|
|
}
|
|
d, err := delivery.DecodeComposedDigest(e.Value)
|
|
m.composed[e.Key] = composedEntry{value: d, err: errText(err)}
|
|
case delivery.SourceBucket:
|
|
if e.Key != delivery.SourceMainKey {
|
|
return
|
|
}
|
|
if e.Deleted {
|
|
m.source = nil
|
|
return
|
|
}
|
|
src, err := delivery.DecodeSource(e.Value)
|
|
m.source = &sourceEntry{revision: e.Revision, group: src.Group, heads: src.Heads, err: errText(err)}
|
|
case delivery.AnswersBucket:
|
|
if e.Deleted {
|
|
delete(m.answers, e.Key)
|
|
return
|
|
}
|
|
// The answer's warrant is `<router id>@<time>`: it ends that asking, and only that one (a later asking of the key is
|
|
// still open). An answer that cannot be read ends none.
|
|
id := ""
|
|
if a, err := delivery.DecodeAnswer(e.Value); err == nil {
|
|
id, _, _ = strings.Cut(a.Warrant, "@")
|
|
}
|
|
m.answers[e.Key] = id
|
|
case delivery.BuilderBuildsBucket:
|
|
if e.Deleted {
|
|
delete(m.keys, e.Key)
|
|
return
|
|
}
|
|
b, err := delivery.DecodeBuilderBuild(e.Value)
|
|
if err != nil {
|
|
m.keys[e.Key] = keyEntry{err: err.Error()}
|
|
return
|
|
}
|
|
m.keys[e.Key] = keyEntry{value: b}
|
|
case BuildsBucket:
|
|
module, rest, ok := strings.Cut(e.Key, ".")
|
|
if !ok {
|
|
return
|
|
}
|
|
switch {
|
|
case rest == "newest":
|
|
if e.Deleted {
|
|
delete(m.newest, module)
|
|
return
|
|
}
|
|
m.newest[module] = strings.TrimSpace(string(e.Value))
|
|
case strings.HasPrefix(rest, "build."):
|
|
if e.Deleted {
|
|
delete(m.builds, e.Key)
|
|
return
|
|
}
|
|
var b build
|
|
if json.Unmarshal(e.Value, &b) != nil {
|
|
return
|
|
}
|
|
// The source fingerprint is read with the SDK's record, not with the view's own struct.
|
|
var rec delivery.BuildRecord
|
|
if json.Unmarshal(e.Value, &rec) != nil {
|
|
return
|
|
}
|
|
m.builds[e.Key] = buildEntry{build: b, rec: rec}
|
|
}
|
|
}
|
|
}
|
|
|
|
func errText(err error) string {
|
|
if err == nil {
|
|
return ""
|
|
}
|
|
return err.Error()
|
|
}
|
|
|
|
// Snapshot derives one row per module and node with an open assignment, a row marked left behind for each have with
|
|
// none, and a row per node's own key. Rows are ordered by node and module.
|
|
func (m *Model) Snapshot() Snapshot {
|
|
s := m.snapshot()
|
|
m.mu.Lock()
|
|
b := m.backlog
|
|
m.mu.Unlock()
|
|
s.Backlog = []BacklogRow{}
|
|
if b != nil {
|
|
s.Backlog = b.Rows() // read from the bus outside the model's lock
|
|
}
|
|
return s
|
|
}
|
|
|
|
// SetBacklog gives the model the seats' workers to read the backlog of asks from.
|
|
func (m *Model) SetBacklog(b *Backlog) {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
m.backlog = b
|
|
}
|
|
|
|
func (m *Model) snapshot() Snapshot {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
rows := []Row{}
|
|
for key, a := range m.assigns {
|
|
if !a.open() {
|
|
continue
|
|
}
|
|
r := Row{Node: a.Node, Module: a.Module, Status: StatusAssigned, Assigned: a.Build}
|
|
if a.PulledIn {
|
|
r.RequiredBy = a.RequiredBy
|
|
}
|
|
var waits []string
|
|
h, has := m.haves[key]
|
|
switch {
|
|
case !has:
|
|
waits = append(waits, "no have yet")
|
|
case h.err != "":
|
|
r.Since, r.Error = h.at, h.err
|
|
default:
|
|
r.Have, r.Phase, r.Health = h.have.Build, string(h.have.Phase), string(h.have.Health)
|
|
r.Since, r.Detail = h.at, h.have.Detail
|
|
if a.Build != "" && h.have.Build != a.Build {
|
|
waits = append(waits, "the node has not applied "+a.Build)
|
|
}
|
|
}
|
|
if id := m.newest[a.Module]; id != "" && a.Build != "" && id != a.Build {
|
|
if b, ok := m.builds[a.Module+".build."+id]; ok && b.done() {
|
|
waits = append(waits, a.Module+" has a newer build "+id+" not yet assigned")
|
|
}
|
|
}
|
|
r.Waits = strings.Join(waits, "; ")
|
|
rows = append(rows, r)
|
|
}
|
|
for key, h := range m.haves {
|
|
if a, ok := m.assigns[key]; ok && a.open() {
|
|
continue
|
|
}
|
|
node, module, _, _ := delivery.ParseHaveKey(key)
|
|
r := Row{Node: node, Module: module, Status: StatusLeftBehind, Since: h.at, Error: h.err}
|
|
if h.err == "" && h.have.Phase == delivery.PhaseRemoved {
|
|
// The node-engine says it took the module away: plainly removed, not left behind.
|
|
r.Status = StatusRemoved
|
|
}
|
|
if h.err == "" {
|
|
r.Have, r.Phase, r.Health, r.Detail = h.have.Build, string(h.have.Phase), string(h.have.Health), h.have.Detail
|
|
}
|
|
rows = append(rows, r)
|
|
}
|
|
sort.Slice(rows, func(i, j int) bool {
|
|
if rows[i].Node != rows[j].Node {
|
|
return rows[i].Node < rows[j].Node
|
|
}
|
|
return rows[i].Module < rows[j].Module
|
|
})
|
|
nodes := []NodeRow{}
|
|
for name, n := range m.nodes {
|
|
nodes = append(nodes, NodeRow{Node: name, Windows: count(n.node.Windows), Busy: count(n.node.Busy),
|
|
Strays: count(n.node.Strays), Since: n.at, Error: n.err})
|
|
}
|
|
sort.Slice(nodes, func(i, j int) bool { return nodes[i].Node < nodes[j].Node })
|
|
offline, stuck, unreadable := []OfflineRow{}, []StuckRow{}, []UnreadableRow{}
|
|
for key, p := range m.presence {
|
|
if p.err != "" {
|
|
unreadable = append(unreadable, UnreadableRow{Bucket: delivery.PresenceBucket, Key: key, Error: p.err})
|
|
continue
|
|
}
|
|
if p.value.Online {
|
|
continue
|
|
}
|
|
node, part, instance, err := delivery.ParsePresenceKey(key)
|
|
if err != nil {
|
|
unreadable = append(unreadable, UnreadableRow{Bucket: delivery.PresenceBucket, Key: key, Error: err.Error()})
|
|
continue
|
|
}
|
|
if part != "" {
|
|
offline = append(offline, OfflineRow{Kind: "part", Part: part, Instance: instance, Since: p.at})
|
|
} else {
|
|
offline = append(offline, OfflineRow{Kind: "node", Node: node, Since: p.at})
|
|
}
|
|
}
|
|
for key, st := range m.stuck {
|
|
if st.err != "" {
|
|
unreadable = append(unreadable, UnreadableRow{Bucket: delivery.StuckBucket, Key: key, Error: st.err})
|
|
continue
|
|
}
|
|
if st.value.Moved {
|
|
continue
|
|
}
|
|
stuck = append(stuck, StuckRow{Bucket: st.value.Bucket, Key: st.value.Key, Waited: st.value.Waited, Since: st.value.Since})
|
|
}
|
|
for key, g := range m.groups {
|
|
if g.err != "" {
|
|
unreadable = append(unreadable, UnreadableRow{Bucket: delivery.CandidatesBucket, Key: key, Error: g.err})
|
|
}
|
|
}
|
|
for key, mod := range m.modules {
|
|
if mod.err != "" {
|
|
unreadable = append(unreadable, UnreadableRow{Bucket: delivery.ModulesBucket, Key: key, Error: mod.err})
|
|
}
|
|
}
|
|
for key, a := range m.asks {
|
|
if a.err != "" {
|
|
unreadable = append(unreadable, UnreadableRow{Bucket: delivery.AsksBucket, Key: key, Error: a.err})
|
|
}
|
|
}
|
|
for key, a := range m.aasks {
|
|
if a.err != "" {
|
|
unreadable = append(unreadable, UnreadableRow{Bucket: delivery.AssignerAsksBucket, Key: key, Error: a.err})
|
|
}
|
|
}
|
|
for key, w := range m.wants {
|
|
if w.err != "" {
|
|
unreadable = append(unreadable, UnreadableRow{Bucket: delivery.AssignerWantsBucket, Key: key, Error: w.err})
|
|
}
|
|
}
|
|
sort.Slice(offline, func(i, j int) bool {
|
|
a, b := offline[i], offline[j]
|
|
return a.Kind+a.Node+a.Part+a.Instance < b.Kind+b.Node+b.Part+b.Instance
|
|
})
|
|
sort.Slice(stuck, func(i, j int) bool {
|
|
return stuck[i].Bucket+"."+stuck[i].Key < stuck[j].Bucket+"."+stuck[j].Key
|
|
})
|
|
sort.Slice(unreadable, func(i, j int) bool {
|
|
return unreadable[i].Bucket+"."+unreadable[i].Key < unreadable[j].Bucket+"."+unreadable[j].Key
|
|
})
|
|
return Snapshot{Rows: rows, Nodes: nodes, Offline: offline, Stuck: stuck, Unreadable: unreadable,
|
|
Groups: m.groupRows(), Fingerprints: m.fingerprintRows(), Asks: m.askRows(), Wants: m.wantRows(), Source: m.sourceRow()}
|
|
}
|
|
|
|
// groupRows are the readable candidate groups, ordered by group, their pull requests by repository and number.
|
|
func (m *Model) groupRows() []GroupRow {
|
|
out := []GroupRow{}
|
|
for key, g := range m.groups {
|
|
if g.err != "" {
|
|
continue
|
|
}
|
|
row := GroupRow{Group: key, Pulls: []GroupPull{}}
|
|
for _, p := range g.value.Pulls {
|
|
head := p.Head.Head
|
|
if len(head) > 12 {
|
|
head = head[:12]
|
|
}
|
|
row.Pulls = append(row.Pulls, GroupPull{Repository: p.Repository, Number: p.Number, Head: head, Modules: append([]string{}, p.Modules...)})
|
|
}
|
|
sort.Slice(row.Pulls, func(i, j int) bool {
|
|
a, b := row.Pulls[i], row.Pulls[j]
|
|
return a.Repository < b.Repository || (a.Repository == b.Repository && a.Number < b.Number)
|
|
})
|
|
out = append(out, row)
|
|
}
|
|
sort.Slice(out, func(i, j int) bool { return out[i].Group < out[j].Group })
|
|
return out
|
|
}
|
|
|
|
// fingerprintRows compare each module the indexer fingerprinted with the source fingerprint of the module's newest
|
|
// successful trunk build (delivery.BuildRecord.IsTrunkBuild: an outcome, not failed, on main or with no ref) that recorded one. The newest key is any state and ref, so it is not
|
|
// used: the records are.
|
|
func (m *Model) fingerprintRows() []FingerprintRow {
|
|
newest := map[string]buildEntry{}
|
|
for _, b := range m.builds {
|
|
if b.rec.Module == "" || !b.rec.IsTrunkBuild() || b.rec.SourceFingerprint == "" {
|
|
continue
|
|
}
|
|
if old, ok := newest[b.rec.Module]; !ok || b.rec.At.After(old.rec.At) || (b.rec.At.Equal(old.rec.At) && b.rec.ID > old.rec.ID) {
|
|
newest[b.rec.Module] = b
|
|
}
|
|
}
|
|
out := []FingerprintRow{}
|
|
for name, mod := range m.modules {
|
|
if mod.err != "" {
|
|
continue
|
|
}
|
|
row := FingerprintRow{Module: name, Repository: mod.value.Repository, Commit: mod.value.Commit}
|
|
if mod.value.FP2 != "" || mod.value.None2 != "" {
|
|
// The new form (R4e): the indexer's fp2 against the builder's key of it.
|
|
out = append(out, m.newFormRow(name, row, mod.value))
|
|
continue
|
|
}
|
|
b, hasBuild := newest[name]
|
|
switch {
|
|
case mod.value.None != "":
|
|
row.State, row.Reason = FPNone, mod.value.None
|
|
case !hasBuild:
|
|
row.State, row.Indexer = FPNoBuild, mod.value.FP
|
|
case b.rec.SourceFingerprint == mod.value.FP:
|
|
row.State, row.Indexer, row.Build = FPSame, mod.value.FP, b.rec.SourceFingerprint
|
|
default:
|
|
row.State, row.Indexer, row.Build = FPDiffers, mod.value.FP, b.rec.SourceFingerprint
|
|
}
|
|
out = append(out, row)
|
|
}
|
|
sort.Slice(out, func(i, j int) bool { return out[i].Module < out[j].Module })
|
|
return out
|
|
}
|
|
|
|
// newFormRow is a module's row of the new form: none with its reason, or its fp2 against the state of the builder's key
|
|
// of it (I5, issue 1851; the build records name no fingerprint of the new form).
|
|
func (m *Model) newFormRow(name string, row FingerprintRow, v delivery.Module) FingerprintRow {
|
|
if v.FP2 == "" {
|
|
row.State, row.Reason = FPNone, v.None2
|
|
return row
|
|
}
|
|
row.Indexer = v.FP2
|
|
key, err := delivery.BuilderBuildKey(name, v.FP2)
|
|
if err != nil {
|
|
row.State, row.Reason = FPNoKey, err.Error()
|
|
return row
|
|
}
|
|
k, ok := m.keys[key]
|
|
switch {
|
|
case !ok:
|
|
row.State = FPNoKey
|
|
case k.err != "":
|
|
row.State, row.Reason = FPBuildFailed, "the key cannot be read: "+k.err
|
|
case k.value.State == delivery.BuildDone:
|
|
row.State, row.Build = FPBuilt, v.FP2
|
|
case k.value.State == delivery.BuildFailed:
|
|
row.State, row.Reason = FPBuildFailed, k.value.Error
|
|
default:
|
|
row.State, row.Reason = FPBuilding, strings.TrimSpace(k.value.State+" "+k.value.Waits)
|
|
}
|
|
return row
|
|
}
|
|
|
|
// count is the number of items of a carried JSON array or object, and 0 for anything else.
|
|
func count(raw json.RawMessage) int {
|
|
var list []json.RawMessage
|
|
if json.Unmarshal(raw, &list) == nil {
|
|
return len(list)
|
|
}
|
|
var obj map[string]json.RawMessage
|
|
if json.Unmarshal(raw, &obj) == nil {
|
|
return len(obj)
|
|
}
|
|
return 0
|
|
}
|
|
|
|
// Counted is the number of rows that are modules on the nodes, not left behind.
|
|
func (s Snapshot) Counted() int {
|
|
n := 0
|
|
for _, r := range s.Rows {
|
|
if r.Status == StatusAssigned {
|
|
n++
|
|
}
|
|
}
|
|
return n
|
|
}
|
|
|
|
// The buckets the view reads besides the have bucket (which mesh-sdk's delivery package names).
|
|
const (
|
|
AssignmentsBucket = "mesh-controller_assignments"
|
|
BuildsBucket = "mesh-controller_builds"
|
|
)
|
|
|
|
// The statuses of a row.
|
|
const (
|
|
StatusAssigned = "assigned"
|
|
StatusLeftBehind = "left behind"
|
|
StatusRemoved = "removed"
|
|
)
|
|
|
|
// askRows are the readable asks with no answer under their key, ordered by key: each with its headline, its words, the
|
|
// pull requests it covers (repository, number, head shortened to 12 digits) and how to answer.
|
|
func (m *Model) askRows() []AskRow {
|
|
out := []AskRow{}
|
|
for key, a := range m.asks {
|
|
if a.err != "" || m.answers[key] == a.value.Ask.ID {
|
|
continue
|
|
}
|
|
row := AskRow{Key: key, Group: a.value.Group, Headline: a.value.Ask.Headline, Words: a.value.Ask.Explanation,
|
|
Expires: a.value.Ask.Expires, How: AnswerOnThePhone, Pulls: []string{}}
|
|
for _, h := range a.value.Heads {
|
|
head := h.Head
|
|
if len(head) > 12 {
|
|
head = head[:12]
|
|
}
|
|
row.Pulls = append(row.Pulls, fmt.Sprintf("%s#%d @%s", h.Repository, h.Number, head))
|
|
}
|
|
out = append(out, row)
|
|
}
|
|
for key, a := range m.aasks {
|
|
if a.err != "" || m.answers[key] == a.ask.ID {
|
|
continue
|
|
}
|
|
row := AskRow{Key: key, Headline: a.ask.Headline, Words: a.ask.Explanation, Expires: a.ask.Expires, How: AnswerInTheView, Pulls: []string{},
|
|
Chosen: m.chosenOf(key, a.ask)}
|
|
for _, o := range a.ask.Options {
|
|
row.Options = append(row.Options, o.ID)
|
|
}
|
|
out = append(out, row)
|
|
}
|
|
sort.Slice(out, func(i, j int) bool { return out[i].Key < out[j].Key })
|
|
return out
|
|
}
|
|
|
|
// decodeAssignerAsk reads a key of the assigner's asks (mesh-sdk go/delivery DecodeAssignerAsk): the router's ask under its
|
|
// asking's id, of a kind that needs no proof, offering an option.
|
|
func decodeAssignerAsk(key string, value []byte) (asks.Ask, error) {
|
|
a, err := delivery.DecodeAssignerAsk(key, value)
|
|
if err != nil {
|
|
return asks.Ask{}, fmt.Errorf("the assigner's ask %s is not readable: %w", key, err)
|
|
}
|
|
return a.Ask, nil
|
|
}
|
|
|
|
// chosenOf is the option the view wrote for this asking of the ask: an answer to another asking (a stale one) is none.
|
|
func (m *Model) chosenOf(key string, ask asks.Ask) string {
|
|
if a, ok := m.chosen[key]; ok && a.Asking == ask.ID {
|
|
return a.Option
|
|
}
|
|
return ""
|
|
}
|
|
|
|
// Writer puts a key of the view's bucket answers, the only state the view writes.
|
|
type Writer interface {
|
|
Put(key string, value []byte) error
|
|
}
|
|
|
|
// SetWriter gives the model the bucket its answers are written to.
|
|
func (m *Model) SetWriter(w Writer) {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
m.writer = w
|
|
}
|
|
|
|
// Answer writes the view's answer to the open ask at askKey: option, one the ask offers, and the operator's words. It writes
|
|
// `<ask key>.view` and nothing else; an ask that needs proof of who answered (the merge's Approve, issue 953) is refused with
|
|
// its reason, as is an ask the view does not hold open, an option the ask does not offer, and one the view answered already.
|
|
func (m *Model) Answer(askKey, option, words string) error {
|
|
key, err := delivery.ViewAnswerKey(askKey)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
option = strings.TrimSpace(option)
|
|
if option == "" {
|
|
return fmt.Errorf("an answer names no option")
|
|
}
|
|
m.mu.Lock()
|
|
a, held := m.aasks[askKey]
|
|
open := held && a.err == "" && m.answers[askKey] != a.ask.ID
|
|
chosen, w := m.chosenOf(askKey, a.ask), m.writer
|
|
m.mu.Unlock()
|
|
if !open {
|
|
return fmt.Errorf("the view holds no open ask %s", askKey)
|
|
}
|
|
if !slices.ContainsFunc(a.ask.Options, func(o asks.Option) bool { return o.ID == option }) {
|
|
ids := []string{}
|
|
for _, o := range a.ask.Options {
|
|
ids = append(ids, o.ID)
|
|
}
|
|
return fmt.Errorf("the ask %s offers no option %q (it offers %s)", askKey, option, strings.Join(ids, ", "))
|
|
}
|
|
if chosen != "" {
|
|
return fmt.Errorf("the view answered %s with %q already: the teller fills the ask from it", askKey, chosen)
|
|
}
|
|
if w == nil {
|
|
return fmt.Errorf("the view cannot write its answers here")
|
|
}
|
|
body, err := delivery.EncodeViewAnswer(delivery.ViewAnswer{Asking: a.ask.ID, Option: option, Words: strings.TrimSpace(words)})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return w.Put(key, body)
|
|
}
|
|
|
|
// wantRows are the readable wants beside the haves of their nodes, ordered by node and module.
|
|
func (m *Model) wantRows() []WantRow {
|
|
out := []WantRow{}
|
|
for key, w := range m.wants {
|
|
if w.err != "" {
|
|
continue
|
|
}
|
|
node, module, _ := strings.Cut(key, ".")
|
|
v := w.value
|
|
row := WantRow{Node: node, Module: module, Want: v.FP, Newest: v.Newest, Rule: v.Rule, Words: v.Words}
|
|
h, has := m.haves[key]
|
|
if has && h.err == "" {
|
|
row.Have, row.Phase, row.Health = h.have.FP, string(h.have.Phase), string(h.have.Health)
|
|
}
|
|
if ckey, err := delivery.ComposedKey(node, module, v.FP); err == nil && !v.Removed {
|
|
if c, ok := m.composed[ckey]; ok && c.err == "" {
|
|
row.Judgement, row.Judged = string(c.value.Judgement), c.value.Words
|
|
}
|
|
}
|
|
switch {
|
|
case v.Removed:
|
|
row.State = WantRemoved
|
|
case v.Rule != 0 || v.Newest != "":
|
|
row.State = WantHeldBack
|
|
case row.Have == v.FP && h.have.Health == delivery.HealthFailed:
|
|
row.State = WantFailed
|
|
case row.Have == v.FP && h.have.Phase == delivery.PhaseRunning && h.have.Health == delivery.HealthOK:
|
|
row.State = WantApplied
|
|
default:
|
|
row.State = WantPending
|
|
}
|
|
out = append(out, row)
|
|
}
|
|
sort.Slice(out, func(i, j int) bool {
|
|
if out[i].Node != out[j].Node {
|
|
return out[i].Node < out[j].Node
|
|
}
|
|
return out[i].Module < out[j].Module
|
|
})
|
|
return out
|
|
}
|
|
|
|
// sourceRow derives the state of the source revision the gate holds (ADR 0429 §1; recorded by no part): delivered when every want
|
|
// equals its module's fingerprint at the indexer and every have equals its want, running and healthy, and every node a module is
|
|
// assigned to has a want. It is nil when the gate holds no readable source.
|
|
func (m *Model) sourceRow() *SourceRow {
|
|
if m.source == nil || m.source.err != "" {
|
|
return nil
|
|
}
|
|
waiting := []string{}
|
|
short := func(fp string) string {
|
|
fp = strings.TrimPrefix(fp, delivery.SourcePrefix2)
|
|
if len(fp) > 12 {
|
|
fp = fp[:12]
|
|
}
|
|
return fp
|
|
}
|
|
looked := 0
|
|
for key, w := range m.wants {
|
|
node, module, _ := strings.Cut(key, ".")
|
|
mod, ok := m.modules[module]
|
|
if w.err != "" || w.value.Removed || !ok || mod.err != "" || mod.value.FP2 == "" {
|
|
continue
|
|
}
|
|
looked++
|
|
h := m.haves[key]
|
|
switch {
|
|
case w.value.FP != mod.value.FP2:
|
|
waiting = append(waiting, fmt.Sprintf("%s on %s wants %s, not the module's %s", module, node, short(w.value.FP), short(mod.value.FP2)))
|
|
case h.err != "" || h.have.FP != w.value.FP:
|
|
waiting = append(waiting, fmt.Sprintf("%s on %s does not yet run %s", module, node, short(w.value.FP)))
|
|
case h.have.Health == delivery.HealthFailed:
|
|
waiting = append(waiting, fmt.Sprintf("%s on %s runs %s and its health check failed", module, node, short(w.value.FP)))
|
|
case h.have.Phase != delivery.PhaseRunning || h.have.Health != delivery.HealthOK:
|
|
waiting = append(waiting, fmt.Sprintf("%s on %s runs %s, not yet running and healthy", module, node, short(w.value.FP)))
|
|
}
|
|
}
|
|
for key, a := range m.assigns {
|
|
if !a.open() {
|
|
continue
|
|
}
|
|
if _, ok := m.wants[key]; ok {
|
|
continue
|
|
}
|
|
if mod, ok := m.modules[a.Module]; ok && mod.err == "" && mod.value.FP2 != "" {
|
|
waiting = append(waiting, fmt.Sprintf("%s on %s has no want yet", a.Module, a.Node))
|
|
}
|
|
}
|
|
sort.Strings(waiting)
|
|
row := &SourceRow{Revision: m.source.revision, State: SourceDelivered, Wants: looked}
|
|
if len(waiting) > 0 {
|
|
row.State, row.Waiting = SourceNotDelivered, waiting
|
|
}
|
|
return row
|
|
}
|