mesh-tools
7855ea126c
The callers that only ask (status, the doctor's probes, facts, bindings, data, acts, nodes show, images, release, recorded push, consumers) resolve through resolveNode and compose through composeQuestion, over one Input read per operation (withInputCache). A question composes with Reading or Foreseeing and never calls PairCredential's maker: the credential a send would make is named, and perform makes it on a send.
444 lines
14 KiB
Go
444 lines
14 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"flag"
|
|
"fmt"
|
|
"os"
|
|
"slices"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
|
"github.com/novox/mesh-controller/internal/artifacts"
|
|
"github.com/novox/mesh-controller/internal/inventory"
|
|
)
|
|
|
|
// What the records say the registries may keep, asked on demand (novox/hq ADR 0251, to-be 51).
|
|
//
|
|
// Three commands, each a verb of the mesh-controller seat:
|
|
//
|
|
// - `artifacts` — every reference the mesh recorded making, with its state: what the store's own
|
|
// tools set beside what the store holds;
|
|
// - `collect` — the sweep that runs after a build, run when a person asks, a dry run without
|
|
// --confirm;
|
|
// - `images` — every container image a machine's declaration names, now and as last sent: what
|
|
// a machine keeps when it prunes its images.
|
|
|
|
// artifactEntry is one recorded reference as `artifacts` answers it. Compact: the answer carries every
|
|
// kept and eligible reference the mesh has, in one bus message.
|
|
type artifactEntry struct {
|
|
Reference string `json:"reference"`
|
|
Repository string `json:"repository"`
|
|
Digest string `json:"digest"`
|
|
Kind string `json:"kind"`
|
|
State string `json:"state"`
|
|
Why []string `json:"why,omitempty"`
|
|
}
|
|
|
|
type artifactCounts struct {
|
|
Kept int `json:"kept"`
|
|
Eligible int `json:"eligible"`
|
|
Collected int `json:"collected"`
|
|
}
|
|
|
|
type artifactsAnswer struct {
|
|
Store string `json:"store"`
|
|
KeptBuilds int `json:"kept_builds"`
|
|
Counts artifactCounts `json:"counts"`
|
|
References []artifactEntry `json:"references"`
|
|
}
|
|
|
|
// artifactEntryOf reads a recorded reference into its repository, digest and kind; false for one that
|
|
// is not into the mesh's artifact store or names nothing by digest there.
|
|
func artifactEntryOf(s inventory.ArtifactState) (artifactEntry, bool) {
|
|
path, ours := catalogue.InArtifactStore(s.Reference)
|
|
if !ours {
|
|
return artifactEntry{}, false
|
|
}
|
|
e := artifactEntry{Reference: s.Reference, State: s.State, Why: s.Why}
|
|
if repository, digest, ok := strings.Cut(path, "@sha256:"); ok {
|
|
e.Repository, e.Digest, e.Kind = repository, "sha256:"+digest, "image"
|
|
return e, true
|
|
}
|
|
if repository, digest, ok := strings.Cut(path, "/blobs/sha256:"); ok {
|
|
e.Repository, e.Digest, e.Kind = repository, "sha256:"+digest, "archive"
|
|
return e, true
|
|
}
|
|
return artifactEntry{}, false
|
|
}
|
|
|
|
// artifactsOf is the answer for these states: counted whole, listed narrowed.
|
|
//
|
|
// **Collected references are listed only when asked for** (novox/hq ADR 0251, issue 314). They are
|
|
// history: every artifact the mesh ever let go stays collected for ever, so their number only grows,
|
|
// and a verb's answer is one bus message — an answer that outgrows it is lost on its way back. Kept
|
|
// and eligible are bounded by what the mesh holds now; the collected are counted, always.
|
|
func artifactsOf(states []inventory.ArtifactState, repository string, withCollected bool) artifactsAnswer {
|
|
a := artifactsAnswer{KeptBuilds: inventory.KeptBuilds, References: []artifactEntry{}}
|
|
for _, s := range states {
|
|
e, ours := artifactEntryOf(s)
|
|
if !ours {
|
|
continue
|
|
}
|
|
switch s.State {
|
|
case inventory.ArtifactKept:
|
|
a.Counts.Kept++
|
|
case inventory.ArtifactEligible:
|
|
a.Counts.Eligible++
|
|
case inventory.ArtifactCollected:
|
|
a.Counts.Collected++
|
|
}
|
|
if repository != "" && e.Repository != repository {
|
|
continue
|
|
}
|
|
if s.State == inventory.ArtifactCollected && !withCollected {
|
|
continue
|
|
}
|
|
a.References = append(a.References, e)
|
|
}
|
|
return a
|
|
}
|
|
|
|
func artifactsCommand(ctx context.Context, args []string) error {
|
|
set := flag.NewFlagSet("artifacts", flag.ContinueOnError)
|
|
asJSON := set.Bool("json", false, "answer as JSON")
|
|
repository := set.String("repository", "", "only this repository's references (<module>/<artifact>); the counts stay whole")
|
|
collected := set.Bool("collected", false, "list the collected references too")
|
|
positionals, err := parseAround(set, args)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if len(positionals) != 0 {
|
|
return errors.New("artifacts [--json] [--repository <module>/<artifact>] [--collected]")
|
|
}
|
|
open, err := openStores(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer open.Close()
|
|
states, err := open.inventory.Artifacts(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
answer := artifactsOf(states, strings.TrimSpace(*repository), *collected)
|
|
if shelf, err := open.inventory.Catalogue(ctx); err == nil {
|
|
answer.Store, _ = artifactStoreAddress(ctx, open.inventory, shelf, "")
|
|
}
|
|
if *asJSON {
|
|
encoder := json.NewEncoder(os.Stdout)
|
|
encoder.SetIndent("", " ")
|
|
return encoder.Encode(answer)
|
|
}
|
|
fmt.Printf("artifact store %s\nkept %d\neligible %d\ncollected %d\n",
|
|
storeOrNone(answer.Store), answer.Counts.Kept, answer.Counts.Eligible, answer.Counts.Collected)
|
|
for _, e := range answer.References {
|
|
why := ""
|
|
if len(e.Why) > 0 {
|
|
why = " (" + strings.Join(e.Why, ", ") + ")"
|
|
}
|
|
fmt.Printf(" %-9s %s%s\n", e.State, e.Reference, why)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func storeOrNone(s string) string {
|
|
if s == "" {
|
|
return "(not on the network)"
|
|
}
|
|
return s
|
|
}
|
|
|
|
// collectAnswer is what `collect` did, or — a dry run — would do.
|
|
type collectAnswer struct {
|
|
DryRun bool `json:"dry_run"`
|
|
Store string `json:"store"`
|
|
KeptArchives int `json:"kept_archives"`
|
|
// Held is how many kept archives the store holds by a manifest: asked with HEADs in a dry run,
|
|
// after the holding in a real one.
|
|
Held int `json:"held"`
|
|
Unheld int `json:"unheld"`
|
|
HoldersWritten int `json:"holders_written"`
|
|
Missing int `json:"missing"`
|
|
Eligible int `json:"eligible"`
|
|
// WouldLetGo is the eligible references, oldest first, at most mostListed of them.
|
|
WouldLetGo []string `json:"would_let_go"`
|
|
NotListed int `json:"not_listed,omitempty"`
|
|
LetGo []string `json:"let_go"`
|
|
Skipped int `json:"skipped"`
|
|
Left int `json:"left"`
|
|
Stopped string `json:"stopped,omitempty"`
|
|
}
|
|
|
|
// mostListed is how many references `collect` lists by name; the rest are counted.
|
|
const mostListed = 1000
|
|
|
|
// onDemand are the bounds of a sweep a person asks for: more than after a build, inside a module's
|
|
// sixty-second ask (novox/hq ADR 0251).
|
|
const (
|
|
collectMostDefault = 500
|
|
collectMostAtMost = 5000
|
|
collectBudget = 45 * time.Second
|
|
)
|
|
|
|
// causeCollect is the cause `collect` records when none is given.
|
|
const causeCollect = "collect-on-demand"
|
|
|
|
func collectCommand(ctx context.Context, args []string) error {
|
|
set := flag.NewFlagSet("collect", flag.ContinueOnError)
|
|
asJSON := set.Bool("json", false, "answer as JSON")
|
|
confirm := set.Bool("confirm", false, "let go of what is eligible, rather than only say what would go")
|
|
most := set.Int("most", collectMostDefault, "let go of at most this many")
|
|
f := addHandActFlags(set)
|
|
positionals, err := parseAround(set, args)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if len(positionals) != 0 {
|
|
return errors.New("collect [--json] [--confirm --why <text>] [--most <n>]")
|
|
}
|
|
if *confirm {
|
|
// A deletion a person asks for says why, before anything is asked of the store.
|
|
if err := f.require("collect"); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if *most < 1 {
|
|
return fmt.Errorf("collect: --most is at least 1, not %d", *most)
|
|
}
|
|
bounds := sweepBounds{most: min(*most, collectMostAtMost), budget: collectBudget, platforms: true}
|
|
|
|
open, err := openStores(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer open.Close()
|
|
inv := open.inventory
|
|
references, err := inv.ToCollect(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
kept, err := inv.KeptArchives(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
shelf, err := inv.Catalogue(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
address, err := artifactStoreAddress(ctx, inv, shelf, "")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if *confirm {
|
|
if strings.TrimSpace(*f.cause) == "" {
|
|
*f.cause = causeCollect
|
|
}
|
|
f.record(ctx, "collect", []string{"--most", fmt.Sprint(bounds.most)})
|
|
}
|
|
answer := runCollect(ctx, inv, artifacts.Store{Address: address}, references, kept, *confirm, bounds)
|
|
if *asJSON {
|
|
encoder := json.NewEncoder(os.Stdout)
|
|
encoder.SetIndent("", " ")
|
|
return encoder.Encode(answer)
|
|
}
|
|
printCollect(answer)
|
|
return nil
|
|
}
|
|
|
|
// runCollect is `collect` against a store: a dry run asks only HEADs, a real one is the sweep.
|
|
func runCollect(ctx context.Context, inv *inventory.Inventory, store artifacts.Store, references, kept []string,
|
|
real bool, bounds sweepBounds) collectAnswer {
|
|
a := collectAnswer{DryRun: !real, Store: store.Address, KeptArchives: len(kept), Eligible: len(references),
|
|
WouldLetGo: []string{}, LetGo: []string{}}
|
|
listed := references
|
|
if len(listed) > mostListed {
|
|
listed, a.NotListed = listed[:mostListed], len(listed)-mostListed
|
|
}
|
|
a.WouldLetGo = append(a.WouldLetGo, listed...)
|
|
if store.Address == "" {
|
|
a.Stopped = "this mesh has no artifact store on its network: nothing was asked of it"
|
|
a.Left = len(references)
|
|
return a
|
|
}
|
|
if !real {
|
|
// Reads only: each kept archive asked about with HEADs, as `collection` does.
|
|
report := collectionReport{Unheld: []string{}, Missing: []string{}}
|
|
unasked, stopped := askHeld(ctx, store, kept, &report)
|
|
a.Held, a.Unheld, a.Missing = report.Held, len(report.Unheld), len(report.Missing)
|
|
if unasked > 0 {
|
|
a.Stopped = fmt.Sprintf("%d kept archive(s) not asked about: %s", unasked, stopped)
|
|
}
|
|
return a
|
|
}
|
|
r := sweep(ctx, inv, store, references, kept, bounds)
|
|
a.HoldersWritten, a.Missing, a.Skipped, a.Left, a.Stopped = r.HoldersWritten, r.Missing, r.Skipped, r.Left, r.Stopped
|
|
if r.Held {
|
|
a.Held = len(kept) - r.Missing
|
|
}
|
|
a.LetGo = append(a.LetGo, r.LetGo...)
|
|
return a
|
|
}
|
|
|
|
func printCollect(a collectAnswer) {
|
|
if a.DryRun {
|
|
fmt.Println("a dry run: nothing was held and nothing was let go (--confirm --why <text> to collect)")
|
|
}
|
|
fmt.Printf("artifact store %s\n", storeOrNone(a.Store))
|
|
fmt.Printf("kept archives %d (held %d, unheld %d, missing %d)\n", a.KeptArchives, a.Held, a.Unheld, a.Missing)
|
|
if a.HoldersWritten > 0 {
|
|
fmt.Printf("holders written %d\n", a.HoldersWritten)
|
|
}
|
|
fmt.Printf("eligible to let go %d\n", a.Eligible)
|
|
if !a.DryRun {
|
|
fmt.Printf("let go %d (skipped %d, left %d)\n", len(a.LetGo), a.Skipped, a.Left)
|
|
}
|
|
if a.Stopped != "" {
|
|
fmt.Printf("stopped: %s\n", a.Stopped)
|
|
}
|
|
for _, reference := range a.WouldLetGo {
|
|
fmt.Printf(" %s\n", reference)
|
|
}
|
|
if a.NotListed > 0 {
|
|
fmt.Printf(" … and %d more not listed\n", a.NotListed)
|
|
}
|
|
}
|
|
|
|
// declaredImage is one image a machine's declaration names.
|
|
type declaredImage struct {
|
|
Image string `json:"image"`
|
|
Resources []string `json:"resources"`
|
|
// In says which declarations name it: "declaration" (what the mesh would send now), "sent" (what
|
|
// the machine was last sent).
|
|
In []string `json:"in"`
|
|
}
|
|
|
|
type imagesAnswer struct {
|
|
Node string `json:"node"`
|
|
Images []declaredImage `json:"images"`
|
|
// SentKnown is whether what the machine was last sent is kept with its images; false when it is
|
|
// not kept, or was kept before the images were.
|
|
SentKnown bool `json:"sent_known"`
|
|
}
|
|
|
|
// imagesOf is every container image of the declaration about to be sent and of the summary of the one
|
|
// last sent.
|
|
func imagesOf(node string, body []byte, sent []sentResource, sentKept bool) (imagesAnswer, error) {
|
|
a := imagesAnswer{Node: node, Images: []declaredImage{}, SentKnown: sentKept}
|
|
byImage := map[string]*declaredImage{}
|
|
add := func(image, resource, in string) {
|
|
if image == "" {
|
|
return
|
|
}
|
|
d := byImage[image]
|
|
if d == nil {
|
|
d = &declaredImage{Image: image}
|
|
byImage[image] = d
|
|
}
|
|
if !slices.Contains(d.Resources, resource) {
|
|
d.Resources = append(d.Resources, resource)
|
|
}
|
|
if !slices.Contains(d.In, in) {
|
|
d.In = append(d.In, in)
|
|
}
|
|
}
|
|
if body != nil {
|
|
var envelope struct {
|
|
Resources []map[string]any `json:"resources"`
|
|
}
|
|
if err := json.Unmarshal(body, &envelope); err != nil {
|
|
return a, fmt.Errorf("a declaration that is not one: %w", err)
|
|
}
|
|
for _, r := range envelope.Resources {
|
|
if r["type"] != "container" {
|
|
continue
|
|
}
|
|
image, _ := r["image"].(string)
|
|
add(image, fmt.Sprint(r["id"]), "declaration")
|
|
}
|
|
}
|
|
for _, s := range sent {
|
|
if s.Type != "container" {
|
|
continue
|
|
}
|
|
if s.Image == "" {
|
|
// Kept before the summary carried images: what was sent cannot be said.
|
|
a.SentKnown = false
|
|
continue
|
|
}
|
|
add(s.Image, s.ID, "sent")
|
|
}
|
|
for _, d := range byImage {
|
|
sort.Strings(d.Resources)
|
|
a.Images = append(a.Images, *d)
|
|
}
|
|
sort.Slice(a.Images, func(i, j int) bool { return a.Images[i].Image < a.Images[j].Image })
|
|
return a, nil
|
|
}
|
|
|
|
func imagesCommand(ctx context.Context, args []string) error {
|
|
set := flag.NewFlagSet("images", flag.ContinueOnError)
|
|
asJSON := set.Bool("json", false, "answer as JSON")
|
|
positionals, err := parseAround(set, args)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if len(positionals) != 1 {
|
|
return errors.New("images <node> [--json]")
|
|
}
|
|
node := positionals[0]
|
|
open, err := openStores(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer open.Close()
|
|
if _, err := open.inventory.NodeByName(ctx, node); err != nil {
|
|
return fmt.Errorf("%s: %w", node, err)
|
|
}
|
|
// What the mesh would send now, composed as `plan` composes it.
|
|
var body []byte
|
|
plan, settings, err := resolveNode(ctx, open, node)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if len(plan.Modules) > 0 {
|
|
declared, err := declarationFor(ctx, open, node, plan, settings)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if body, err = declared.Body(); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
// And what it was last sent, as `plan --diff` compares against.
|
|
prev, err := open.inventory.SentSummary(ctx, node)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
var sent []sentResource
|
|
if len(prev) > 0 {
|
|
if err := json.Unmarshal(prev, &sent); err != nil {
|
|
return fmt.Errorf("%s: what it was last sent is kept in a form this version does not read: %w", node, err)
|
|
}
|
|
}
|
|
answer, err := imagesOf(node, body, sent, len(prev) > 0)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if *asJSON {
|
|
encoder := json.NewEncoder(os.Stdout)
|
|
encoder.SetIndent("", " ")
|
|
return encoder.Encode(answer)
|
|
}
|
|
for _, d := range answer.Images {
|
|
fmt.Printf("%s (%s; %s)\n", d.Image, strings.Join(d.In, ", "), strings.Join(d.Resources, ", "))
|
|
}
|
|
if !answer.SentKnown {
|
|
fmt.Printf("%s: which images it was last sent is not kept — the next push keeps it\n", node)
|
|
}
|
|
return nil
|
|
}
|