1021 lines
32 KiB
Go
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
|
|
}
|