CAT-D1, D2, D3b, D4, D6 (wiring): distribution's program, its tools bundle and its manifest
The program serves the store's port 5000 in front of the registry container, now on the node's loopback (127.0.0.1:5001), listed before the program so the port is free when it starts (E14). The tools bundle serves each seat's verbs as <seat>.<verb> and registry_incoming, registry_holds, registry_old_name and registry_direct_writes, on distribution's own account, with the caller the tool runner proved. The manifest claims container-image-registry and oci-artifact-registry with their verbs, keeps mesh-artifact-store and provides artifact-store until the old name closes (E15), asks its bus state in requires (incoming, transfers, holds, windows, findings, events) and declares its health outside the program. The bootstrap's one-off credential is discarded at the first publish answered. Not yet: the runtime's object store helper for incoming (mesh-sdk), the windows' driver on delivered and manifest-reads-changed, the read-side health probe calling versions per seat. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
This commit is contained in:
@@ -0,0 +1,280 @@
|
||||
// Command distribution-tools serves distribution's seat verbs and tools (novox/hq design 51): the verbs of
|
||||
// container-image-registry and oci-artifact-registry, each as `<seat>.<verb>`, and registry_incoming, registry_holds,
|
||||
// registry_old_name and registry_direct_writes. The runtime launches it on distribution's own account; the caller of
|
||||
// every verb is the one the tool runner proved (stdio.CallerOf), never a name in the arguments. stdout is the MCP
|
||||
// channel; what it says, it says on stderr.
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/url"
|
||||
"os"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/health"
|
||||
|
||||
"distribution/internal/holder"
|
||||
)
|
||||
|
||||
func logf(format string, a ...any) { fmt.Fprintf(os.Stderr, "[distribution-tools] "+format+"\n", a...) }
|
||||
|
||||
// kept is one state of distribution's, through the runtime.
|
||||
type kept[T any] struct {
|
||||
state string
|
||||
key func(T) string
|
||||
}
|
||||
|
||||
func (k kept[T]) Get(key string) (T, bool, error) {
|
||||
var v T
|
||||
e, err := stdio.State(k.state).Get(key)
|
||||
if err != nil || e == nil {
|
||||
return v, false, err
|
||||
}
|
||||
return v, true, json.Unmarshal(e.Value, &v)
|
||||
}
|
||||
func (k kept[T]) Put(v T) error {
|
||||
_, err := stdio.State(k.state).Put(k.key(v), v)
|
||||
return err
|
||||
}
|
||||
func (k kept[T]) Delete(key string) error { return stdio.State(k.state).Delete(key) }
|
||||
func (k kept[T]) List() ([]T, error) {
|
||||
keys, err := stdio.State(k.state).Keys()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var out []T
|
||||
for _, key := range keys {
|
||||
if v, ok, err := k.Get(key); err == nil && ok {
|
||||
out = append(out, v)
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// errNoObjects is said while the runtime offers a bundle no object store: the helper is mesh-sdk's (ADR 0611 §7, the
|
||||
// parts helper in mesh-sdk's runtime), asked of the milestone's mesh-sdk work. Until then a part is refused loudly.
|
||||
var errNoObjects = errors.New("incoming cannot be written: the runtime's object store helper (mesh-sdk, ADR 0611 §7) is not in this build")
|
||||
|
||||
type objects struct{}
|
||||
|
||||
func (objects) Put(string, []byte) error { return errNoObjects }
|
||||
func (objects) Get(string) ([]byte, error) { return nil, errNoObjects }
|
||||
func (objects) Delete(string) error { return nil }
|
||||
func (objects) Used() (uint64, error) { return 0, nil }
|
||||
|
||||
// builderVersions asks the builder's builder_versions (ADR 0605 §4).
|
||||
type builderVersions struct{}
|
||||
|
||||
func (builderVersions) Recorded(name, version, digest string) (bool, string, error) {
|
||||
raw, err := stdio.Ask("mesh-delivery-builder.builder_versions", map[string]any{"name": name, "version": version, "digest": digest})
|
||||
if err != nil {
|
||||
return false, "", err
|
||||
}
|
||||
var a struct {
|
||||
Recorded *bool `json:"recorded"`
|
||||
Why string `json:"why"`
|
||||
}
|
||||
if err := json.Unmarshal(raw, &a); err != nil || a.Recorded == nil {
|
||||
return false, "", fmt.Errorf("builder_versions answered no recorded: %s", raw)
|
||||
}
|
||||
return *a.Recorded, a.Why, nil
|
||||
}
|
||||
|
||||
func say(seat, event string, body holder.Event) error {
|
||||
return stdio.SeatPublish("mesh.seat."+seat+".event."+event, body, "")
|
||||
}
|
||||
|
||||
func finding(kind, subject, words string, raised bool) {
|
||||
key := health.FindingKey(health.ScopeSeat, "distribution", kind+"."+strings.NewReplacer(".", "_", "@", "_", "/", "_").Replace(subject))
|
||||
if !raised {
|
||||
_ = stdio.State(health.FindingsState).Delete(key)
|
||||
return
|
||||
}
|
||||
f := health.Finding{V: health.FindingVersion, Key: key, Kind: kind, Ref: key[strings.LastIndex(key, ".")+1:],
|
||||
Subject: health.Subject{Scope: health.ScopeSeat, ID: "distribution"}, Severity: health.Warning,
|
||||
Summary: words, Headline: words, Explanation: words, Resolved: "cleared once it no longer holds",
|
||||
Source: kind, Observer: "distribution", At: time.Now().UTC()}
|
||||
if _, err := stdio.State(health.FindingsState).Put(key, f); err != nil {
|
||||
logf("the finding %s could not be written: %v", key, err)
|
||||
}
|
||||
}
|
||||
|
||||
// removeBootstrap discards the bootstrap's one-off credential once a publish is answered (design 51 *On a fresh mesh*).
|
||||
var removeBootstrap = sync.OnceFunc(func() {
|
||||
if err := os.Remove(holder.BootstrapFile); err == nil {
|
||||
logf("the bootstrap's one-off credential is discarded: the holders serve publish")
|
||||
}
|
||||
})
|
||||
|
||||
func arg[T any](args map[string]any) (T, error) {
|
||||
var v T
|
||||
clean := map[string]any{}
|
||||
for k, x := range args {
|
||||
if k != "_meta" {
|
||||
clean[k] = x
|
||||
}
|
||||
}
|
||||
b, _ := json.Marshal(clean)
|
||||
return v, json.Unmarshal(b, &v)
|
||||
}
|
||||
|
||||
func str(args map[string]any, k string) string { s, _ := args[k].(string); return strings.TrimSpace(s) }
|
||||
|
||||
func tools(v *holder.Verbs, windows kept[holder.Window]) []stdio.Tool {
|
||||
obj := func(props map[string]any, req ...string) map[string]any {
|
||||
return map[string]any{"type": "object", "required": req, "properties": props}
|
||||
}
|
||||
s := func(d string) map[string]any { return map[string]any{"type": "string", "description": d} }
|
||||
var out []stdio.Tool
|
||||
for _, seat := range holder.Seats {
|
||||
out = append(out,
|
||||
stdio.Tool{Name: seat + "." + holder.VerbPublish, Description: "Open a transfer of a publish (ADR 0611): its name, version, kind, digest and parts; answers the parts the holder lacks.",
|
||||
Input: obj(map[string]any{"transfer": s("the transfer's id"), "name": s("<module>/<artifact>"), "version": s("the builder's version"), "kind": s("image, bundle, archive or upstream"), "digest": s("sha256:…"), "parts": map[string]any{"type": "array"}}, "transfer", "name", "kind", "digest", "parts"),
|
||||
Run: func(args map[string]any) (any, error) {
|
||||
r, err := arg[holder.PublishRequest](args)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
_, _ = v.DropIdle()
|
||||
a, err := v.Publish(seat, stdio.CallerOf(args), r)
|
||||
if err == nil {
|
||||
removeBootstrap()
|
||||
}
|
||||
return a, err
|
||||
}},
|
||||
stdio.Tool{Name: seat + "." + holder.VerbPublishPart, Description: "Send one chunk (at most 1 MiB) of one part of an open transfer, from the caller that opened it.",
|
||||
Input: obj(map[string]any{"transfer": s("its id"), "digest": s("the part's digest"), "sequence": map[string]any{"type": "integer"}, "last": map[string]any{"type": "boolean"}, "data": s("base64")}, "transfer", "digest", "data"),
|
||||
Run: func(args map[string]any) (any, error) {
|
||||
r, err := arg[holder.PartRequest](args)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return v.PublishPart(seat, stdio.CallerOf(args), r)
|
||||
}},
|
||||
stdio.Tool{Name: seat + "." + holder.VerbVersions, Description: "The versions the seat holds of a name.", Input: obj(map[string]any{"name": s("<module>/<artifact>")}, "name"),
|
||||
Run: func(args map[string]any) (any, error) {
|
||||
return v.Versions(seat, stdio.CallerOf(args), str(args, "name"))
|
||||
}},
|
||||
stdio.Tool{Name: seat + "." + holder.VerbRemove, Description: "Remove a tagged version (the builder), or for retention an untagged one by its digest (the artifact sweep); says removed.",
|
||||
Input: obj(map[string]any{"name": s("name"), "version": s("a tagged version"), "digest": s("for retention"), "kind": s("kind"), "reason": s("why")}, "name", "reason"),
|
||||
Run: func(args map[string]any) (any, error) {
|
||||
r, err := arg[holder.RemoveRequest](args)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return map[string]any{"removed": true}, v.Remove(seat, stdio.CallerOf(args), r)
|
||||
}},
|
||||
stdio.Tool{Name: seat + "." + holder.VerbHold, Description: "Keep one untagged bundle, archive or copied base from retention (the artifact sweep alone); says held.",
|
||||
Input: obj(map[string]any{"name": s("name"), "kind": s("bundle, archive or upstream"), "digest": s("sha256:…"), "why": s("why")}, "name", "kind", "digest"),
|
||||
Run: func(args map[string]any) (any, error) {
|
||||
r, err := arg[holder.HoldRequest](args)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return map[string]any{"held": true}, v.HoldOne(seat, stdio.CallerOf(args), r)
|
||||
}},
|
||||
stdio.Tool{Name: seat + "." + holder.VerbRelease, Description: "Let a hold go (the artifact sweep, or the operator's session through registry_holds); says released.",
|
||||
Input: obj(map[string]any{"name": s("name"), "digest": s("sha256:…"), "why": s("why")}, "name", "digest"),
|
||||
Run: func(args map[string]any) (any, error) {
|
||||
r, err := arg[holder.HoldRequest](args)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return map[string]any{"released": true}, v.Release(seat, stdio.CallerOf(args), r)
|
||||
}},
|
||||
)
|
||||
}
|
||||
out = append(out,
|
||||
stdio.Tool{Name: holder.ImageSeat + "." + holder.VerbTags, Description: "The tags of an image.", Input: obj(map[string]any{"name": s("name")}, "name"),
|
||||
Run: func(args map[string]any) (any, error) {
|
||||
return v.Tags(holder.ImageSeat, stdio.CallerOf(args), str(args, "name"))
|
||||
}},
|
||||
stdio.Tool{Name: holder.ImageSeat + "." + holder.VerbMirror, Description: "Where the store holds a copy of an outside image, by its pinned digest (mesh-build-image alone).",
|
||||
Input: obj(map[string]any{"source": s("<host>/<path>"), "digest": s("sha256:…")}, "source", "digest"),
|
||||
Run: func(args map[string]any) (any, error) {
|
||||
r, err := arg[holder.MirrorRequest](args)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
ref, err := v.Mirror(holder.ImageSeat, stdio.CallerOf(args), r)
|
||||
return map[string]any{"reference": ref}, err
|
||||
}},
|
||||
stdio.Tool{Name: "registry_incoming", Description: "Open transfers of incoming and its fill; with drop, drops one by hand (a hand act, why required).",
|
||||
Input: obj(map[string]any{"drop": s("a transfer id to drop"), "why": s("why it is dropped")}),
|
||||
Run: func(args map[string]any) (any, error) {
|
||||
idle, _ := v.DropIdle()
|
||||
if id := str(args, "drop"); id != "" {
|
||||
if str(args, "why") == "" {
|
||||
return nil, errors.New("a drop is a hand act: say why")
|
||||
}
|
||||
t, err := v.Drop(id)
|
||||
logf("transfer %s dropped by hand by %s: %s", id, stdio.CallerOf(args), str(args, "why"))
|
||||
return map[string]any{"dropped": t.Transfer}, err
|
||||
}
|
||||
all, err := v.Transfers.List()
|
||||
used, _ := v.Objects.Used()
|
||||
return map[string]any{"transfers": all, "used": used, "cap": holder.IncomingCap, "dropped_idle": idle}, err
|
||||
}},
|
||||
stdio.Tool{Name: "registry_holds", Description: "What is held, since when and for whom; with release, lets one go through the seat's release as the operator's session (a hand act).",
|
||||
Input: obj(map[string]any{"seat": s("the seat"), "release": s("the name to release"), "digest": s("its digest"), "why": s("why")}),
|
||||
Run: func(args map[string]any) (any, error) {
|
||||
if name := str(args, "release"); name != "" {
|
||||
if str(args, "why") == "" {
|
||||
return nil, errors.New("a release by hand is a hand act: say why")
|
||||
}
|
||||
// The caller proved on this call (the operator's session) is carried on; distribution's own never.
|
||||
return map[string]any{"released": name}, v.Release(str(args, "seat"), stdio.CallerOf(args), holder.HoldRequest{Name: name, Digest: str(args, "digest"), Why: str(args, "why")})
|
||||
}
|
||||
return v.HeldNow()
|
||||
}},
|
||||
)
|
||||
for _, w := range []string{holder.OldNameWindow, holder.DirectWritesWindow} {
|
||||
name := "registry_old_name"
|
||||
if w == holder.DirectWritesWindow {
|
||||
name = "registry_direct_writes"
|
||||
}
|
||||
out = append(out, stdio.Tool{Name: name, Description: "Whether the " + w + " window is open and the nodes holding it, since when; with drop, drops a retired node from it by hand (a hand act, why required).",
|
||||
Input: obj(map[string]any{"drop": s("a retired node"), "why": s("why")}),
|
||||
Run: func(args map[string]any) (any, error) {
|
||||
cur, _, err := windows.Get(w)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
cur.Name = w
|
||||
if node := str(args, "drop"); node != "" {
|
||||
if cur, err = cur.Drop(node, stdio.CallerOf(args), str(args, "why"), time.Now().UTC()); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := windows.Put(cur); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
_, bootstrap := os.Stat(holder.BootstrapFile)
|
||||
return map[string]any{"window": cur, "bootstrap_credential_present": bootstrap == nil}, nil
|
||||
}})
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func main() {
|
||||
callers, err := holder.ParseCallers(os.Getenv("MESH_CONFIG_CALLERS"))
|
||||
if err != nil {
|
||||
logf("%v", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
base, _ := url.Parse("http://127.0.0.1:5001")
|
||||
store := &holder.OCI{Base: base, Registry: os.Getenv("MESH_REGISTRY_REACH")}
|
||||
h := &holder.Holder{Callers: callers, Objects: objects{}, Builder: builderVersions{}, Store: store, Say: say, Finding: finding,
|
||||
Transfers: kept[holder.Transfer]{state: holder.TransfersState, key: func(t holder.Transfer) string { return t.Transfer }}}
|
||||
v := &holder.Verbs{Holder: h, Catalog: store, Holds: kept[holder.Hold]{state: "holds", key: holder.Hold.Key}}
|
||||
windows := kept[holder.Window]{state: "windows", key: func(w holder.Window) string { return w.Name }}
|
||||
if err := stdio.Serve("", tools(v, windows)); err != nil {
|
||||
logf("%v", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,117 @@
|
||||
// Command distribution is distribution's program (novox/hq design 51): it serves the artifact store's port in front of
|
||||
// the registry container, which listens on its node's loopback only. Reads (GET, HEAD) go through to every node's
|
||||
// container runtime; a write is refused but while the direct-write window is open, never over a tag the seat holds
|
||||
// published, or with the bootstrap's one-off credential (holder.Front). The window is distribution's state windows,
|
||||
// written by its tools bundle; the program reads it on each write and holds no timer.
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"os"
|
||||
"os/signal"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"distribution/internal/holder"
|
||||
)
|
||||
|
||||
func main() {
|
||||
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
|
||||
defer stop()
|
||||
if err := run(ctx, os.Getenv); err != nil {
|
||||
fmt.Fprintln(os.Stderr, "distribution:", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
}
|
||||
|
||||
func run(ctx context.Context, getenv func(string) string) error {
|
||||
logger := log.New(os.Stderr, "distribution: ", log.LstdFlags)
|
||||
store, err := url.Parse(or(getenv("MESH_REGISTRY_LOOPBACK"), "http://127.0.0.1:5001"))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
oci := &holder.OCI{Base: store}
|
||||
var open atomic.Bool
|
||||
open.Store(true) // until the window's state says it closed: no wider than today's store
|
||||
cred, err := process.ReadBrokerAccount(getenv)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
conn, js, err := cred.Dial()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer conn.Close()
|
||||
// The direct-write window's record, written by the tools bundle on distribution's own account.
|
||||
windows := process.OptionalBucket(ctx, js, "distribution_windows", "the direct-write window is read as open until the program starts again", logger.Printf)
|
||||
if windows != nil {
|
||||
go follow(ctx, windows, &open, logger)
|
||||
}
|
||||
front := &holder.Front{Store: store, Window: &open, Logf: logger.Printf,
|
||||
Published: func(name, tag string) (bool, error) {
|
||||
d, err := oci.Digest(name, tag)
|
||||
return d != "", err
|
||||
},
|
||||
Bootstrap: func() (string, bool) {
|
||||
b, err := os.ReadFile(holder.BootstrapFile)
|
||||
if err != nil {
|
||||
return "", false
|
||||
}
|
||||
return strings.TrimSpace(string(b)), true
|
||||
}}
|
||||
srv := &http.Server{Addr: or(getenv("MESH_REGISTRY_LISTEN"), ":5000"), Handler: front, ReadHeaderTimeout: 30 * time.Second}
|
||||
errs := make(chan error, 1)
|
||||
go func() { errs <- srv.ListenAndServe() }()
|
||||
logger.Printf("serving the store's port %s in front of %s", srv.Addr, store)
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
shut, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
|
||||
defer cancel()
|
||||
return srv.Shutdown(shut)
|
||||
case err := <-errs:
|
||||
if errors.Is(err, http.ErrServerClosed) {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
// follow keeps open as the direct-write window's record says, on every change of it (a watch, no timer).
|
||||
func follow(ctx context.Context, kv jetstream.KeyValue, open *atomic.Bool, logger *log.Logger) {
|
||||
w, err := kv.Watch(ctx, holder.DirectWritesWindow)
|
||||
if err != nil {
|
||||
logger.Printf("the direct-write window cannot be watched (%v): read as open", err)
|
||||
return
|
||||
}
|
||||
for e := range w.Updates() {
|
||||
if e == nil || e.Operation() != jetstream.KeyValuePut {
|
||||
continue
|
||||
}
|
||||
var win holder.Window
|
||||
if json.Unmarshal(e.Value(), &win) != nil {
|
||||
continue
|
||||
}
|
||||
closed := !win.Open && !win.Closed.IsZero()
|
||||
if closed && open.Swap(false) {
|
||||
logger.Printf("the direct-write window is closed: every write but distribution's own is refused, naming publish")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func or(v, d string) string {
|
||||
if strings.TrimSpace(v) == "" {
|
||||
return d
|
||||
}
|
||||
return v
|
||||
}
|
||||
@@ -1,3 +1,16 @@
|
||||
module distribution
|
||||
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261103221822-85ba0a9a70d8
|
||||
github.com/nats-io/nats.go v1.54.0
|
||||
)
|
||||
|
||||
require (
|
||||
github.com/klauspost/compress v1.20.0 // indirect
|
||||
github.com/nats-io/nkeys v0.4.16 // indirect
|
||||
github.com/nats-io/nuid v1.0.1 // indirect
|
||||
golang.org/x/crypto v0.57.0 // indirect
|
||||
golang.org/x/sys v0.48.0 // indirect
|
||||
)
|
||||
|
||||
@@ -0,0 +1,26 @@
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261103221822-85ba0a9a70d8 h1:XEbIVImp6yGjA91rp0iZBoDdL5KfamjOuDLwWvqh0hc=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261103221822-85ba0a9a70d8/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
|
||||
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
|
||||
github.com/google/go-tpm v0.9.8/go.mod h1:h9jEsEECg7gtLis0upRBQU+GhYVH6jMjrFxI8u6bVUY=
|
||||
github.com/klauspost/compress v1.20.0 h1:a3C1ke2ohxFymNlb2HWAHjDeKCI90scRskErZkR0ezA=
|
||||
github.com/klauspost/compress v1.20.0/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI=
|
||||
github.com/minio/highwayhash v1.0.4 h1:asJizugGgchQod2ja9NJlGOWq4s7KsAWr5XUc9Clgl4=
|
||||
github.com/minio/highwayhash v1.0.4/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ=
|
||||
github.com/nats-io/jwt/v2 v2.8.1 h1:V0xpGuD/N8Mi+fQNDynXohVvp7ZztevW5io8CUWlPmU=
|
||||
github.com/nats-io/jwt/v2 v2.8.1/go.mod h1:nWnOEEiVMiKHQpnAy4eXlizVEtSfzacZ1Q43LIRavZg=
|
||||
github.com/nats-io/nats-server/v2 v2.11.17 h1:GKEghcFK6A+aFx11Yf1LjgLC3txAwvyhnYzhBIQZA8I=
|
||||
github.com/nats-io/nats-server/v2 v2.11.17/go.mod h1:B1sFVz4StNosQ903ak4N1G01Fl/9f8e06mXpFIE2K24=
|
||||
github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA=
|
||||
github.com/nats-io/nats.go v1.54.0/go.mod h1:y+DZoD1oBOYfZTU681eTUiUjI0vbqYGixNVFHcjHJ0k=
|
||||
github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg=
|
||||
github.com/nats-io/nkeys v0.4.16/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs=
|
||||
github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
|
||||
github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
|
||||
golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M=
|
||||
golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA=
|
||||
golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo=
|
||||
golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og=
|
||||
golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=
|
||||
golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno=
|
||||
|
||||
@@ -3,10 +3,12 @@ package holder
|
||||
import (
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"net/url"
|
||||
"os"
|
||||
"reflect"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
@@ -383,4 +385,51 @@ func TestTheHolderOfTheTwoImageSeats(t *testing.T) {
|
||||
t.Error("a closed window closed again")
|
||||
}
|
||||
})
|
||||
t.Run("the manifest", func(t *testing.T) {
|
||||
raw, err := os.ReadFile("../../module.json")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var m struct {
|
||||
Provides []struct{ Name string } `json:"provides"`
|
||||
Claims []struct {
|
||||
Name string `json:"name"`
|
||||
Serves []string `json:"serves"`
|
||||
} `json:"claims"`
|
||||
Resources []map[string]any `json:"resources"`
|
||||
}
|
||||
if err := json.Unmarshal(raw, &m); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// E15: artifact-store is still offered until the old name closes, so a revert needs no edge.
|
||||
if len(m.Provides) != 1 || m.Provides[0].Name != "artifact-store" {
|
||||
t.Errorf("provides = %+v, want artifact-store kept", m.Provides)
|
||||
}
|
||||
claims := map[string][]string{}
|
||||
for _, c := range m.Claims {
|
||||
claims[c.Name] = c.Serves
|
||||
}
|
||||
if !reflect.DeepEqual(claims[ImageSeat], VerbsOf(ImageSeat)) || !reflect.DeepEqual(claims[ArtifactSeat], VerbsOf(ArtifactSeat)) {
|
||||
t.Errorf("claims = %v, want each seat serving its verbs", claims)
|
||||
}
|
||||
if _, ok := claims[OldSeat]; !ok {
|
||||
t.Error("the old seat mesh-artifact-store is no longer claimed while its window is open")
|
||||
}
|
||||
// E14: the registry container, on the loopback, is listed before the program, so the port is free when it starts.
|
||||
store, program := -1, -1
|
||||
for i, r := range m.Resources {
|
||||
switch r["id"] {
|
||||
case "store":
|
||||
store = i
|
||||
if ports, _ := r["ports"].([]any); len(ports) != 1 || ports[0] != "127.0.0.1:5001:5000" {
|
||||
t.Errorf("the store's ports = %v, want the loopback only", r["ports"])
|
||||
}
|
||||
case "program":
|
||||
program = i
|
||||
}
|
||||
}
|
||||
if store < 0 || program < 0 || store > program {
|
||||
t.Errorf("store at %d, program at %d: want the store listed first", store, program)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"module": "distribution",
|
||||
"version": "1",
|
||||
"version": "2",
|
||||
"provides": [
|
||||
{
|
||||
"name": "artifact-store",
|
||||
@@ -11,17 +11,111 @@
|
||||
}
|
||||
],
|
||||
"claims": [
|
||||
{
|
||||
"name": "container-image-registry",
|
||||
"scope": "mesh",
|
||||
"serves": [
|
||||
"hold",
|
||||
"mirror",
|
||||
"publish",
|
||||
"publish-part",
|
||||
"release",
|
||||
"remove",
|
||||
"tags",
|
||||
"versions"
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "oci-artifact-registry",
|
||||
"scope": "mesh",
|
||||
"serves": [
|
||||
"hold",
|
||||
"publish",
|
||||
"publish-part",
|
||||
"release",
|
||||
"remove",
|
||||
"versions"
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "mesh-artifact-store",
|
||||
"scope": "mesh"
|
||||
}
|
||||
],
|
||||
"tools": [
|
||||
"registry_direct_writes",
|
||||
"registry_holds",
|
||||
"registry_incoming",
|
||||
"registry_old_name"
|
||||
],
|
||||
"invokes": [
|
||||
"mesh-delivery-builder.builder_versions",
|
||||
"mesh-delivery-view.node_haves",
|
||||
"seat:node-engine.manifest-reads"
|
||||
],
|
||||
"consumes": [
|
||||
"mesh-delivery-view.delivered"
|
||||
],
|
||||
"emits": [
|
||||
"old-name-closed",
|
||||
"direct-writes-closed"
|
||||
],
|
||||
"config": {
|
||||
"callers": {
|
||||
"type": "string",
|
||||
"trusted": true,
|
||||
"description": "per verb, the modules that may call it (JSON object of verb to modules; operator for the operator's session); empty is design 51's list"
|
||||
}
|
||||
},
|
||||
"requires": [
|
||||
{
|
||||
"name": "incoming",
|
||||
"storage": "object-store",
|
||||
"template": "objects",
|
||||
"options": {
|
||||
"max-mib": 8192
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "transfers",
|
||||
"storage": "key-value",
|
||||
"template": "state",
|
||||
"options": {
|
||||
"ttl-seconds": 86400
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "holds",
|
||||
"storage": "key-value",
|
||||
"template": "state"
|
||||
},
|
||||
{
|
||||
"name": "windows",
|
||||
"storage": "key-value",
|
||||
"template": "state"
|
||||
},
|
||||
{
|
||||
"name": "findings",
|
||||
"storage": "key-value",
|
||||
"template": "state"
|
||||
},
|
||||
{
|
||||
"name": "events",
|
||||
"storage": "stream",
|
||||
"template": "events",
|
||||
"options": {
|
||||
"max-age-days": 7,
|
||||
"max-mib": 16
|
||||
}
|
||||
}
|
||||
],
|
||||
"capabilities": [
|
||||
"container-runtime"
|
||||
],
|
||||
"own-secrets": {
|
||||
"broker": "/var/lib/mesh/registry/broker"
|
||||
},
|
||||
"secrets-owner": "distribution",
|
||||
"serves": {
|
||||
"artifact-store": {
|
||||
"port": 5000
|
||||
@@ -33,7 +127,7 @@
|
||||
"port": 5000,
|
||||
"protocol": "tcp",
|
||||
"from": "mesh",
|
||||
"why": "every machine pulls images and artifacts from here"
|
||||
"why": "every machine pulls images and artifacts from here; distribution's program serves it in front of the store"
|
||||
}
|
||||
],
|
||||
"data": {
|
||||
@@ -49,19 +143,28 @@
|
||||
]
|
||||
},
|
||||
"resources": [
|
||||
{
|
||||
"id": "account",
|
||||
"type": "user",
|
||||
"name": "distribution",
|
||||
"shell": "/usr/bin/nologin",
|
||||
"home": "/var/lib/mesh/registry"
|
||||
},
|
||||
{
|
||||
"id": "state",
|
||||
"type": "directory",
|
||||
"path": "/var/lib/mesh/registry",
|
||||
"mode": "0700",
|
||||
"unknown": "program"
|
||||
"unknown": "program",
|
||||
"owner": "distribution"
|
||||
},
|
||||
{
|
||||
"id": "registry-data",
|
||||
"type": "directory",
|
||||
"path": "/var/lib/mesh-registry",
|
||||
"mode": "0700",
|
||||
"unknown": "program"
|
||||
"unknown": "program",
|
||||
"owner": "distribution"
|
||||
},
|
||||
{
|
||||
"id": "store",
|
||||
@@ -69,7 +172,7 @@
|
||||
"name": "mesh-registry",
|
||||
"image": "registry@sha256:a3d8aaa63ed8681a604f1dea0aa03f100d5895b6a58ace528858a7b332415373",
|
||||
"ports": [
|
||||
"5000:5000"
|
||||
"127.0.0.1:5001:5000"
|
||||
],
|
||||
"volumes": [
|
||||
"/var/lib/mesh-registry:/var/lib/registry"
|
||||
@@ -78,6 +181,25 @@
|
||||
"REGISTRY_STORAGE_DELETE_ENABLED": "true"
|
||||
}
|
||||
},
|
||||
{
|
||||
"id": "program",
|
||||
"type": "process",
|
||||
"name": "distribution",
|
||||
"artifact": "program",
|
||||
"run": [
|
||||
"./distribution"
|
||||
],
|
||||
"env": {
|
||||
"MESH_BROKER_FILE": "/var/lib/mesh/registry/broker",
|
||||
"MESH_REGISTRY_LOOPBACK": "http://127.0.0.1:5001",
|
||||
"MESH_REGISTRY_LISTEN": ":5000"
|
||||
},
|
||||
"user": "distribution",
|
||||
"health": {
|
||||
"kind": "http",
|
||||
"url": "http://127.0.0.1:5000/v2/"
|
||||
}
|
||||
},
|
||||
{
|
||||
"id": "collect",
|
||||
"type": "container",
|
||||
@@ -95,5 +217,28 @@
|
||||
"store"
|
||||
]
|
||||
}
|
||||
]
|
||||
],
|
||||
"build": {
|
||||
"artifacts": [
|
||||
{
|
||||
"name": "program",
|
||||
"kind": "bundle",
|
||||
"language": "go",
|
||||
"system": "arch",
|
||||
"from": "cmd/distribution",
|
||||
"binary": "distribution"
|
||||
},
|
||||
{
|
||||
"name": "tools",
|
||||
"kind": "bundle",
|
||||
"language": "go",
|
||||
"system": "arch",
|
||||
"from": "cmd/distribution-tools",
|
||||
"binary": "distribution-tools",
|
||||
"loads": [
|
||||
"distribution-tools"
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user