Files
mesh-catalog/modules/artifact-store-tools/cmd/artifact-store-tools/holder.go
T
jschoubben 35ff67bda0
mesh/delivery-group group feat/774-l3-reland rejected: composed together: the checker's verdict at the members' heads (b48c90fea31f): integration: mesh-self-ch…
mesh/delivery closed: a newer head of the same pull request
mesh/check failed: integration: mesh-self-check/cmd/mesh-self-check.TestTheSelfCheckWholeRunsOnTheBus: the run answered {Run:doctor-1793351737-1 At:20…
Design 47's last milestones: the keeper, self-check, healers, hand-acts, provisioner, vault, artifact sweep, disk readings, credential forms (issues 774, 1952, 1953, 1954, 1955)
One commit with exactly the tree of the reviewed head 35b4509c0a, on main 12c2574c9d, so two compiled programs added and removed on the branch leave no trace in history (issue 1577).
2026-10-30 02:21:51 +01:00

1021 lines
32 KiB
Go

package main
// **The seat mesh-artifact-sweep** (novox/hq ADR 0571 clause 2, issue 1955): this module holds the artifact store's sweep, out
// of the controller. Its keep set is the store's own listing joined with every want's fingerprint (the assigner's
// `wants`) and each module's five newest builds (the builder's done keys, each build's end read from the key's history;
// ADR 0526 §1): a digest a want names is kept however old; a build beyond the five newest that no want names is
// eligible; a manifest in the store that no build names is unrecorded and never removed (ADR 0189 §10, ADR 0251 §6).
//
// At each build's end (a done key written) it sweeps as the controller's take-in did: every kept archive held first
// (ADR 0251 §8), then what is eligible let go of, oldest first, within the same bound per sweep, an index left for a
// `collect` a person confirms (ADR 0251 §7), each deletion recorded in the hand-act log (seat mesh-hand-acts) and said
// as artifacts-gone. **Nothing is let go until the controller's lease names a holder that is no controller**: until
// then the controller's take-in sweeps, and two sweeps never run at once. It also observes S25, image-missing: a want's
// image the store does not hold, a finding in its own state `findings` the condition keeper reads.
import (
"context"
"encoding/json"
"errors"
"fmt"
"sort"
"strings"
"sync"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/buscompose"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/handacts"
"git.novox.be/novox/mesh-sdk/go/health"
"artifactstoretools/internal/registry"
)
// What the holder reads and keeps.
const (
WantsRead = delivery.AssignerModule + ".wants"
BuildsRead = delivery.BuilderModule + ".builds"
HavesRead = "node-engine.have"
// ManifestsRead is the indexer's manifests: the images a manifest pins by itself.
ManifestsRead = delivery.IndexerModule + ".manifests"
LeaseRead = buscompose.ControllerLease
// FindingsState is the holder's own findings (health.FindingsState): image-missing.
FindingsState = health.FindingsState
// Module is this module, the observer its findings name.
Module = "artifact-store-tools"
// GoneSubject is where the seat says artifacts-gone.
GoneSubject = "mesh.seat." + catalogue.ArtifactSweepSeat + ".event." + delivery.ArtifactsGoneEvent
)
// The sweep's bounds, the controller's (cmd/mesh-controller/collect.go mostPerSweep, sweepBudget).
const (
KeptBuilds = 5
MostPerSweep = 500
MostAtAll = 5000
SweepBudget = 45 * time.Second
ImageMissing = "image-missing"
)
// Bus is what the holder reads and writes on the bus: its read states (wants, builds, haves, the lease), its own
// findings, and its seat's event. The runtime in main; a test's in memory.
type Bus interface {
Keys(state string) ([]string, error)
Get(state, key string) ([]byte, bool, error)
History(state, key string) ([]stdio.StateRevision, error)
// LogSince is every entry of a read log from a time on (stdio.Log(read).Since).
LogSince(read string, at time.Time) ([]stdio.LogEntry, error)
Put(state, key string, value any) error
Delete(state, key string) error
Publish(subject string, body any) error
}
// Door is the artifact store's own door: registry.Store, behind an interface so a test needs no store.
type Door interface {
Address() string
Hold(ctx context.Context, reference string) (bool, error)
Holds(ctx context.Context, reference string) error
IsIndex(ctx context.Context, reference string) (bool, error)
LetGo(ctx context.Context, reference string) error
// SpareKept sets the platform manifests a kept index names aside before an index is let go of (ADR 0257).
SpareKept(ctx context.Context, keptImages, touching []string) error
}
// Holder is the seat's holder.
type Holder struct {
Bus Bus
Door func() (Door, error)
// Listed is the store's own listing now (Store.View); nil reads none, and then nothing is said gone or collected.
Listed func(ctx context.Context) (*View, error)
Acts handacts.Recorder
Now func() time.Time
Logf func(string, ...any)
sweep sync.Mutex // one sweep at a time
mu sync.Mutex
ended map[string]time.Time // a done key's end, read once from its history
}
// Build is one done build of a module, as its key holds it, with when it ended.
type Build struct {
Module, FP, Key string
At time.Time
Artifacts []delivery.BuiltArtifact
// Manifest is the digest of the manifest it built, in the indexer's `manifests`.
Manifest string
}
// LeaseAway says the controller's lease names a holder that is no controller: then the seat sweeps.
func (h *Holder) LeaseAway() bool {
raw, found, err := h.Bus.Get(LeaseRead, buscompose.ControllerLeaseKey)
away, _ := buscompose.LeaseHandedAway(raw, found, err)
return away
}
// Builds is every done build the builder's keys hold, each with its end.
func (h *Holder) Builds() ([]Build, error) {
keys, err := h.Bus.Keys(BuildsRead)
if err != nil {
return nil, fmt.Errorf("the builder's builds: %w", err)
}
var out []Build
for _, k := range keys {
module, fp, err := delivery.ParseBuilderBuildKey(k)
if err != nil {
continue
}
raw, found, err := h.Bus.Get(BuildsRead, k)
if err != nil {
return nil, fmt.Errorf("the builder's build %s: %w", k, err)
}
if !found {
continue
}
b, err := delivery.DecodeBuilderBuild(raw)
if err != nil || b.State != delivery.BuildDone {
continue
}
at, err := h.endOf(k)
if err != nil {
return nil, err
}
out = append(out, Build{Module: module, FP: fp, Key: k, At: at, Artifacts: b.Artifacts, Manifest: b.Manifest})
}
return out, nil
}
// endOf is when a done key was last written done, read once from its history: a done key is not written again.
func (h *Holder) endOf(key string) (time.Time, error) {
h.mu.Lock()
at, known := h.ended[key]
h.mu.Unlock()
if known {
return at, nil
}
revs, err := h.Bus.History(BuildsRead, key)
if err != nil {
return time.Time{}, fmt.Errorf("the history of the builder's build %s: %w", key, err)
}
for _, r := range revs {
if b, err := delivery.DecodeBuilderBuild(r.Value); err == nil && b.State == delivery.BuildDone && r.At.After(at) {
at = r.At
}
}
h.mu.Lock()
if h.ended == nil {
h.ended = map[string]time.Time{}
}
h.ended[key] = at
h.mu.Unlock()
return at, nil
}
// Want is one node's want for a module.
type Want struct{ Node, Module, FP string }
// Wants is every want that names a fingerprint.
func (h *Holder) Wants() ([]Want, error) {
keys, err := h.Bus.Keys(WantsRead)
if err != nil {
return nil, fmt.Errorf("the wants: %w", err)
}
var out []Want
for _, k := range keys {
node, module, ok := wantKey(k)
if !ok {
continue
}
raw, found, err := h.Bus.Get(WantsRead, k)
if err != nil {
return nil, fmt.Errorf("the want %s: %w", k, err)
}
var w delivery.Want
if !found || json.Unmarshal(raw, &w) != nil || w.Removed || w.FP == "" {
continue
}
out = append(out, Want{Node: node, Module: module, FP: w.FP})
}
return out, nil
}
// wantKey reads `want.<node>.<module>`; the node's own key and a hold are not a module's want.
func wantKey(key string) (node, module string, ok bool) {
parts := strings.Split(key, ".")
if len(parts) != 3 || parts[0] != "want" || parts[2] == "node" {
return "", "", false
}
return parts[1], parts[2], true
}
// Ref is one artifact the mesh made, as the keep set judges it.
type Ref struct {
Reference string
Kind string // image or archive
Builds []Build
Why []string
Kept bool
// At is its newest build's end: the oldest go first.
At time.Time
}
// KeepSet judges every artifact the done builds name in the store: kept, and why, or eligible. Packages and
// references into no store of the mesh's are no business of the store's.
func KeepSet(builds []Build, wants []Want) map[string]*Ref {
keeps := keptBuilds(builds, wants)
out := map[string]*Ref{}
for _, b := range builds {
for _, a := range b.Artifacts {
reference := catalogue.Recorded(a.Reference)
if _, ours := catalogue.InArtifactStore(reference); !ours || a.Kind == "package" {
continue
}
r := out[reference]
if r == nil {
_, _, kind := splitReference(reference)
r = &Ref{Reference: reference, Kind: kind}
out[reference] = r
}
r.Builds = append(r.Builds, b)
r.keptBy(b, keeps[b.Key])
}
}
return out
}
// keptBy adds a build naming r, kept for why (none: eligible).
func (r *Ref) keptBy(b Build, why []string) {
if b.At.After(r.At) {
r.At = b.At
}
if len(why) > 0 {
r.Kept = true
for _, w := range why {
if !contains(r.Why, w) {
r.Why = append(r.Why, w)
}
}
}
}
// keptBuilds is, by build key, why each kept build is kept: one of its module's five newest, or a want names it.
func keptBuilds(builds []Build, wants []Want) map[string][]string {
wanted := map[string][]string{}
for _, w := range wants {
wanted[w.Module+" "+w.FP] = append(wanted[w.Module+" "+w.FP], w.Node)
}
byModule := map[string][]Build{}
for _, b := range builds {
byModule[b.Module] = append(byModule[b.Module], b)
}
keeps := map[string][]string{} // build key → why
for module, bs := range byModule {
sort.Slice(bs, func(i, j int) bool {
if !bs[i].At.Equal(bs[j].At) {
return bs[i].At.After(bs[j].At)
}
return bs[i].Key > bs[j].Key
})
for i, b := range bs {
if i < KeptBuilds {
keeps[b.Key] = append(keeps[b.Key], fmt.Sprintf("one of the %d newest builds of %s", KeptBuilds, module))
}
if nodes := wanted[module+" "+b.FP]; len(nodes) > 0 {
sort.Strings(nodes)
keeps[b.Key] = append(keeps[b.Key], "a want names it on "+strings.Join(nodes, ", "))
}
}
}
return keeps
}
func contains(s []string, v string) bool {
for _, x := range s {
if x == v {
return true
}
}
return false
}
// inStore says whether the listing holds a reference: an image's manifest, an archive's blob. Unknown without a listing.
func inStore(v *View, reference string) (held, known bool) {
if v == nil || v.L == nil {
return false, false
}
repo, digest, kind := splitReference(reference)
switch kind {
case "image":
r := v.L.Repos[repo]
return r != nil && r.Revisions[digest], true
case "archive":
_, has := v.L.Blobs[digest]
return has, true
}
return false, false
}
// judged is the keep set now, with the listing it was read against.
type judged struct {
refs map[string]*Ref
view *View
wants []Want
built []Build
// mirrors are the copies in the store's mirror repositories, by reference: recorded ones also in refs.
mirrors map[string]*Ref
// mirrorsHeld says why no copy is eligible now, when one is not: a kept build whose bases the log does not name.
mirrorsHeld string
}
func (h *Holder) judge(ctx context.Context) (*judged, error) {
builds, err := h.Builds()
if err != nil {
return nil, err
}
wants, err := h.Wants()
if err != nil {
return nil, err
}
j := &judged{refs: KeepSet(builds, wants), wants: wants, built: builds}
if h.Listed != nil {
if j.view, err = h.Listed(ctx); err != nil {
h.logf("the store's listing could not be read, so nothing is said gone: %v", err)
j.view = nil
}
}
if j.view != nil {
against, err := h.against()
if err != nil {
h.logf("the builder's log could not be read, so every copy is kept as unrecorded: %v", err)
against = nil
}
j.mirrors, j.mirrorsHeld = Mirrors(j.view, builds, keptBuilds(builds, wants), against)
for r, m := range j.mirrors {
if len(m.Builds) > 0 {
j.refs[r] = m
}
}
}
return j, nil
}
// against is what each build the builder's log says ended stood on, by build key: the bases' references.
func (h *Holder) against() (map[string][]string, error) {
entries, err := h.Bus.LogSince(delivery.BuildLogRead, h.now().Add(-delivery.BuilderLogDays*24*time.Hour))
if err != nil {
return nil, err
}
out := map[string][]string{}
for _, raw := range entries {
e, err := delivery.DecodeBuildLogEntry(raw.Value)
if err != nil || e.Event != delivery.BuildEndedEvent {
continue
}
key, err := delivery.BuilderBuildKey(e.Module, e.FP)
if err != nil {
continue
}
out[key] = e.Against
}
return out, nil
}
// Mirrors judges each copy in the store's mirror repositories (upstream/<host>/<path>, <module>/on-<argument>; ADR
// 0257) by what the builder's log says each build stood on (ADR 0571 clause 1): named by a kept build, kept; named
// only by builds no longer kept, eligible; named by no build the log holds, unrecorded and never removed (ADR 0189
// §10). While a kept build's bases are not in the log (it ended before the log's 30 days), no copy is eligible: what
// it stood on cannot be told, and held says so.
func Mirrors(v *View, builds []Build, kept map[string][]string, against map[string][]string) (map[string]*Ref, string) {
out := map[string]*Ref{}
for _, name := range v.L.RepoNames() {
if !mirrorRepository.MatchString(name) {
continue
}
for d := range v.L.Repos[name].Revisions {
r := catalogue.ArtifactStoreScheme + name + "@" + d
out[r] = &Ref{Reference: r, Kind: "image"}
}
}
held := ""
for _, b := range builds {
bases, logged := against[b.Key]
if !logged {
if len(kept[b.Key]) > 0 && held == "" {
held = "the builder's log does not say what the kept build " + b.Key + " stood on, so every copy is kept"
}
continue
}
for _, base := range bases {
if m := out[catalogue.Recorded(base)]; m != nil {
m.Builds = append(m.Builds, b)
why := kept[b.Key]
if len(why) > 0 {
why = []string{"a kept build stood on it: " + b.Key}
}
m.keptBy(b, why)
}
}
}
if held != "" {
for _, m := range out {
if len(m.Builds) > 0 && !m.Kept {
m.Kept, m.Why = true, []string{held}
}
}
}
return out, held
}
// Artifacts is the seat's `artifacts`, in the controller's answer's shape (Records): every reference the mesh made that
// lives in the store, kept with why or eligible, and with collected the ones the store no longer holds.
func (h *Holder) Artifacts(repository string) (*Records, error) {
j, err := h.judge(context.Background())
if err != nil {
return nil, err
}
r := j.records(repository, true)
if d, err := h.Door(); err == nil {
r.Store = d.Address()
}
return r, nil
}
func (j *judged) records(repository string, collected bool) *Records {
r := &Records{KeptBuilds: KeptBuilds, Counts: map[string]int{"kept": 0, "eligible": 0, "collected": 0}, CollectedAsked: collected}
for _, ref := range j.sorted() {
state := "eligible"
if ref.Kept {
state = "kept"
}
if held, known := inStore(j.view, ref.Reference); known && !held {
state = "collected"
if ref.Kept {
state = "kept" // a kept one gone is said missing by collect, never let go of
}
}
r.Counts[state]++
if (repository != "" && !strings.HasPrefix(strings.TrimPrefix(ref.Reference, catalogue.ArtifactStoreScheme), repository)) ||
(state == "collected" && !collected) {
continue
}
rec := Record{Reference: ref.Reference, Kind: ref.Kind, State: state, Why: ref.Why}
fill(&rec)
r.References = append(r.References, rec)
}
if j.view == nil {
r.Note = "the store's listing could not be read, so a reference it no longer holds is not said collected"
}
return r
}
// sorted is the references oldest first, then by reference.
func (j *judged) sorted() []*Ref {
out := make([]*Ref, 0, len(j.refs))
for _, r := range j.refs {
out = append(out, r)
}
sort.Slice(out, func(a, b int) bool {
if !out[a].At.Equal(out[b].At) {
return out[a].At.Before(out[b].At)
}
return out[a].Reference < out[b].Reference
})
return out
}
// Collect is the seat's `collect`, in the controller's answer's shape (CollectAnswer): a dry run unless confirm; with
// confirm and why, a sweep that also lets indexes go with their platform manifests.
func (h *Holder) Collect(why string, confirm bool, most int) (*CollectAnswer, error) {
if confirm && strings.TrimSpace(why) == "" {
return nil, errors.New("a confirmed collect needs why: nothing was let go of")
}
if most <= 0 {
most = MostPerSweep
}
if most > MostAtAll {
return nil, fmt.Errorf("collect lets go of at most %d at once, not %d", MostAtAll, most)
}
if confirm && !h.LeaseAway() {
return nil, errors.New("the controller holds its lease, and its take-in sweeps the store until it hands it away: " +
"this seat lets nothing go before then (novox/hq ADR 0571 clause 2); ask the controller's collect")
}
return h.Sweep(context.Background(), Sweeping{Confirmed: confirm, DryRun: !confirm, Most: most, Why: why, By: "collect"})
}
// Sweeping is one sweep's asks.
type Sweeping struct {
// Confirmed is a collect a person confirmed: an index goes with its platform manifests.
Confirmed bool
// DryRun lets nothing go and holds nothing: what would go is listed.
DryRun bool
Most int
Why string
// By names what asked: collect, or the build whose end it follows.
By string
}
// Sweep holds every kept archive, then lets go of what is eligible, oldest first, within the bounds, recording each
// deletion and saying artifacts-gone.
func (h *Holder) Sweep(ctx context.Context, s Sweeping) (*CollectAnswer, error) {
h.sweep.Lock()
defer h.sweep.Unlock()
j, err := h.judge(ctx)
if err != nil {
return nil, err
}
a := &CollectAnswer{DryRun: s.DryRun, WouldLetGo: []string{}, LetGo: []string{}}
var keptArchives, keptImages, eligible []string
for _, r := range j.sorted() {
switch {
case r.Kept && r.Kind == "archive":
keptArchives = append(keptArchives, r.Reference)
case r.Kept:
keptImages = append(keptImages, r.Reference)
default:
if held, known := inStore(j.view, r.Reference); known && !held {
continue // collected already
}
eligible = append(eligible, r.Reference)
}
}
a.KeptArchives, a.Eligible = len(keptArchives), len(eligible)
if s.DryRun {
for i, r := range eligible {
if i >= s.Most {
a.Left = len(eligible) - i
break
}
a.WouldLetGo = append(a.WouldLetGo, r)
}
return a, nil
}
door, err := h.Door()
if err != nil {
return nil, err
}
a.Store = door.Address()
within, stop := context.WithTimeout(ctx, SweepBudget)
defer stop()
// Hold before letting go (ADR 0251 §8): a kept archive that cannot be held stops the sweep before any delete.
var missing []string
for _, r := range keptArchives {
wrote, err := door.Hold(within, r)
switch {
case err == nil:
if wrote {
a.HoldersWritten++
}
case errors.Is(err, registry.Gone):
missing = append(missing, r)
case errors.Is(err, registry.ErrNotOurs):
default:
a.Stopped = fmt.Sprintf("not every kept archive could be held, so nothing was let go: holding %s: %v", r, err)
a.Left = len(eligible)
return a, nil
}
}
a.Held, a.Missing = len(keptArchives)-len(missing), len(missing)
h.sayGone(j, missing, "a kept archive the store does not hold")
if s.Confirmed {
if err := door.SpareKept(within, keptImages, eligible); err != nil {
a.Stopped = fmt.Sprintf("which platform manifests a kept index names could not be read, so nothing was let go: %v", err)
a.Left = len(eligible)
return a, nil
}
}
for i, r := range eligible {
if i >= s.Most || within.Err() != nil {
a.Left = len(eligible) - i
if i >= s.Most {
a.Stopped = fmt.Sprintf("asked about %d, the most one sweep asks", s.Most)
} else {
a.Stopped = fmt.Sprintf("ran out of its %s", SweepBudget)
}
break
}
if !s.Confirmed {
// An index waits for a confirmed collect (ADR 0251 §7, ADR 0257 §4).
index, err := door.IsIndex(within, r)
if err != nil && !errors.Is(err, registry.Gone) && !errors.Is(err, registry.ErrNotOurs) {
a.Stopped = fmt.Sprintf("the store could not say whether %s is an index, so nothing more was asked: %v", r, err)
a.Left = len(eligible) - i
break
}
if index {
a.Skipped++
continue
}
}
err := door.LetGo(within, r)
switch {
case err == nil || errors.Is(err, registry.Gone):
case errors.Is(err, registry.ErrNotOurs):
a.Skipped++
continue
default:
a.Stopped = fmt.Sprintf("the store refused to let go of %s, so nothing more was asked: %v", r, err)
a.Left = len(eligible) - i
}
if a.Stopped != "" {
break
}
// Recorded in the hand-act log, each one (ADR 0565 §10; the operator's "option A").
why := s.Why
if why == "" {
why = "eligible: no want names it and it is not among its module's " + fmt.Sprint(KeptBuilds) + " newest builds"
}
if _, err := h.Acts.Record(handacts.Request{What: "let go of " + r + " (" + s.By + ")", Why: why,
Cause: "artifact-store-sweep", Verb: "collect"}); err != nil {
h.logf("%s was let go of and could not be recorded in the hand-act log: %v", r, err)
}
a.LetGo = append(a.LetGo, r)
}
h.sayGone(j, a.LetGo, "let go of")
return a, nil
}
// sayGone says artifacts-gone for each build that named one of refs: the builder builds a key again whose artifact is gone.
func (h *Holder) sayGone(j *judged, refs []string, how string) {
byKey := map[string]*delivery.ArtifactsGone{}
var keys []string
for _, r := range refs {
ref := j.refs[r]
if ref == nil {
continue
}
for _, b := range ref.Builds {
g := byKey[b.Key]
if g == nil {
g = &delivery.ArtifactsGone{Module: b.Module, Key: b.Key, Fingerprint: b.FP, At: h.now().UTC()}
byKey[b.Key] = g
keys = append(keys, b.Key)
}
g.Gone = append(g.Gone, r)
}
}
sort.Strings(keys)
for _, k := range keys {
raw, err := delivery.EncodeArtifactsGone(*byKey[k])
if err == nil {
err = h.Bus.Publish(GoneSubject, json.RawMessage(raw))
}
if err != nil {
h.logf("%s: %s %s, and artifacts-gone could not be said: %v", k, how, strings.Join(byKey[k].Gone, ", "), err)
}
}
}
// AfterBuild is the sweep at a build's end: only once the lease names no controller.
func (h *Holder) AfterBuild(key string) {
if !h.LeaseAway() {
return
}
a, err := h.Sweep(context.Background(), Sweeping{Most: MostPerSweep, By: "after the build " + key})
if err != nil {
h.logf("the sweep after %s: %v", key, err)
return
}
if len(a.LetGo) > 0 || a.Stopped != "" || a.Missing > 0 {
h.logf("the sweep after %s let go of %d (%d left, %d missing): %s", key, len(a.LetGo), a.Left, a.Missing, a.Stopped)
}
}
// Held is the seat's `held`: each reference, held, gone or unknown with why. Holds and writes nothing.
func (h *Holder) Held(refs []string) (delivery.HeldAnswer, error) {
if len(refs) == 0 {
return delivery.HeldAnswer{}, errors.New("held judges the references given, and none was")
}
if len(refs) > 64 {
return delivery.HeldAnswer{}, fmt.Errorf("held judges at most 64 references at once, and %d were given", len(refs))
}
for _, r := range refs {
if _, ours := catalogue.InArtifactStore(r); !ours {
return delivery.HeldAnswer{}, fmt.Errorf("%s is not a reference into the mesh's artifact store (artifact-store://...): nothing was asked", r)
}
}
door, err := h.Door()
if err != nil {
return delivery.HeldAnswer{}, err
}
asking, stop := context.WithTimeout(context.Background(), 20*time.Second)
defer stop()
a := delivery.HeldAnswer{Store: door.Address(), Held: []string{}, Gone: []string{}}
for _, r := range refs {
switch err := door.Holds(asking, r); {
case err == nil:
a.Held = append(a.Held, r)
case errors.Is(err, registry.Gone):
a.Gone = append(a.Gone, r)
default:
if a.Unknown == nil {
a.Unknown = map[string]string{}
}
a.Unknown[r] = err.Error()
}
}
return a, nil
}
// Image is one image a machine's wants and haves name, as docker reads the answer.
type Image struct {
Image string `json:"image"`
Resources []string `json:"resources"`
In []string `json:"in"`
}
// Images is the seat's `images`, the controller's answer's shape: every image one machine's wants (in: declared) and
// haves (in: sent) name, by the builds their fingerprints name, each with the module naming it.
type Images struct {
Node string `json:"node"`
Images []Image `json:"images"`
SentKnown bool `json:"sent_known"`
}
// ImagesOf answers `images` for node.
func (h *Holder) ImagesOf(node string, address string) (Images, error) {
out := Images{Node: node, Images: []Image{}}
wants, err := h.Wants()
if err != nil {
return out, err
}
byImage := map[string]*Image{}
add := func(module, fp, in string) error {
key, err := delivery.BuilderBuildKey(module, fp)
if err != nil {
return nil
}
raw, found, err := h.Bus.Get(BuildsRead, key)
if err != nil || !found {
return err
}
b, err := delivery.DecodeBuilderBuild(raw)
if err != nil {
return nil
}
images := []string{}
for _, a := range b.Artifacts {
if (a.Kind == "image" || a.Kind == "upstream") && a.Reference != "" {
images = append(images, catalogue.Routed(catalogue.Recorded(a.Reference), address))
}
}
pinned, err := h.Pinned(b.Manifest)
if err != nil {
return err
}
for _, image := range append(images, pinned...) {
i := byImage[image]
if i == nil {
i = &Image{Image: image}
byImage[image] = i
}
if !contains(i.Resources, module) {
i.Resources = append(i.Resources, module)
}
if !contains(i.In, in) {
i.In = append(i.In, in)
}
}
return nil
}
for _, w := range wants {
if w.Node == node {
if err := add(w.Module, w.FP, "declared"); err != nil {
return out, err
}
}
}
keys, err := h.Bus.Keys(HavesRead)
if err != nil {
return out, fmt.Errorf("the haves: %w", err)
}
for _, k := range keys {
n, module, isNode, err := delivery.ParseHaveKey(k)
if err != nil || isNode || n != node {
continue
}
raw, found, err := h.Bus.Get(HavesRead, k)
if err != nil {
return out, err
}
var have delivery.Have
if !found || json.Unmarshal(raw, &have) != nil || have.FP == "" {
continue
}
out.SentKnown = true
if err := add(module, have.FP, "sent"); err != nil {
return out, err
}
}
for _, i := range byImage {
sort.Strings(i.Resources)
sort.Strings(i.In)
out.Images = append(out.Images, *i)
}
sort.Slice(out.Images, func(a, b int) bool { return out.Images[a].Image < out.Images[b].Image })
return out, nil
}
// Pinned is every container image a manifest names by itself, by digest (`<registry>/<path>@sha256:…`), read from the
// indexer's manifests: what a definition pins outright rather than through a build's artifact. A placeholder is not one.
func (h *Holder) Pinned(manifest string) ([]string, error) {
if manifest == "" {
return nil, nil
}
key, err := delivery.ManifestsKey(manifest)
if err != nil {
return nil, nil
}
raw, found, err := h.Bus.Get(ManifestsRead, key)
if err != nil || !found {
return nil, err
}
m, err := delivery.DecodeManifest(raw)
if err != nil {
return nil, nil
}
var parsed struct {
Resources []map[string]any `json:"resources"`
}
if json.Unmarshal([]byte(m.JSON), &parsed) != nil {
return nil, nil
}
var out []string
for _, r := range parsed.Resources {
image, _ := r["image"].(string)
if fmt.Sprint(r["type"]) == "container" && strings.Contains(image, "@sha256:") && !strings.Contains(image, "${") &&
!contains(out, image) {
out = append(out, image)
}
}
return out, nil
}
// inStoreAt is the store's reference of an image its address names, "" for an image of another registry (the
// controller's imageInStore).
func inStoreAt(image, address string) string {
if address == "" || !strings.HasPrefix(image, address+"/") || !strings.Contains(image, "@sha256:") {
return ""
}
r := catalogue.Recorded(image)
if _, ours := catalogue.InArtifactStore(r); !ours {
return ""
}
return r
}
// Observe writes S25's findings (image-missing): each want whose build names an image the store does not hold, on the
// want's machine, and deletes each finding that no longer holds. Nothing is said without a listing.
func (h *Holder) Observe(ctx context.Context) error {
j, err := h.judge(ctx)
if err != nil {
return err
}
if j.view == nil {
return nil
}
address := ""
if d, err := h.Door(); err == nil {
address = d.Address()
}
builds := map[string]Build{}
for _, b := range j.built {
builds[b.Module+" "+b.FP] = b
}
now := map[string]health.Finding{}
for _, w := range j.wants {
b, ok := builds[w.Module+" "+w.FP]
if !ok {
continue
}
var refs []string
for _, a := range b.Artifacts {
if a.Kind == "image" {
refs = append(refs, catalogue.Recorded(a.Reference))
}
}
// And what its manifest pins by itself under the store's address, as the controller judged S25.
if pinned, err := h.Pinned(b.Manifest); err == nil && address != "" {
for _, image := range pinned {
if r := inStoreAt(image, address); r != "" {
refs = append(refs, r)
}
}
}
for _, reference := range refs {
if held, known := inStore(j.view, reference); !known || held {
continue
}
f := imageMissing(w.Node, w.Module, reference, h.now())
now[f.Key] = f
break
}
}
keys, err := h.Bus.Keys(FindingsState)
if err != nil {
return fmt.Errorf("the findings: %w", err)
}
for _, k := range keys {
if _, still := now[k]; !still {
if err := h.Bus.Delete(FindingsState, k); err != nil {
return err
}
}
}
for k, f := range now {
raw, found, err := h.Bus.Get(FindingsState, k)
if err != nil {
return err
}
if found {
if was, err := health.DecodeFinding(raw); err == nil && was.Summary == f.Summary {
continue // written only when what it says changes
}
}
if _, err := health.EncodeFinding(f); err != nil {
return err
}
if err := h.Bus.Put(FindingsState, k, f); err != nil {
return err
}
}
return nil
}
// imageMissing is S25's finding, in the controller's words (cmd/mesh-controller/image_missing.go watchMissingImages).
func imageMissing(node, module, image string, at time.Time) health.Finding {
id := module + "." + node
return health.Finding{V: health.FindingVersion, Key: health.FindingKey(health.ScopeModule, id, ImageMissing), Kind: ImageMissing,
Subject: health.Subject{Scope: health.ScopeModule, ID: id, Machine: node},
Severity: health.Warning, Resolver: health.ResolverOperator,
Summary: fmt.Sprintf("%s is left out of every send to %s: its declaration names an image that the artifact store does "+
"not hold (the evidence names it); the running container is untouched", module, node),
Headline: capital(module) + " is not updated on " + node,
Explanation: fmt.Sprintf("The new version of %s names a container image that the mesh's store does not hold, so it "+
"was not sent to %s. What runs there is untouched, and the rest of %s is sent as usual.", module, node, node),
Needs: fmt.Sprintf("build %s again; the details name the image.", module),
Resolved: "Resolved: the store holds the image of " + module + " again",
Evidence: []health.Evidence{{At: at.UTC(), Said: "image " + image + " is not in the store"}},
Source: "S25", Observer: Module, At: at.UTC()}
}
func capital(s string) string {
if s == "" {
return s
}
return strings.ToUpper(s[:1]) + s[1:]
}
func (h *Holder) now() time.Time {
if h.Now != nil {
return h.Now()
}
return time.Now()
}
func (h *Holder) logf(format string, a ...any) {
if h.Logf != nil {
h.Logf(format, a...)
}
}
// Mirror is one copy as the seat's `mirrors` answers it.
type Mirror struct {
Reference string `json:"reference"`
State string `json:"state"`
Why []string `json:"why,omitempty"`
StoodOn []string `json:"stood_on_by,omitempty"`
}
// MirrorsAnswer is the seat's `mirrors`: every copy, and why none is eligible when a kept build's bases are unknown.
type MirrorsAnswer struct {
Copies []Mirror `json:"copies"`
Held string `json:"held,omitempty"`
}
// MirrorsNow is every copy in the store's mirror repositories as the keep set judges it now.
func (h *Holder) MirrorsNow() (MirrorsAnswer, error) {
j, err := h.judge(context.Background())
if err != nil {
return MirrorsAnswer{}, err
}
if j.view == nil {
return MirrorsAnswer{}, errors.New("the store's listing could not be read, so no copy can be named")
}
out := MirrorsAnswer{Copies: []Mirror{}, Held: j.mirrorsHeld}
for _, r := range j.mirrors {
m := Mirror{Reference: r.Reference, State: "unrecorded", Why: r.Why}
switch {
case r.Kept:
m.State = "kept"
case len(r.Builds) > 0:
m.State = "eligible"
}
for _, b := range r.Builds {
m.StoodOn = append(m.StoodOn, b.Key)
}
out.Copies = append(out.Copies, m)
}
sort.Slice(out.Copies, func(a, b int) bool { return out.Copies[a].Reference < out.Copies[b].Reference })
return out, nil
}