Files
mesh-catalog/modules/artifact-store-tools/cmd/artifact-store-tools/holder.go
T

1021 lines
32 KiB
Go

package main
// **The seat artifact-store** (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.ArtifactStoreSeat + ".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
}