distribution: provide and serve no artifact-store, and answer the seats' list from the store's own files (ADR 0632 §1, §3, §4; issue 1966)
The store is found by the seats' holder and its listen registry; list says, per repository, each manifest's digest, whether it is an index and what it names, its tags, its size and whether it is held, so the sweep reads the store through the seat. E15 now asks that nothing is provided or served. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
This commit is contained in:
@@ -1,175 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
)
|
||||
|
||||
// readScript prints each manifest named on its command line, each after a line naming its path. A
|
||||
// manifest is JSON, which holds no raw line break inside a string, so no line of one starts with the
|
||||
// marker. One the store does not have is said, not skipped.
|
||||
const readScript = `cd ` + storageRoot + `
|
||||
for p in "$@"; do
|
||||
echo "#manifest $p"
|
||||
if [ -f "$p" ]; then cat "$p"; echo; else echo "#absent"; fi
|
||||
done
|
||||
echo '#end'`
|
||||
|
||||
// readBatch is how many manifests one exec reads: well under any command line's limit.
|
||||
const readBatch = 400
|
||||
|
||||
// Store reaches the artifact store's own files, through its own container, and only reads them.
|
||||
type Store struct {
|
||||
Container string
|
||||
Run Runner
|
||||
UID int
|
||||
|
||||
mu sync.Mutex
|
||||
// cache holds every manifest read: a manifest is named by its content's digest, so it never changes.
|
||||
cache map[string]*Manifest
|
||||
}
|
||||
|
||||
// List reads the store's files.
|
||||
func (s *Store) List(ctx context.Context) (*Listing, error) {
|
||||
if s.Container == "" {
|
||||
return nil, fmt.Errorf("this bundle was not told the artifact store's container (MESH_ARTIFACT_STORE_CONTAINER)")
|
||||
}
|
||||
out, err := docker(ctx, s.Run, s.UID, "exec", s.Container, "sh", "-c", listScript)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return ParseListing(out)
|
||||
}
|
||||
|
||||
// Key names one manifest in one repository.
|
||||
func Key(repo, digest string) string { return repo + "@" + digest }
|
||||
|
||||
// blobPath is where the store keeps a blob, under storageRoot.
|
||||
func blobPath(digest string) string {
|
||||
algorithm, hex, _ := strings.Cut(digest, ":")
|
||||
if algorithm == "" || len(hex) < 2 || strings.ContainsAny(digest, "/ \n") {
|
||||
return ""
|
||||
}
|
||||
return "blobs/" + algorithm + "/" + hex[:2] + "/" + hex + "/data"
|
||||
}
|
||||
|
||||
// Manifests reads every manifest the listing names, from the cache or the store's files. Answers each
|
||||
// one read, and each one not read with why.
|
||||
func (s *Store) Manifests(ctx context.Context, l *Listing) (map[string]*Manifest, map[string]string) {
|
||||
need := map[string]bool{}
|
||||
s.mu.Lock()
|
||||
if s.cache == nil {
|
||||
s.cache = map[string]*Manifest{}
|
||||
}
|
||||
for _, name := range l.RepoNames() {
|
||||
for d := range l.Repos[name].Revisions {
|
||||
if _, ok := s.cache[d]; !ok {
|
||||
need[d] = true
|
||||
}
|
||||
}
|
||||
}
|
||||
s.mu.Unlock()
|
||||
|
||||
digests := make([]string, 0, len(need))
|
||||
for d := range need {
|
||||
digests = append(digests, d)
|
||||
}
|
||||
sort.Strings(digests)
|
||||
failed := map[string]string{}
|
||||
for start := 0; start < len(digests); start += readBatch {
|
||||
batch := digests[start:min(start+readBatch, len(digests))]
|
||||
read, err := s.read(ctx, batch)
|
||||
s.mu.Lock()
|
||||
for _, d := range batch {
|
||||
switch m := read[d]; {
|
||||
case err != nil:
|
||||
failed[d] = err.Error()
|
||||
case m == nil:
|
||||
failed[d] = "the store holds no manifest content for it"
|
||||
default:
|
||||
s.cache[d] = m
|
||||
}
|
||||
}
|
||||
s.mu.Unlock()
|
||||
}
|
||||
|
||||
got := map[string]*Manifest{}
|
||||
unread := map[string]string{}
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
for _, name := range l.RepoNames() {
|
||||
for d := range l.Repos[name].Revisions {
|
||||
if m, ok := s.cache[d]; ok {
|
||||
got[Key(name, d)] = m
|
||||
} else if why, ok := failed[d]; ok {
|
||||
unread[Key(name, d)] = why
|
||||
} else {
|
||||
unread[Key(name, d)] = "not read"
|
||||
}
|
||||
}
|
||||
}
|
||||
return got, unread
|
||||
}
|
||||
|
||||
// read reads one batch of manifests. A blob that is not a manifest, or not there, is answered as nil;
|
||||
// an exec that fails fails the batch.
|
||||
func (s *Store) read(ctx context.Context, digests []string) (map[string]*Manifest, error) {
|
||||
args := []string{"exec", s.Container, "sh", "-c", readScript, "sh"}
|
||||
byPath := map[string]string{}
|
||||
for _, d := range digests {
|
||||
if p := blobPath(d); p != "" {
|
||||
byPath[p] = d
|
||||
args = append(args, p)
|
||||
}
|
||||
}
|
||||
out, err := docker(ctx, s.Run, s.UID, args...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return parseManifests(out, byPath)
|
||||
}
|
||||
|
||||
// parseManifests reads readScript's output.
|
||||
func parseManifests(out string, byPath map[string]string) (map[string]*Manifest, error) {
|
||||
got := map[string]*Manifest{}
|
||||
current := ""
|
||||
var body strings.Builder
|
||||
flush := func() {
|
||||
if d := byPath[current]; d != "" {
|
||||
m, err := ParseManifest([]byte(body.String()))
|
||||
if err != nil {
|
||||
m = nil
|
||||
}
|
||||
got[d] = m
|
||||
}
|
||||
current = ""
|
||||
body.Reset()
|
||||
}
|
||||
ended := false
|
||||
for _, line := range strings.Split(out, "\n") {
|
||||
switch {
|
||||
case strings.HasPrefix(line, "#manifest "):
|
||||
flush()
|
||||
current = strings.TrimPrefix(line, "#manifest ")
|
||||
case line == "#absent":
|
||||
if d := byPath[current]; d != "" {
|
||||
got[d] = nil
|
||||
}
|
||||
current = ""
|
||||
body.Reset()
|
||||
case line == "#end":
|
||||
flush()
|
||||
ended = true
|
||||
default:
|
||||
body.WriteString(line)
|
||||
body.WriteString("\n")
|
||||
}
|
||||
}
|
||||
if !ended {
|
||||
return nil, fmt.Errorf("reading the store's manifests was cut short (it never reached its end)")
|
||||
}
|
||||
return got, nil
|
||||
}
|
||||
@@ -1,38 +0,0 @@
|
||||
package main
|
||||
|
||||
import "testing"
|
||||
|
||||
func TestManifestsAreReadFromTheStoresFilesAndAMissingOneIsSaid(t *testing.T) {
|
||||
got, err := parseManifests("#manifest blobs/sha256/aa/aa1/data\n"+image(d("cfg"), d("l"))+"\n\n#manifest blobs/sha256/bb/bb1/data\n#absent\n#manifest blobs/sha256/cc/cc1/data\nnot json\n#end\n",
|
||||
map[string]string{"blobs/sha256/aa/aa1/data": "sha256:aa1", "blobs/sha256/bb/bb1/data": "sha256:bb1", "blobs/sha256/cc/cc1/data": "sha256:cc1"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if m := got["sha256:aa1"]; m == nil || len(m.Layers) != 1 || m.Layers[0] != d("l") {
|
||||
t.Errorf("aa1 %+v", m)
|
||||
}
|
||||
if got["sha256:bb1"] != nil || got["sha256:cc1"] != nil {
|
||||
t.Error("an absent or unreadable manifest was read as one")
|
||||
}
|
||||
if _, err := parseManifests("#manifest x\n{}\n", map[string]string{"x": "sha256:x"}); err == nil {
|
||||
t.Error("a cut read was accepted")
|
||||
}
|
||||
|
||||
f := world()
|
||||
delete(f.manifests, d("m-old"))
|
||||
v := view(t, f)
|
||||
if v.Unread[Key("app/server", d("m-old"))] == "" {
|
||||
t.Error("a manifest whose content is gone was not said as unread")
|
||||
}
|
||||
if out := Usage(v, 3); out["unread"] == nil {
|
||||
t.Error("the answer did not say a manifest was unread")
|
||||
}
|
||||
}
|
||||
|
||||
func TestADigestNeverBecomesAPathOutsideTheBlobs(t *testing.T) {
|
||||
for _, bad := range []string{"sha256:", "nocolon", "sha256:a/../../etc", "sha256:ab cd"} {
|
||||
if p := blobPath(bad); p != "" {
|
||||
t.Errorf("%q became %q", bad, p)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,96 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os/exec"
|
||||
"regexp"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Ran is what a command did: its output, its exit status, and why it never ran to an answer.
|
||||
type Ran struct {
|
||||
Stdout string
|
||||
Stderr string
|
||||
Status int
|
||||
// Err is "ENOENT" when the program is not installed, or says it was ended for taking too long.
|
||||
Err string
|
||||
}
|
||||
|
||||
// Runner runs one command, so every tool can be tested without a daemon.
|
||||
type Runner func(ctx context.Context, name string, args ...string) Ran
|
||||
|
||||
// ListTimeout is how long listing the store's files may take: below the runtime's thirty-second call
|
||||
// limit, so a store that hangs is answered as such rather than as a call the runtime gave up on.
|
||||
const ListTimeout = 15 * time.Second
|
||||
|
||||
// ExecRunner runs a command on this machine, bounded by ListTimeout.
|
||||
func ExecRunner(ctx context.Context, name string, args ...string) Ran {
|
||||
ctx, cancel := context.WithTimeout(ctx, ListTimeout)
|
||||
defer cancel()
|
||||
cmd := exec.CommandContext(ctx, name, args...)
|
||||
var out, errb bytes.Buffer
|
||||
cmd.Stdout, cmd.Stderr = &out, &errb
|
||||
err := cmd.Run()
|
||||
r := Ran{Stdout: out.String(), Stderr: errb.String()}
|
||||
var exitErr *exec.ExitError
|
||||
switch {
|
||||
case errors.Is(ctx.Err(), context.DeadlineExceeded):
|
||||
r.Status, r.Err = 124, fmt.Sprintf("no answer within %d s", int(ListTimeout/time.Second))
|
||||
case errors.Is(err, exec.ErrNotFound):
|
||||
r.Status, r.Err = 127, "ENOENT"
|
||||
case errors.As(err, &exitErr):
|
||||
r.Status = exitErr.ExitCode()
|
||||
case err != nil:
|
||||
r.Status, r.Err = 1, err.Error()
|
||||
}
|
||||
return r
|
||||
}
|
||||
|
||||
var socketRefused = regexp.MustCompile(`(?i)permission denied.*docker.*sock|docker\.sock.*permission denied`)
|
||||
|
||||
// docker runs one docker command as this account and answers its stdout. A socket that refuses the
|
||||
// account is asked again through `sudo -n`, as the container runtime's own tools do; never as root
|
||||
// otherwise, and never with a prompt.
|
||||
func docker(ctx context.Context, run Runner, uid int, args ...string) (string, error) {
|
||||
r := run(ctx, "docker", args...)
|
||||
program := "docker"
|
||||
if r.Status != 0 && r.Err == "" && uid != 0 && socketRefused.MatchString(r.Stderr) {
|
||||
program = "sudo"
|
||||
r = run(ctx, "sudo", append([]string{"-n", "docker"}, args...)...)
|
||||
}
|
||||
if r.Status == 0 && r.Err == "" {
|
||||
return r.Stdout, nil
|
||||
}
|
||||
said := strings.TrimSpace(r.Stderr + "\n" + r.Stdout)
|
||||
switch {
|
||||
case r.Err == "ENOENT" && program == "sudo":
|
||||
return "", errors.New("the runtime's socket refused this account, and sudo is not installed here to escalate with")
|
||||
case r.Err == "ENOENT":
|
||||
return "", errors.New("docker is not installed on this machine, so the store's files cannot be listed")
|
||||
case r.Err != "":
|
||||
return "", fmt.Errorf("listing the store's files did not answer: %s", r.Err)
|
||||
case program == "sudo" && strings.HasPrefix(said, "sudo:"):
|
||||
return "", fmt.Errorf("the runtime's socket refused this account and it may not escalate without a prompt: %s", firstLine(said))
|
||||
case strings.Contains(said, "No such container"):
|
||||
return "", fmt.Errorf("the store's container is not on this machine: %s", firstLine(said))
|
||||
case strings.Contains(said, "is not running"):
|
||||
return "", fmt.Errorf("the store's container is not running: %s", firstLine(said))
|
||||
}
|
||||
if l := firstLine(said); l != "" {
|
||||
return "", fmt.Errorf("listing the store's files failed (%d): %s", r.Status, l)
|
||||
}
|
||||
return "", fmt.Errorf("listing the store's files failed with status %d", r.Status)
|
||||
}
|
||||
|
||||
func firstLine(s string) string {
|
||||
for _, l := range strings.Split(s, "\n") {
|
||||
if l = strings.TrimSpace(l); l != "" {
|
||||
return l
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
@@ -1,180 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// What the artifact store holds, read from its own files, and nothing written to them.
|
||||
//
|
||||
// **Why its files.** The store's door lists repositories and the tags in each, but not a manifest no tag
|
||||
// names — and since the mesh pins every machine by digest, that is almost every manifest the store
|
||||
// holds. The files list them all, with every blob's size, and hold every manifest's content. They are
|
||||
// read through the store's own container, which is the only process that has them, and only read
|
||||
// (novox/hq ADR 0251 §1). So this module runs on the machine that holds the store; anywhere else its
|
||||
// tools say the container is not there.
|
||||
|
||||
// storageRoot is where the registry keeps its files, inside its container.
|
||||
const storageRoot = "/var/lib/registry/docker/registry/v2"
|
||||
|
||||
// listScript lists, in three sections, every blob with its size, every manifest each repository holds,
|
||||
// and what each tag names now. Busybox's find and stat (the store's image is Alpine): no -printf. An
|
||||
// upload in progress is not listed, and nor are a tag's former targets.
|
||||
const listScript = `set -e
|
||||
cd ` + storageRoot + `
|
||||
echo '#blobs'
|
||||
if [ -d blobs ]; then find blobs -type f -name data | xargs -r stat -c '%s %n'; fi
|
||||
echo '#revisions'
|
||||
if [ -d repositories ]; then find repositories -type f -path '*/_manifests/revisions/sha256/*/link'; fi
|
||||
echo '#tags'
|
||||
if [ -d repositories ]; then find repositories -type f -path '*/_manifests/tags/*/current/link' -exec grep -H . {} + || true; fi
|
||||
echo '#end'`
|
||||
|
||||
// Repo is one repository as the store's files list it.
|
||||
type Repo struct {
|
||||
Name string
|
||||
// Revisions are the digests of every manifest it holds, tagged or not.
|
||||
Revisions map[string]bool
|
||||
// Tags are what each tag names now.
|
||||
Tags map[string]string
|
||||
}
|
||||
|
||||
// Listing is the store's files: every blob with its size, and every repository.
|
||||
type Listing struct {
|
||||
Blobs map[string]int64
|
||||
Repos map[string]*Repo
|
||||
}
|
||||
|
||||
func (l *Listing) repo(name string) *Repo {
|
||||
r := l.Repos[name]
|
||||
if r == nil {
|
||||
r = &Repo{Name: name, Revisions: map[string]bool{}, Tags: map[string]string{}}
|
||||
l.Repos[name] = r
|
||||
}
|
||||
return r
|
||||
}
|
||||
|
||||
// RepoNames are the repositories in order.
|
||||
func (l *Listing) RepoNames() []string {
|
||||
out := make([]string, 0, len(l.Repos))
|
||||
for n := range l.Repos {
|
||||
out = append(out, n)
|
||||
}
|
||||
sort.Strings(out)
|
||||
return out
|
||||
}
|
||||
|
||||
// ParseListing reads listScript's output. A section that never ended means the listing was cut short,
|
||||
// which is an error: a partial listing would make held things look absent.
|
||||
func ParseListing(out string) (*Listing, error) {
|
||||
l := &Listing{Blobs: map[string]int64{}, Repos: map[string]*Repo{}}
|
||||
section := ""
|
||||
ended := false
|
||||
for _, line := range strings.Split(out, "\n") {
|
||||
line = strings.TrimSpace(line)
|
||||
if line == "" {
|
||||
continue
|
||||
}
|
||||
if strings.HasPrefix(line, "#") {
|
||||
section = line
|
||||
ended = line == "#end"
|
||||
continue
|
||||
}
|
||||
switch section {
|
||||
case "#blobs":
|
||||
size, path, ok := strings.Cut(line, " ")
|
||||
n, err := strconv.ParseInt(size, 10, 64)
|
||||
if !ok || err != nil {
|
||||
return nil, fmt.Errorf("the store's listing has a blob line it cannot read: %q", line)
|
||||
}
|
||||
// blobs/sha256/ab/<hex>/data
|
||||
parts := strings.Split(strings.TrimPrefix(path, "./"), "/")
|
||||
if len(parts) != 5 || parts[0] != "blobs" || parts[4] != "data" {
|
||||
return nil, fmt.Errorf("the store's listing has a blob at an unexpected place: %q", path)
|
||||
}
|
||||
l.Blobs[parts[1]+":"+parts[3]] = n
|
||||
case "#revisions":
|
||||
// repositories/<repo>/_manifests/revisions/sha256/<hex>/link
|
||||
name, rest, ok := strings.Cut(strings.TrimPrefix(strings.TrimPrefix(line, "./"), "repositories/"), "/_manifests/revisions/")
|
||||
parts := strings.Split(rest, "/")
|
||||
if !ok || len(parts) != 3 || parts[2] != "link" {
|
||||
return nil, fmt.Errorf("the store's listing has a manifest at an unexpected place: %q", line)
|
||||
}
|
||||
l.repo(name).Revisions[parts[0]+":"+parts[1]] = true
|
||||
case "#tags":
|
||||
// repositories/<repo>/_manifests/tags/<tag>/current/link:sha256:<hex>
|
||||
path, digest, ok := strings.Cut(line, ":sha256:")
|
||||
name, rest, ok2 := strings.Cut(strings.TrimPrefix(strings.TrimPrefix(path, "./"), "repositories/"), "/_manifests/tags/")
|
||||
tag, _, ok3 := strings.Cut(rest, "/current/link")
|
||||
if !ok || !ok2 || !ok3 || tag == "" {
|
||||
return nil, fmt.Errorf("the store's listing has a tag line it cannot read: %q", line)
|
||||
}
|
||||
l.repo(name).Tags[tag] = "sha256:" + digest
|
||||
default:
|
||||
return nil, fmt.Errorf("the store's listing has a line outside any section: %q", line)
|
||||
}
|
||||
}
|
||||
if !ended {
|
||||
return nil, fmt.Errorf("the store's listing was cut short (it never reached its end)")
|
||||
}
|
||||
return l, nil
|
||||
}
|
||||
|
||||
// Manifest is what one manifest names.
|
||||
type Manifest struct {
|
||||
MediaType string
|
||||
ConfigMedia string
|
||||
// Refs are the blobs it names: its configuration and its layers.
|
||||
Refs []string
|
||||
// Layers are its layers alone, in order.
|
||||
Layers []string
|
||||
// Children are the manifests an index names.
|
||||
Children []string
|
||||
}
|
||||
|
||||
// Holder is the shape of a manifest that keeps one blob: an empty configuration and one layer. The
|
||||
// controller holds every archive the mesh keeps this way (novox/hq issue 253), and keeps a named
|
||||
// document the same way, under a tag.
|
||||
func (m *Manifest) Holder() bool {
|
||||
return m != nil && m.ConfigMedia == "application/vnd.oci.empty.v1+json" && len(m.Layers) == 1
|
||||
}
|
||||
|
||||
// ParseManifest reads a manifest's content, of any of the shapes a registry holds.
|
||||
func ParseManifest(raw []byte) (*Manifest, error) {
|
||||
type desc struct {
|
||||
MediaType string `json:"mediaType"`
|
||||
Digest string `json:"digest"`
|
||||
}
|
||||
var v struct {
|
||||
MediaType string `json:"mediaType"`
|
||||
Config *desc `json:"config"`
|
||||
Layers []desc `json:"layers"`
|
||||
Manifests []desc `json:"manifests"`
|
||||
FSLayers []struct {
|
||||
BlobSum string `json:"blobSum"`
|
||||
} `json:"fsLayers"`
|
||||
}
|
||||
if err := json.Unmarshal(raw, &v); err != nil {
|
||||
return nil, fmt.Errorf("not a manifest: %v", err)
|
||||
}
|
||||
m := &Manifest{MediaType: v.MediaType}
|
||||
if v.Config != nil && v.Config.Digest != "" {
|
||||
m.ConfigMedia = v.Config.MediaType
|
||||
m.Refs = append(m.Refs, v.Config.Digest)
|
||||
}
|
||||
for _, d := range v.Layers {
|
||||
m.Refs = append(m.Refs, d.Digest)
|
||||
m.Layers = append(m.Layers, d.Digest)
|
||||
}
|
||||
for _, d := range v.FSLayers {
|
||||
m.Refs = append(m.Refs, d.BlobSum)
|
||||
m.Layers = append(m.Layers, d.BlobSum)
|
||||
}
|
||||
for _, d := range v.Manifests {
|
||||
m.Children = append(m.Children, d.Digest)
|
||||
}
|
||||
return m, nil
|
||||
}
|
||||
@@ -1,76 +0,0 @@
|
||||
package registry
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"time"
|
||||
)
|
||||
|
||||
// blobSize is how large the store says a blob is; Gone when it does not have it.
|
||||
func (s Store) blobSize(ctx context.Context, repository, digest string) (int64, error) {
|
||||
request, err := http.NewRequestWithContext(ctx, http.MethodHead, s.url(repository, "blobs", digest), nil)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
response, err := s.client().Do(request)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("cannot reach the artifact store at %s: %w", s.Address, err)
|
||||
}
|
||||
defer response.Body.Close()
|
||||
switch response.StatusCode {
|
||||
case http.StatusOK:
|
||||
case http.StatusNotFound:
|
||||
return 0, Gone
|
||||
default:
|
||||
return 0, fmt.Errorf("the artifact store answered %s for %s/blobs/%s", response.Status, repository, digest)
|
||||
}
|
||||
// Read from the header rather than ContentLength: a HEAD's ContentLength is what the response
|
||||
// says it would have sent, which Go reports faithfully, but a proxy in between is free to drop
|
||||
// it, and the header is what the registry itself wrote.
|
||||
if length := response.Header.Get("Content-Length"); length != "" {
|
||||
if n, err := strconv.ParseInt(length, 10, 64); err == nil && n >= 0 {
|
||||
return n, nil
|
||||
}
|
||||
}
|
||||
if response.ContentLength >= 0 {
|
||||
return response.ContentLength, nil
|
||||
}
|
||||
return 0, fmt.Errorf("the artifact store holds %s/blobs/%s and will not say how large it is", repository, digest)
|
||||
}
|
||||
|
||||
// has is whether the store answers 200 for a HEAD at that URL.
|
||||
func (s Store) has(ctx context.Context, url string, accept ...string) (bool, error) {
|
||||
request, err := http.NewRequestWithContext(ctx, http.MethodHead, url, nil)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
for _, media := range accept {
|
||||
request.Header.Add("Accept", media)
|
||||
}
|
||||
response, err := s.client().Do(request)
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("cannot reach the artifact store at %s: %w", s.Address, err)
|
||||
}
|
||||
defer response.Body.Close()
|
||||
switch response.StatusCode {
|
||||
case http.StatusOK:
|
||||
return true, nil
|
||||
case http.StatusNotFound:
|
||||
return false, nil
|
||||
default:
|
||||
return false, fmt.Errorf("the artifact store answered %s for %s", response.Status, url)
|
||||
}
|
||||
}
|
||||
|
||||
func (s Store) url(repository, kind, digest string) string {
|
||||
return "http://" + s.Address + "/v2/" + repository + "/" + kind + "/" + digest
|
||||
}
|
||||
|
||||
func (s Store) client() *http.Client {
|
||||
if s.HTTP != nil {
|
||||
return s.HTTP
|
||||
}
|
||||
return &http.Client{Timeout: 30 * time.Second}
|
||||
}
|
||||
@@ -1,140 +0,0 @@
|
||||
// Package registry reads the mesh's artifact store over its own door: whether it holds what a reference names, and
|
||||
// whether a manifest is an index. It writes nothing: what the artifact sweep holds and removes it asks the registry
|
||||
// seats the store's holder serves (CAT-AS1, novox/hq ADR 0617 §2).
|
||||
package registry
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"strings"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
)
|
||||
|
||||
// Store is the artifact store at an address, as this machine reaches it.
|
||||
type Store struct {
|
||||
// Address is `host:port` — the store as the caller reaches it now, composed and never
|
||||
// recorded (novox/hq 04-ISSUES/102).
|
||||
Address string
|
||||
// HTTP is the client used; nil is a client with a modest timeout.
|
||||
HTTP *http.Client
|
||||
}
|
||||
|
||||
// Gone is the answer when the store does not hold it: the outcome wanted, already true.
|
||||
var Gone = errors.New("the store does not hold it")
|
||||
|
||||
// ErrNotOurs is a reference this sweep will not address: not the mesh's own, or naming nothing
|
||||
// the store holds by digest.
|
||||
//
|
||||
// **A fact about the record, not about the store** (novox/hq issue 226). The two deserve opposite
|
||||
// responses — skip one and go on, abandon the sweep for the other — and collapsing them into "an
|
||||
// error" is how a cautious loop became one that did nothing while reporting the right number.
|
||||
var ErrNotOurs = errors.New("not a reference into the mesh's artifact store")
|
||||
|
||||
// anyManifest is every manifest the store may hold, indexes included, so it answers with the
|
||||
// document as it is.
|
||||
var anyManifest = []string{
|
||||
"application/vnd.oci.image.index.v1+json",
|
||||
"application/vnd.docker.distribution.manifest.list.v2+json",
|
||||
"application/vnd.oci.image.manifest.v1+json",
|
||||
"application/vnd.docker.distribution.manifest.v2+json",
|
||||
}
|
||||
|
||||
// Platforms is the manifests an index names, by digest; none for a manifest that is not an index.
|
||||
// Gone when the store does not hold it.
|
||||
func (s Store) Platforms(ctx context.Context, repository, digest string) ([]string, error) {
|
||||
request, err := http.NewRequestWithContext(ctx, http.MethodGet, s.url(repository, "manifests", digest), nil)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
request.Header.Set("Accept", strings.Join(anyManifest, ", "))
|
||||
response, err := s.client().Do(request)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer response.Body.Close()
|
||||
switch response.StatusCode {
|
||||
case http.StatusOK:
|
||||
case http.StatusNotFound:
|
||||
return nil, Gone
|
||||
default:
|
||||
return nil, fmt.Errorf("the artifact store answered %s when asked for %s@%s", response.Status, repository, digest)
|
||||
}
|
||||
var document struct {
|
||||
Manifests []struct {
|
||||
Digest string `json:"digest"`
|
||||
} `json:"manifests"`
|
||||
}
|
||||
if err := json.NewDecoder(io.LimitReader(response.Body, 4<<20)).Decode(&document); err != nil {
|
||||
return nil, fmt.Errorf("%s@%s is not a manifest: %w", repository, digest, err)
|
||||
}
|
||||
var out []string
|
||||
for _, m := range document.Manifests {
|
||||
if strings.HasPrefix(m.Digest, "sha256:") {
|
||||
out = append(out, m.Digest)
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// split reads a recorded path into the repository, which endpoint names the thing, and the digest.
|
||||
//
|
||||
// Two shapes, which are the two the mesh records: `<repository>@sha256:<hex>` is a manifest, and
|
||||
// `<repository>/blobs/sha256:<hex>` is a blob.
|
||||
func split(path string) (repository, kind, digest string, err error) {
|
||||
if before, after, ok := strings.Cut(path, "@sha256:"); ok {
|
||||
return before, "manifests", "sha256:" + after, nil
|
||||
}
|
||||
if before, after, ok := strings.Cut(path, "/blobs/sha256:"); ok {
|
||||
return before, "blobs", "sha256:" + after, nil
|
||||
}
|
||||
return "", "", "", fmt.Errorf("%w: %q names nothing the store holds by digest", ErrNotOurs, path)
|
||||
}
|
||||
|
||||
// IsIndex is whether a recorded reference is an index the store holds: one that names manifests. A
|
||||
// blob, or a manifest that names none, is not. Gone when the store does not hold it.
|
||||
func (s Store) IsIndex(ctx context.Context, reference string) (bool, error) {
|
||||
path, ours := catalogue.InArtifactStore(reference)
|
||||
if !ours {
|
||||
return false, fmt.Errorf("%w: %s", ErrNotOurs, reference)
|
||||
}
|
||||
repository, kind, digest, err := split(path)
|
||||
if err != nil || kind != "manifests" {
|
||||
return false, err
|
||||
}
|
||||
children, err := s.Platforms(ctx, repository, digest)
|
||||
return len(children) > 0, err
|
||||
}
|
||||
|
||||
// Holds is whether the store holds what a recorded reference names: the blob of an archive, the manifest of an image.
|
||||
// Gone when it does not; asks and changes nothing (ADR 0526 clause 3: a registration by hand is refused at bytes that
|
||||
// are not there).
|
||||
func (s Store) Holds(ctx context.Context, reference string) error {
|
||||
path, ours := catalogue.InArtifactStore(reference)
|
||||
if !ours {
|
||||
return fmt.Errorf("%w: %s", ErrNotOurs, reference)
|
||||
}
|
||||
repository, kind, digest, err := split(path)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if s.Address == "" {
|
||||
return fmt.Errorf("this mesh has no artifact store on its network to ask about %s", reference)
|
||||
}
|
||||
if kind == "blobs" {
|
||||
_, err := s.blobSize(ctx, repository, digest)
|
||||
return err
|
||||
}
|
||||
held, err := s.has(ctx, s.url(repository, kind, digest), anyManifest...)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !held {
|
||||
return Gone
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -1,76 +0,0 @@
|
||||
package registry
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
)
|
||||
|
||||
// The store's door is read only (CAT-AS1: what the sweep holds and removes it asks the registry seats): Holds says
|
||||
// whether the store holds what a reference names, in either reference form, IsIndex whether it is an index; a reference
|
||||
// that is not the mesh's own is never asked. A fake store answers by path and records what it was asked.
|
||||
func TestTheDoorReadsWhatTheStoreHoldsAndWritesNothing(t *testing.T) {
|
||||
const index = `{"mediaType":"application/vnd.oci.image.index.v1+json","manifests":[{"digest":"sha256:c1"}]}`
|
||||
const image = `{"mediaType":"application/vnd.oci.image.manifest.v1+json","layers":[]}`
|
||||
var asked []string
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
asked = append(asked, r.Method+" "+r.URL.Path)
|
||||
switch r.URL.Path {
|
||||
case "/v2/web/app/manifests/sha256:abc", "/v2/web/files/blobs/sha256:def":
|
||||
if r.Method == http.MethodGet {
|
||||
_, _ = w.Write([]byte(image))
|
||||
} else {
|
||||
w.Header().Set("Content-Length", "7")
|
||||
}
|
||||
case "/v2/web/app/manifests/sha256:idx":
|
||||
if r.Method == http.MethodGet {
|
||||
_, _ = w.Write([]byte(index))
|
||||
}
|
||||
default:
|
||||
w.WriteHeader(http.StatusNotFound)
|
||||
}
|
||||
}))
|
||||
defer srv.Close()
|
||||
store := Store{Address: strings.TrimPrefix(srv.URL, "http://")}
|
||||
ctx := context.Background()
|
||||
for name, c := range map[string]struct {
|
||||
reference string
|
||||
holds error // nil, Gone or ErrNotOurs
|
||||
index bool
|
||||
}{
|
||||
"an image": {reference: catalogue.ArtifactStoreScheme + "web/app@sha256:abc"},
|
||||
"an image in the seat's form": {reference: catalogue.SeatReference("container-image-registry", "web", "app", "sha256:abc")},
|
||||
"an archive": {reference: catalogue.ArtifactStoreScheme + "web/files/blobs/sha256:def"},
|
||||
"an archive in the seat's form": {reference: catalogue.SeatReference("oci-artifact-registry", "web", "files", "sha256:def")},
|
||||
"an index": {reference: catalogue.ArtifactStoreScheme + "web/app@sha256:idx", index: true},
|
||||
"an image the store lacks": {reference: catalogue.ArtifactStoreScheme + "web/app@sha256:999", holds: Gone},
|
||||
"an archive the store lacks": {reference: catalogue.ArtifactStoreScheme + "web/files/blobs/sha256:999", holds: Gone},
|
||||
"another registry's image": {reference: "docker.io/library/alpine@sha256:abc", holds: ErrNotOurs},
|
||||
"a reference naming no digest": {reference: catalogue.ArtifactStoreScheme + "web/app:latest", holds: ErrNotOurs},
|
||||
} {
|
||||
t.Run(name, func(t *testing.T) {
|
||||
asked = nil
|
||||
if err := store.Holds(ctx, c.reference); !errors.Is(err, c.holds) && (err != nil || c.holds != nil) {
|
||||
t.Errorf("Holds = %v, want %v", err, c.holds)
|
||||
}
|
||||
if c.holds == nil {
|
||||
if got, err := store.IsIndex(ctx, c.reference); err != nil || got != c.index {
|
||||
t.Errorf("IsIndex = %v, %v, want %v", got, err, c.index)
|
||||
}
|
||||
}
|
||||
for _, a := range asked {
|
||||
if !strings.HasPrefix(a, "HEAD ") && !strings.HasPrefix(a, "GET ") {
|
||||
t.Errorf("the door asked %s: it writes nothing", a)
|
||||
}
|
||||
}
|
||||
if errors.Is(c.holds, ErrNotOurs) && len(asked) != 0 {
|
||||
t.Errorf("a reference that is not the mesh's own was asked: %v", asked)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -1,6 +1,9 @@
|
||||
# distribution
|
||||
|
||||
The `distribution` module of the mesh catalogue: runs the container image `registry`; claims the seat `mesh-artifact-store`.
|
||||
The `distribution` module of the mesh catalogue: runs the container image `registry`; holds the seats `container-image-registry`
|
||||
and `oci-artifact-registry`, and `mesh-artifact-store` until its window closes. It provides no provision: the store's address
|
||||
is its listen `registry`, open to the mesh, which every node-engine finds through the seat's holder (novox/hq ADR 0632). Its
|
||||
seats' `list` answers what the store holds, read from the store's own files through its container `mesh-registry`.
|
||||
|
||||
## Files it does not know
|
||||
|
||||
|
||||
@@ -6,6 +6,7 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
@@ -328,6 +329,19 @@ func tools(v *holder.Verbs, windows kept[holder.Window], callers func() (holder.
|
||||
}
|
||||
return reg.RemoveAnswer{Removed: got.Removed, Kept: got.Kept, Reason: got.Reason, Tags: got.Tags, Artifact: ®.Artifact{Name: in.Name, Version: in.Version, Tag: in.Tag, Digest: in.Digest}}, nil
|
||||
}},
|
||||
stdio.Tool{Name: seat + "." + holder.VerbList, Description: "What the store holds of one repository, or of every repository the seat holds: each manifest's digest, whether it is an index and the manifests it names, the tags naming it, its size and whether it is held (ADR 0632 §3). Read from the store's own files; changes nothing.",
|
||||
Input: obj(map[string]any{"name": s("<module>/<artifact>; every repository when not given")}),
|
||||
Run: func(args map[string]any) (any, error) {
|
||||
v, err := now()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
repos, err := v.List(context.Background(), seat, stdio.CallerOf(args), str(args, "name"))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return reg.ListAnswer{Repositories: repos}, nil
|
||||
}},
|
||||
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"), "digest": s("sha256:…"), "reason": s("which of the sweep's rules keeps it")}, "name", "digest", "reason"),
|
||||
Run: func(args map[string]any) (any, error) {
|
||||
@@ -507,7 +521,9 @@ func main() {
|
||||
store := &holder.OCI{Base: base, Upstream: &holder.Upstream{}}
|
||||
h := &holder.Holder{Objects: parts.BusObjects(holder.IncomingObjects), Builder: delivery.BuilderVersions{Ask: stdio.Ask}, Store: store, Say: say, Finding: finding,
|
||||
Transfers: parts.BusTransfers[holder.Opening](holder.TransfersState), Cap: parts.BusCap(holder.IncomingObjects), Record: recordAct, Idle: parts.BusArm(holder.IdleState)}
|
||||
v := &holder.Verbs{Holder: h, Catalog: store, Holds: kept[holder.Hold]{state: "holds", key: holder.Hold.Key}, Mirrors: mirrors{}}
|
||||
files := &holder.Files{Container: or(os.Getenv("MESH_REGISTRY_CONTAINER"), "mesh-registry"), Run: holder.ExecRunner, UID: os.Getuid()}
|
||||
v := &holder.Verbs{Holder: h, Catalog: store, Holds: kept[holder.Hold]{state: "holds", key: holder.Hold.Key}, Mirrors: mirrors{},
|
||||
Files: files}
|
||||
go followMirrors(v)
|
||||
go parts.FollowIdle(holder.IdleState, parts.IDs(h.Transfers), h.Expired, func(id string, dropped bool, err error) {
|
||||
if err != nil {
|
||||
@@ -522,3 +538,10 @@ func main() {
|
||||
os.Exit(1)
|
||||
}
|
||||
}
|
||||
|
||||
func or(v, d string) string {
|
||||
if strings.TrimSpace(v) == "" {
|
||||
return d
|
||||
}
|
||||
return v
|
||||
}
|
||||
|
||||
@@ -36,9 +36,9 @@ const (
|
||||
// Seats are the seats distribution answers verbs on.
|
||||
var Seats = []string{ImageSeat, ArtifactSeat}
|
||||
|
||||
// VerbsOf are a seat's verbs: mirror and tags on the image seat alone; hold and release on both of distribution's.
|
||||
// VerbsOf are a seat's verbs: mirror and tags on the image seat alone; hold, release and list on both of distribution's.
|
||||
func VerbsOf(seat string) []string {
|
||||
v := []string{VerbPublish, VerbPublishPart, VerbVersions, VerbRemove, VerbHold, VerbRelease}
|
||||
v := []string{VerbPublish, VerbPublishPart, VerbVersions, VerbRemove, VerbHold, VerbRelease, VerbList}
|
||||
if seat == ImageSeat {
|
||||
v = append(v, VerbMirror, VerbTags)
|
||||
}
|
||||
@@ -54,7 +54,7 @@ type Callers = stdio.VerbCallers
|
||||
// DefaultCallers is the list design 51 *Who may call which verb* names: publish and publish-part the build modules;
|
||||
// mirror mesh-build-image; remove the builder (a tagged version) and the artifact sweep's holder (retention); hold the
|
||||
// artifact sweep's holder alone; release that holder, and the operator's session only through registry_holds (Operator
|
||||
// below). versions is read by anyone on the mesh, as the store's reads are.
|
||||
// below). versions, tags and list are read by anyone on the mesh, as the store's reads are.
|
||||
func DefaultCallers() Callers {
|
||||
builds := []string{"mesh-build-go", "mesh-build-image", "mesh-build-node"}
|
||||
return Callers{
|
||||
@@ -66,6 +66,7 @@ func DefaultCallers() Callers {
|
||||
VerbRelease: {SweepHolder, Operator},
|
||||
VerbVersions: {Anyone},
|
||||
VerbTags: {Anyone},
|
||||
VerbList: {Anyone},
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,435 @@
|
||||
package holder
|
||||
|
||||
// **The store's own files, read through its own container** (novox/hq ADR 0632 §3): what `list` answers. The store's
|
||||
// read side lists repositories and the tags in each, but not a manifest no tag names, and since the mesh pins every
|
||||
// machine by digest that is almost every manifest the store holds. The files list them all, with every blob's size, and
|
||||
// hold every manifest's content. They are read through the store's container, distribution's own (the resource store),
|
||||
// and only read; the artifact sweep and the read tools ask `list` and never reach the files themselves. Moved here from
|
||||
// artifact-store-tools, which read them by docker exec into this module's container until ADR 0632.
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os/exec"
|
||||
"regexp"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// storageRoot is where the registry keeps its files, inside its container.
|
||||
const storageRoot = "/var/lib/registry/docker/registry/v2"
|
||||
|
||||
// listScript lists, in three sections, every blob with its size, every manifest each repository holds,
|
||||
// and what each tag names now. Busybox's find and stat (the store's image is Alpine): no -printf. An
|
||||
// upload in progress is not listed, and nor are a tag's former targets.
|
||||
const listScript = `set -e
|
||||
cd ` + storageRoot + `
|
||||
echo '#blobs'
|
||||
if [ -d blobs ]; then find blobs -type f -name data | xargs -r stat -c '%s %n'; fi
|
||||
echo '#revisions'
|
||||
if [ -d repositories ]; then find repositories -type f -path '*/_manifests/revisions/sha256/*/link'; fi
|
||||
echo '#tags'
|
||||
if [ -d repositories ]; then find repositories -type f -path '*/_manifests/tags/*/current/link' -exec grep -H . {} + || true; fi
|
||||
echo '#end'`
|
||||
|
||||
// Repo is one repository as the store's files list it.
|
||||
type Repo struct {
|
||||
Name string
|
||||
// Revisions are the digests of every manifest it holds, tagged or not.
|
||||
Revisions map[string]bool
|
||||
// Tags are what each tag names now.
|
||||
Tags map[string]string
|
||||
}
|
||||
|
||||
// Listing is the store's files: every blob with its size, and every repository.
|
||||
type Listing struct {
|
||||
Blobs map[string]int64
|
||||
Repos map[string]*Repo
|
||||
}
|
||||
|
||||
func (l *Listing) repo(name string) *Repo {
|
||||
r := l.Repos[name]
|
||||
if r == nil {
|
||||
r = &Repo{Name: name, Revisions: map[string]bool{}, Tags: map[string]string{}}
|
||||
l.Repos[name] = r
|
||||
}
|
||||
return r
|
||||
}
|
||||
|
||||
// RepoNames are the repositories in order.
|
||||
func (l *Listing) RepoNames() []string {
|
||||
out := make([]string, 0, len(l.Repos))
|
||||
for n := range l.Repos {
|
||||
out = append(out, n)
|
||||
}
|
||||
sort.Strings(out)
|
||||
return out
|
||||
}
|
||||
|
||||
// ParseListing reads listScript's output. A section that never ended means the listing was cut short,
|
||||
// which is an error: a partial listing would make held things look absent.
|
||||
func ParseListing(out string) (*Listing, error) {
|
||||
l := &Listing{Blobs: map[string]int64{}, Repos: map[string]*Repo{}}
|
||||
section := ""
|
||||
ended := false
|
||||
for _, line := range strings.Split(out, "\n") {
|
||||
line = strings.TrimSpace(line)
|
||||
if line == "" {
|
||||
continue
|
||||
}
|
||||
if strings.HasPrefix(line, "#") {
|
||||
section = line
|
||||
ended = line == "#end"
|
||||
continue
|
||||
}
|
||||
switch section {
|
||||
case "#blobs":
|
||||
size, path, ok := strings.Cut(line, " ")
|
||||
n, err := strconv.ParseInt(size, 10, 64)
|
||||
if !ok || err != nil {
|
||||
return nil, fmt.Errorf("the store's listing has a blob line it cannot read: %q", line)
|
||||
}
|
||||
// blobs/sha256/ab/<hex>/data
|
||||
parts := strings.Split(strings.TrimPrefix(path, "./"), "/")
|
||||
if len(parts) != 5 || parts[0] != "blobs" || parts[4] != "data" {
|
||||
return nil, fmt.Errorf("the store's listing has a blob at an unexpected place: %q", path)
|
||||
}
|
||||
l.Blobs[parts[1]+":"+parts[3]] = n
|
||||
case "#revisions":
|
||||
// repositories/<repo>/_manifests/revisions/sha256/<hex>/link
|
||||
name, rest, ok := strings.Cut(strings.TrimPrefix(strings.TrimPrefix(line, "./"), "repositories/"), "/_manifests/revisions/")
|
||||
parts := strings.Split(rest, "/")
|
||||
if !ok || len(parts) != 3 || parts[2] != "link" {
|
||||
return nil, fmt.Errorf("the store's listing has a manifest at an unexpected place: %q", line)
|
||||
}
|
||||
l.repo(name).Revisions[parts[0]+":"+parts[1]] = true
|
||||
case "#tags":
|
||||
// repositories/<repo>/_manifests/tags/<tag>/current/link:sha256:<hex>
|
||||
path, digest, ok := strings.Cut(line, ":sha256:")
|
||||
name, rest, ok2 := strings.Cut(strings.TrimPrefix(strings.TrimPrefix(path, "./"), "repositories/"), "/_manifests/tags/")
|
||||
tag, _, ok3 := strings.Cut(rest, "/current/link")
|
||||
if !ok || !ok2 || !ok3 || tag == "" {
|
||||
return nil, fmt.Errorf("the store's listing has a tag line it cannot read: %q", line)
|
||||
}
|
||||
l.repo(name).Tags[tag] = "sha256:" + digest
|
||||
default:
|
||||
return nil, fmt.Errorf("the store's listing has a line outside any section: %q", line)
|
||||
}
|
||||
}
|
||||
if !ended {
|
||||
return nil, fmt.Errorf("the store's listing was cut short (it never reached its end)")
|
||||
}
|
||||
return l, nil
|
||||
}
|
||||
|
||||
// Manifest is what one manifest names.
|
||||
type Manifest struct {
|
||||
MediaType string
|
||||
ConfigMedia string
|
||||
// Refs are the blobs it names: its configuration and its layers.
|
||||
Refs []string
|
||||
// Layers are its layers alone, in order.
|
||||
Layers []string
|
||||
// Children are the manifests an index names.
|
||||
Children []string
|
||||
}
|
||||
|
||||
// Holder is the shape of a manifest that keeps one blob: an empty configuration and one layer. The
|
||||
// controller holds every archive the mesh keeps this way (novox/hq issue 253), and keeps a named
|
||||
// document the same way, under a tag.
|
||||
func (m *Manifest) Holder() bool {
|
||||
return m != nil && m.ConfigMedia == "application/vnd.oci.empty.v1+json" && len(m.Layers) == 1
|
||||
}
|
||||
|
||||
// ParseManifest reads a manifest's content, of any of the shapes a registry holds.
|
||||
func ParseManifest(raw []byte) (*Manifest, error) {
|
||||
type desc struct {
|
||||
MediaType string `json:"mediaType"`
|
||||
Digest string `json:"digest"`
|
||||
}
|
||||
var v struct {
|
||||
MediaType string `json:"mediaType"`
|
||||
Config *desc `json:"config"`
|
||||
Layers []desc `json:"layers"`
|
||||
Manifests []desc `json:"manifests"`
|
||||
FSLayers []struct {
|
||||
BlobSum string `json:"blobSum"`
|
||||
} `json:"fsLayers"`
|
||||
}
|
||||
if err := json.Unmarshal(raw, &v); err != nil {
|
||||
return nil, fmt.Errorf("not a manifest: %v", err)
|
||||
}
|
||||
m := &Manifest{MediaType: v.MediaType}
|
||||
if v.Config != nil && v.Config.Digest != "" {
|
||||
m.ConfigMedia = v.Config.MediaType
|
||||
m.Refs = append(m.Refs, v.Config.Digest)
|
||||
}
|
||||
for _, d := range v.Layers {
|
||||
m.Refs = append(m.Refs, d.Digest)
|
||||
m.Layers = append(m.Layers, d.Digest)
|
||||
}
|
||||
for _, d := range v.FSLayers {
|
||||
m.Refs = append(m.Refs, d.BlobSum)
|
||||
m.Layers = append(m.Layers, d.BlobSum)
|
||||
}
|
||||
for _, d := range v.Manifests {
|
||||
m.Children = append(m.Children, d.Digest)
|
||||
}
|
||||
return m, nil
|
||||
}
|
||||
|
||||
// readScript prints each manifest named on its command line, each after a line naming its path. A
|
||||
// manifest is JSON, which holds no raw line break inside a string, so no line of one starts with the
|
||||
// marker. One the store does not have is said, not skipped.
|
||||
const readScript = `cd ` + storageRoot + `
|
||||
for p in "$@"; do
|
||||
echo "#manifest $p"
|
||||
if [ -f "$p" ]; then cat "$p"; echo; else echo "#absent"; fi
|
||||
done
|
||||
echo '#end'`
|
||||
|
||||
// readBatch is how many manifests one exec reads: well under any command line's limit.
|
||||
const readBatch = 400
|
||||
|
||||
// Files reaches the artifact store's own files, through its own container, and only reads them.
|
||||
type Files struct {
|
||||
Container string
|
||||
Run Runner
|
||||
UID int
|
||||
|
||||
mu sync.Mutex
|
||||
// cache holds every manifest read: a manifest is named by its content's digest, so it never changes.
|
||||
cache map[string]*Manifest
|
||||
}
|
||||
|
||||
// List reads the store's files.
|
||||
func (s *Files) List(ctx context.Context) (*Listing, error) {
|
||||
if s.Container == "" {
|
||||
return nil, fmt.Errorf("this bundle was not told the artifact store's container (MESH_REGISTRY_CONTAINER)")
|
||||
}
|
||||
out, err := docker(ctx, s.Run, s.UID, "exec", s.Container, "sh", "-c", listScript)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return ParseListing(out)
|
||||
}
|
||||
|
||||
// Key names one manifest in one repository.
|
||||
func Key(repo, digest string) string { return repo + "@" + digest }
|
||||
|
||||
// blobPath is where the store keeps a blob, under storageRoot.
|
||||
func blobPath(digest string) string {
|
||||
algorithm, hex, _ := strings.Cut(digest, ":")
|
||||
if algorithm == "" || len(hex) < 2 || strings.ContainsAny(digest, "/ \n") {
|
||||
return ""
|
||||
}
|
||||
return "blobs/" + algorithm + "/" + hex[:2] + "/" + hex + "/data"
|
||||
}
|
||||
|
||||
// Manifests reads every manifest the listing names, from the cache or the store's files. Answers each
|
||||
// one read, and each one not read with why.
|
||||
func (s *Files) Manifests(ctx context.Context, l *Listing) (map[string]*Manifest, map[string]string) {
|
||||
need := map[string]bool{}
|
||||
s.mu.Lock()
|
||||
if s.cache == nil {
|
||||
s.cache = map[string]*Manifest{}
|
||||
}
|
||||
for _, name := range l.RepoNames() {
|
||||
for d := range l.Repos[name].Revisions {
|
||||
if _, ok := s.cache[d]; !ok {
|
||||
need[d] = true
|
||||
}
|
||||
}
|
||||
}
|
||||
s.mu.Unlock()
|
||||
|
||||
digests := make([]string, 0, len(need))
|
||||
for d := range need {
|
||||
digests = append(digests, d)
|
||||
}
|
||||
sort.Strings(digests)
|
||||
failed := map[string]string{}
|
||||
for start := 0; start < len(digests); start += readBatch {
|
||||
batch := digests[start:min(start+readBatch, len(digests))]
|
||||
read, err := s.read(ctx, batch)
|
||||
s.mu.Lock()
|
||||
for _, d := range batch {
|
||||
switch m := read[d]; {
|
||||
case err != nil:
|
||||
failed[d] = err.Error()
|
||||
case m == nil:
|
||||
failed[d] = "the store holds no manifest content for it"
|
||||
default:
|
||||
s.cache[d] = m
|
||||
}
|
||||
}
|
||||
s.mu.Unlock()
|
||||
}
|
||||
|
||||
got := map[string]*Manifest{}
|
||||
unread := map[string]string{}
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
for _, name := range l.RepoNames() {
|
||||
for d := range l.Repos[name].Revisions {
|
||||
if m, ok := s.cache[d]; ok {
|
||||
got[Key(name, d)] = m
|
||||
} else if why, ok := failed[d]; ok {
|
||||
unread[Key(name, d)] = why
|
||||
} else {
|
||||
unread[Key(name, d)] = "not read"
|
||||
}
|
||||
}
|
||||
}
|
||||
return got, unread
|
||||
}
|
||||
|
||||
// read reads one batch of manifests. A blob that is not a manifest, or not there, is answered as nil;
|
||||
// an exec that fails fails the batch.
|
||||
func (s *Files) read(ctx context.Context, digests []string) (map[string]*Manifest, error) {
|
||||
args := []string{"exec", s.Container, "sh", "-c", readScript, "sh"}
|
||||
byPath := map[string]string{}
|
||||
for _, d := range digests {
|
||||
if p := blobPath(d); p != "" {
|
||||
byPath[p] = d
|
||||
args = append(args, p)
|
||||
}
|
||||
}
|
||||
out, err := docker(ctx, s.Run, s.UID, args...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return parseManifests(out, byPath)
|
||||
}
|
||||
|
||||
// parseManifests reads readScript's output.
|
||||
func parseManifests(out string, byPath map[string]string) (map[string]*Manifest, error) {
|
||||
got := map[string]*Manifest{}
|
||||
current := ""
|
||||
var body strings.Builder
|
||||
flush := func() {
|
||||
if d := byPath[current]; d != "" {
|
||||
m, err := ParseManifest([]byte(body.String()))
|
||||
if err != nil {
|
||||
m = nil
|
||||
}
|
||||
got[d] = m
|
||||
}
|
||||
current = ""
|
||||
body.Reset()
|
||||
}
|
||||
ended := false
|
||||
for _, line := range strings.Split(out, "\n") {
|
||||
switch {
|
||||
case strings.HasPrefix(line, "#manifest "):
|
||||
flush()
|
||||
current = strings.TrimPrefix(line, "#manifest ")
|
||||
case line == "#absent":
|
||||
if d := byPath[current]; d != "" {
|
||||
got[d] = nil
|
||||
}
|
||||
current = ""
|
||||
body.Reset()
|
||||
case line == "#end":
|
||||
flush()
|
||||
ended = true
|
||||
default:
|
||||
body.WriteString(line)
|
||||
body.WriteString("\n")
|
||||
}
|
||||
}
|
||||
if !ended {
|
||||
return nil, fmt.Errorf("reading the store's manifests was cut short (it never reached its end)")
|
||||
}
|
||||
return got, nil
|
||||
}
|
||||
|
||||
// Ran is what a command did: its output, its exit status, and why it never ran to an answer.
|
||||
type Ran struct {
|
||||
Stdout string
|
||||
Stderr string
|
||||
Status int
|
||||
// Err is "ENOENT" when the program is not installed, or says it was ended for taking too long.
|
||||
Err string
|
||||
}
|
||||
|
||||
// Runner runs one command, so every tool can be tested without a daemon.
|
||||
type Runner func(ctx context.Context, name string, args ...string) Ran
|
||||
|
||||
// ListTimeout is how long listing the store's files may take: below the runtime's thirty-second call
|
||||
// limit, so a store that hangs is answered as such rather than as a call the runtime gave up on.
|
||||
const ListTimeout = 15 * time.Second
|
||||
|
||||
// ExecRunner runs a command on this machine, bounded by ListTimeout.
|
||||
func ExecRunner(ctx context.Context, name string, args ...string) Ran {
|
||||
ctx, cancel := context.WithTimeout(ctx, ListTimeout)
|
||||
defer cancel()
|
||||
cmd := exec.CommandContext(ctx, name, args...)
|
||||
var out, errb bytes.Buffer
|
||||
cmd.Stdout, cmd.Stderr = &out, &errb
|
||||
err := cmd.Run()
|
||||
r := Ran{Stdout: out.String(), Stderr: errb.String()}
|
||||
var exitErr *exec.ExitError
|
||||
switch {
|
||||
case errors.Is(ctx.Err(), context.DeadlineExceeded):
|
||||
r.Status, r.Err = 124, fmt.Sprintf("no answer within %d s", int(ListTimeout/time.Second))
|
||||
case errors.Is(err, exec.ErrNotFound):
|
||||
r.Status, r.Err = 127, "ENOENT"
|
||||
case errors.As(err, &exitErr):
|
||||
r.Status = exitErr.ExitCode()
|
||||
case err != nil:
|
||||
r.Status, r.Err = 1, err.Error()
|
||||
}
|
||||
return r
|
||||
}
|
||||
|
||||
var socketRefused = regexp.MustCompile(`(?i)permission denied.*docker.*sock|docker\.sock.*permission denied`)
|
||||
|
||||
// docker runs one docker command as this account and answers its stdout. A socket that refuses the
|
||||
// account is asked again through `sudo -n`, as the container runtime's own tools do; never as root
|
||||
// otherwise, and never with a prompt.
|
||||
func docker(ctx context.Context, run Runner, uid int, args ...string) (string, error) {
|
||||
r := run(ctx, "docker", args...)
|
||||
program := "docker"
|
||||
if r.Status != 0 && r.Err == "" && uid != 0 && socketRefused.MatchString(r.Stderr) {
|
||||
program = "sudo"
|
||||
r = run(ctx, "sudo", append([]string{"-n", "docker"}, args...)...)
|
||||
}
|
||||
if r.Status == 0 && r.Err == "" {
|
||||
return r.Stdout, nil
|
||||
}
|
||||
said := strings.TrimSpace(r.Stderr + "\n" + r.Stdout)
|
||||
switch {
|
||||
case r.Err == "ENOENT" && program == "sudo":
|
||||
return "", errors.New("the runtime's socket refused this account, and sudo is not installed here to escalate with")
|
||||
case r.Err == "ENOENT":
|
||||
return "", errors.New("docker is not installed on this machine, so the store's files cannot be listed")
|
||||
case r.Err != "":
|
||||
return "", fmt.Errorf("listing the store's files did not answer: %s", r.Err)
|
||||
case program == "sudo" && strings.HasPrefix(said, "sudo:"):
|
||||
return "", fmt.Errorf("the runtime's socket refused this account and it may not escalate without a prompt: %s", firstLine(said))
|
||||
case strings.Contains(said, "No such container"):
|
||||
return "", fmt.Errorf("the store's container is not on this machine: %s", firstLine(said))
|
||||
case strings.Contains(said, "is not running"):
|
||||
return "", fmt.Errorf("the store's container is not running: %s", firstLine(said))
|
||||
}
|
||||
if l := firstLine(said); l != "" {
|
||||
return "", fmt.Errorf("listing the store's files failed (%d): %s", r.Status, l)
|
||||
}
|
||||
return "", fmt.Errorf("listing the store's files failed with status %d", r.Status)
|
||||
}
|
||||
|
||||
func firstLine(s string) string {
|
||||
for _, l := range strings.Split(s, "\n") {
|
||||
if l = strings.TrimSpace(l); l != "" {
|
||||
return l
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
@@ -827,6 +827,7 @@ func TestTheHolderOfTheTwoImageSeats(t *testing.T) {
|
||||
}
|
||||
var m struct {
|
||||
Provides []struct{ Name string } `json:"provides"`
|
||||
Serves map[string]any `json:"serves"`
|
||||
Claims []struct {
|
||||
Name string `json:"name"`
|
||||
Serves []string `json:"serves"`
|
||||
@@ -837,9 +838,10 @@ func TestTheHolderOfTheTwoImageSeats(t *testing.T) {
|
||||
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)
|
||||
// E15: the provision artifact-store retired (ADR 0632 §3): distribution provides and serves nothing, and the
|
||||
// store is found by its seats' holder and the listen it opens to the mesh.
|
||||
if len(m.Provides) != 0 || len(m.Serves) != 0 {
|
||||
t.Errorf("provides = %+v, serves = %v, want neither: the provision artifact-store retired", m.Provides, m.Serves)
|
||||
}
|
||||
claims := map[string][]string{}
|
||||
for _, c := range m.Claims {
|
||||
|
||||
@@ -0,0 +1,109 @@
|
||||
package holder
|
||||
|
||||
// **`list`: what the store holds, through the seat** (novox/hq ADR 0632 §3). For one repository, or every repository a
|
||||
// seat holds, each manifest's digest, whether it is an index and the manifests it names, the tags naming it, its size
|
||||
// and whether a hold keeps it. Read from the store's own files (files.go), so a manifest no tag names is listed too. The
|
||||
// two seats share one store and split it by what a manifest keeps: a manifest keeping one blob under an empty
|
||||
// configuration is a bundle or an archive, oci-artifact-registry's, named by that blob as its references name it;
|
||||
// every other manifest, an index included, is container-image-registry's.
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sort"
|
||||
|
||||
reg "git.novox.be/novox/mesh-contracts/go/seats/containerimageregistry"
|
||||
)
|
||||
|
||||
// VerbList is the seats' verb list.
|
||||
const VerbList = reg.VerbList
|
||||
|
||||
// StoreFiles is the store's own files as list reads them: Files in the program, a test's stand-in.
|
||||
type StoreFiles interface {
|
||||
List(ctx context.Context) (*Listing, error)
|
||||
Manifests(ctx context.Context, l *Listing) (map[string]*Manifest, map[string]string)
|
||||
}
|
||||
|
||||
// List is `list` on seat: the repositories it holds, or name alone; a repository the seat holds nothing of is not named,
|
||||
// and name not held is an empty answer. A manifest whose content cannot be read fails the answer, by name: a listing
|
||||
// missing one would read as the store not holding it.
|
||||
func (v *Verbs) List(ctx context.Context, seat, caller, name string) ([]reg.Repository, error) {
|
||||
if err := v.Callers.Allowed(VerbList, caller); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if seat != ImageSeat && seat != ArtifactSeat {
|
||||
return nil, fmt.Errorf("list: %s is no seat of distribution's", seat)
|
||||
}
|
||||
if v.Files == nil {
|
||||
return nil, errors.New("list: this process reads no store files")
|
||||
}
|
||||
l, err := v.Files.List(ctx)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("list: the store's files cannot be listed: %w", err)
|
||||
}
|
||||
if name != "" {
|
||||
r, held := l.Repos[name]
|
||||
l = &Listing{Blobs: l.Blobs, Repos: map[string]*Repo{}}
|
||||
if held {
|
||||
l.Repos[name] = r
|
||||
}
|
||||
}
|
||||
docs, unread := v.Files.Manifests(ctx, l)
|
||||
if len(unread) > 0 {
|
||||
keys := make([]string, 0, len(unread))
|
||||
for k := range unread {
|
||||
keys = append(keys, k)
|
||||
}
|
||||
sort.Strings(keys)
|
||||
return nil, fmt.Errorf("list: %d manifests cannot be read, the first %s: %s", len(keys), keys[0], unread[keys[0]])
|
||||
}
|
||||
holds, err := v.Holds.List()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("list: the holds cannot be read: %w", err)
|
||||
}
|
||||
held := map[string]bool{}
|
||||
for _, h := range holds {
|
||||
held[h.Key()] = true
|
||||
}
|
||||
out := []reg.Repository{}
|
||||
for _, repo := range l.RepoNames() {
|
||||
r := l.Repos[repo]
|
||||
tagsAt := map[string][]string{}
|
||||
for t, d := range r.Tags {
|
||||
tagsAt[d] = append(tagsAt[d], t)
|
||||
}
|
||||
digests := make([]string, 0, len(r.Revisions))
|
||||
for d := range r.Revisions {
|
||||
digests = append(digests, d)
|
||||
}
|
||||
sort.Strings(digests)
|
||||
var manifests []reg.Listed
|
||||
for _, d := range digests {
|
||||
m := docs[Key(repo, d)]
|
||||
artifact := ""
|
||||
if m.Holder() {
|
||||
artifact = m.Layers[0]
|
||||
}
|
||||
if (artifact != "") != (seat == ArtifactSeat) {
|
||||
continue
|
||||
}
|
||||
e := reg.Listed{Digest: d, Index: len(m.Children) > 0, Platforms: m.Children, Artifact: artifact,
|
||||
Tags: tagsAt[d], Size: l.Blobs[d]}
|
||||
for _, b := range m.Refs {
|
||||
e.Size += l.Blobs[b]
|
||||
}
|
||||
sort.Strings(e.Tags)
|
||||
by := d
|
||||
if artifact != "" {
|
||||
by = artifact // a bundle or an archive is held by the blob its references name (HoldOne)
|
||||
}
|
||||
e.Held = held[Hold{Seat: seat, Name: repo, Digest: by}.Key()]
|
||||
manifests = append(manifests, e)
|
||||
}
|
||||
if len(manifests) > 0 {
|
||||
out = append(out, reg.Repository{Name: repo, Manifests: manifests})
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
@@ -0,0 +1,226 @@
|
||||
package holder
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"reflect"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
reg "git.novox.be/novox/mesh-contracts/go/seats/containerimageregistry"
|
||||
)
|
||||
|
||||
// dg is a digest made from a short name, so a test reads as names.
|
||||
func dg(name string) string { return fmt.Sprintf("sha256:%064x", []byte(name))[:71] }
|
||||
|
||||
// fakeFiles is a store's files and its manifests: what listScript and readScript print through its container.
|
||||
type fakeFiles struct {
|
||||
blobs map[string]int64
|
||||
revisions map[string][]string // repo -> digests
|
||||
tags map[string]map[string]string
|
||||
manifests map[string]string // digest -> content
|
||||
}
|
||||
|
||||
func (f *fakeFiles) listing() string {
|
||||
var b strings.Builder
|
||||
b.WriteString("#blobs\n")
|
||||
for d, n := range f.blobs {
|
||||
h := strings.TrimPrefix(d, "sha256:")
|
||||
fmt.Fprintf(&b, "%d blobs/sha256/%s/%s/data\n", n, h[:2], h)
|
||||
}
|
||||
b.WriteString("#revisions\n")
|
||||
for repo, ds := range f.revisions {
|
||||
for _, d := range ds {
|
||||
fmt.Fprintf(&b, "repositories/%s/_manifests/revisions/sha256/%s/link\n", repo, strings.TrimPrefix(d, "sha256:"))
|
||||
}
|
||||
}
|
||||
b.WriteString("#tags\n")
|
||||
for repo, ts := range f.tags {
|
||||
for t, d := range ts {
|
||||
fmt.Fprintf(&b, "repositories/%s/_manifests/tags/%s/current/link:%s\n", repo, t, d)
|
||||
}
|
||||
}
|
||||
b.WriteString("#end\n")
|
||||
return b.String()
|
||||
}
|
||||
|
||||
// run is the store's container as docker exec reaches it; anything else fails the test: list only reads.
|
||||
func (f *fakeFiles) run(t *testing.T) Runner {
|
||||
return func(_ context.Context, name string, args ...string) Ran {
|
||||
if name != "docker" || len(args) < 5 || args[0] != "exec" || args[1] != "mesh-registry" || args[3] != "-c" {
|
||||
t.Fatalf("ran %s %v", name, args)
|
||||
}
|
||||
switch args[4] {
|
||||
case listScript:
|
||||
return Ran{Stdout: f.listing()}
|
||||
case readScript:
|
||||
var b strings.Builder
|
||||
for _, p := range args[6:] {
|
||||
fmt.Fprintf(&b, "#manifest %s\n", p)
|
||||
found := false
|
||||
for d, body := range f.manifests {
|
||||
if blobPath(d) == p {
|
||||
b.WriteString(body + "\n")
|
||||
found = true
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
b.WriteString("#absent\n")
|
||||
}
|
||||
}
|
||||
b.WriteString("#end\n")
|
||||
return Ran{Stdout: b.String()}
|
||||
}
|
||||
t.Fatalf("ran a script list does not have: %q", args[4])
|
||||
return Ran{}
|
||||
}
|
||||
}
|
||||
|
||||
func imageDoc(config string, layers ...string) string {
|
||||
var ls []string
|
||||
for _, l := range layers {
|
||||
ls = append(ls, fmt.Sprintf(`{"mediaType":"application/vnd.oci.image.layer.v1.tar+gzip","digest":%q,"size":1}`, l))
|
||||
}
|
||||
return fmt.Sprintf(`{"schemaVersion":2,"mediaType":"application/vnd.oci.image.manifest.v1+json","config":{"mediaType":"application/vnd.oci.image.config.v1+json","digest":%q,"size":1},"layers":[%s]}`,
|
||||
config, strings.Join(ls, ","))
|
||||
}
|
||||
|
||||
func holderDoc(blob string) string {
|
||||
return fmt.Sprintf(`{"schemaVersion":2,"mediaType":"application/vnd.oci.image.manifest.v1+json","config":{"mediaType":"application/vnd.oci.empty.v1+json","digest":%q,"size":2},"layers":[{"mediaType":"application/vnd.oci.image.layer.v1.tar+gzip","digest":%q,"size":1}]}`,
|
||||
dg("empty"), blob)
|
||||
}
|
||||
|
||||
func indexDoc(children ...string) string {
|
||||
var cs []string
|
||||
for _, c := range children {
|
||||
cs = append(cs, fmt.Sprintf(`{"mediaType":"application/vnd.oci.image.manifest.v1+json","digest":%q,"size":1}`, c))
|
||||
}
|
||||
return fmt.Sprintf(`{"schemaVersion":2,"mediaType":"application/vnd.oci.image.index.v1+json","manifests":[%s]}`, strings.Join(cs, ","))
|
||||
}
|
||||
|
||||
// filesWorld is one store: app/server an image, an untagged one and an index naming the first; app/tools a bundle
|
||||
// keeping a-kept; mesh/facts a document under the tag latest.
|
||||
func filesWorld() *fakeFiles {
|
||||
f := &fakeFiles{blobs: map[string]int64{}, revisions: map[string][]string{}, tags: map[string]map[string]string{},
|
||||
manifests: map[string]string{}}
|
||||
put := func(repo, name, content string) {
|
||||
f.manifests[dg(name)] = content
|
||||
f.blobs[dg(name)] = int64(len(content))
|
||||
f.revisions[repo] = append(f.revisions[repo], dg(name))
|
||||
}
|
||||
for name, size := range map[string]int64{"cfg": 10, "layer": 1000, "a-kept": 2000, "empty": 2, "doc-body": 50} {
|
||||
f.blobs[dg(name)] = size
|
||||
}
|
||||
put("app/server", "m-kept", imageDoc(dg("cfg"), dg("layer")))
|
||||
put("app/server", "m-old", imageDoc(dg("cfg")))
|
||||
put("app/server", "idx", indexDoc(dg("m-kept")))
|
||||
put("app/tools", "h-kept", holderDoc(dg("a-kept")))
|
||||
put("mesh/facts", "doc", holderDoc(dg("doc-body")))
|
||||
f.tags["app/server"] = map[string]string{"3": dg("idx")}
|
||||
f.tags["mesh/facts"] = map[string]string{"latest": dg("doc")}
|
||||
return f
|
||||
}
|
||||
|
||||
// list (ADR 0632 §3): each seat answers its part of the one store from the store's own files, untagged manifests
|
||||
// included; an index with the manifests it names; a bundle or archive by the blob its references name, held as a hold
|
||||
// keeps it; and the files' listing read whole or refused.
|
||||
func TestListAnswersTheStoreThroughTheSeat(t *testing.T) {
|
||||
f := filesWorld()
|
||||
b := newBed(true)
|
||||
b.v.Files = &Files{Container: "mesh-registry", Run: f.run(t)}
|
||||
if err := b.v.HoldOne(ArtifactSeat, SweepHolder, HoldRequest{Name: "app/tools", Digest: dg("a-kept"), Why: "kept"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ctx := context.Background()
|
||||
images, err := b.v.List(ctx, ImageSeat, "mesh-anyone", "")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
size := func(name string, blobs ...string) int64 {
|
||||
n := f.blobs[dg(name)]
|
||||
for _, b := range blobs {
|
||||
n += f.blobs[dg(b)]
|
||||
}
|
||||
return n
|
||||
}
|
||||
want := []reg.Repository{{Name: "app/server", Manifests: []reg.Listed{}}}
|
||||
for _, e := range []reg.Listed{
|
||||
{Digest: dg("idx"), Index: true, Platforms: []string{dg("m-kept")}, Tags: []string{"3"}, Size: size("idx")},
|
||||
{Digest: dg("m-kept"), Size: size("m-kept", "cfg", "layer")},
|
||||
{Digest: dg("m-old"), Size: size("m-old", "cfg")},
|
||||
} {
|
||||
want[0].Manifests = append(want[0].Manifests, e)
|
||||
}
|
||||
sortListed(want[0].Manifests)
|
||||
if !reflect.DeepEqual(images, want) {
|
||||
t.Errorf("the image seat lists %+v\nwant %+v", images, want)
|
||||
}
|
||||
artifacts, err := b.v.List(ctx, ArtifactSeat, "mesh-anyone", "")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
wantA := []reg.Repository{
|
||||
{Name: "app/tools", Manifests: []reg.Listed{{Digest: dg("h-kept"), Artifact: dg("a-kept"), Size: size("h-kept", "empty", "a-kept"), Held: true}}},
|
||||
{Name: "mesh/facts", Manifests: []reg.Listed{{Digest: dg("doc"), Artifact: dg("doc-body"), Tags: []string{"latest"}, Size: size("doc", "empty", "doc-body")}}},
|
||||
}
|
||||
if !reflect.DeepEqual(artifacts, wantA) {
|
||||
t.Errorf("the artifact seat lists %+v\nwant %+v", artifacts, wantA)
|
||||
}
|
||||
if one, err := b.v.List(ctx, ImageSeat, "mesh-anyone", "app/tools"); err != nil || len(one) != 0 {
|
||||
t.Errorf("the image seat's list of a bundle's repository: %+v, %v", one, err)
|
||||
}
|
||||
if one, err := b.v.List(ctx, ArtifactSeat, "mesh-anyone", "mesh/facts"); err != nil || len(one) != 1 || one[0].Name != "mesh/facts" {
|
||||
t.Errorf("one repository: %+v, %v", one, err)
|
||||
}
|
||||
// A manifest whose content cannot be read fails the answer: a listing without it would read as the store not holding it.
|
||||
delete(f.manifests, dg("m-old"))
|
||||
b.v.Files = &Files{Container: "mesh-registry", Run: f.run(t)} // what was read is kept: a fresh reader reads again
|
||||
if _, err := b.v.List(ctx, ImageSeat, "mesh-anyone", ""); err == nil || !strings.Contains(err.Error(), "cannot be read") {
|
||||
t.Errorf("an unreadable manifest: %v", err)
|
||||
}
|
||||
// The files' listing: a repository with slashes read, a cut or stray one refused.
|
||||
l, err := ParseListing("#blobs\n12 blobs/sha256/ab/abcd/data\n#revisions\nrepositories/a/b/c/_manifests/revisions/sha256/ff/link\n#tags\nrepositories/a/b/c/_manifests/tags/v1/current/link:sha256:ff\n#end\n")
|
||||
if err != nil || l.Blobs["sha256:abcd"] != 12 || !l.Repos["a/b/c"].Revisions["sha256:ff"] || l.Repos["a/b/c"].Tags["v1"] != "sha256:ff" {
|
||||
t.Errorf("the listing misread: %+v, %v", l, err)
|
||||
}
|
||||
for _, out := range []string{"#blobs\n12 blobs/sha256/ab/abcd/data\n", "#blobs\nnot a line\n#end\n", "stray\n#end\n"} {
|
||||
if _, err := ParseListing(out); err == nil {
|
||||
t.Errorf("accepted %q", out)
|
||||
}
|
||||
}
|
||||
got, err := parseManifests("#manifest blobs/sha256/aa/aa1/data\n"+imageDoc(dg("cfg"), dg("l"))+"\n\n#manifest blobs/sha256/bb/bb1/data\n#absent\n#manifest blobs/sha256/cc/cc1/data\nnot json\n#end\n",
|
||||
map[string]string{"blobs/sha256/aa/aa1/data": "sha256:aa1", "blobs/sha256/bb/bb1/data": "sha256:bb1", "blobs/sha256/cc/cc1/data": "sha256:cc1"})
|
||||
if err != nil || got["sha256:aa1"] == nil || len(got["sha256:aa1"].Refs) != 2 || got["sha256:bb1"] != nil || got["sha256:cc1"] != nil {
|
||||
t.Errorf("manifests read %v, %v", got, err)
|
||||
}
|
||||
if _, err := parseManifests("#manifest x\n{}\n", map[string]string{"x": "sha256:x"}); err == nil {
|
||||
t.Error("a cut read was accepted")
|
||||
}
|
||||
// The runtime's socket refusing the account is asked again through sudo -n; a container not there is said so.
|
||||
var ran []string
|
||||
sudo := func(_ context.Context, name string, args ...string) Ran {
|
||||
ran = append(ran, name)
|
||||
if name == "docker" {
|
||||
return Ran{Status: 1, Stderr: "permission denied while trying to connect to the Docker daemon socket at unix:///var/run/docker.sock"}
|
||||
}
|
||||
return Ran{Stdout: "#end\n"}
|
||||
}
|
||||
if _, err := docker(ctx, sudo, 1000, "exec", "x"); err != nil || !reflect.DeepEqual(ran, []string{"docker", "sudo"}) {
|
||||
t.Errorf("ran %v: %v", ran, err)
|
||||
}
|
||||
if _, err := docker(ctx, func(context.Context, string, ...string) Ran {
|
||||
return Ran{Status: 1, Stderr: "Error: No such container: mesh-registry"}
|
||||
}, 1000, "exec"); err == nil || !strings.Contains(err.Error(), "not on this machine") {
|
||||
t.Errorf("err %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func sortListed(l []reg.Listed) {
|
||||
for i := range l {
|
||||
for j := i + 1; j < len(l); j++ {
|
||||
if l[j].Digest < l[i].Digest {
|
||||
l[i], l[j] = l[j], l[i]
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -83,6 +83,8 @@ type Verbs struct {
|
||||
Catalog Catalog
|
||||
Holds HoldState
|
||||
Mirrors Mirrors
|
||||
// Files are the store's own files, which list reads (ADR 0632 §3).
|
||||
Files StoreFiles
|
||||
}
|
||||
|
||||
// VersionsAnswer is `versions`: the versions the seat holds of name.
|
||||
|
||||
@@ -1,21 +1,13 @@
|
||||
{
|
||||
"module": "distribution",
|
||||
"version": "2",
|
||||
"provides": [
|
||||
{
|
||||
"name": "artifact-store",
|
||||
"scope": "mesh",
|
||||
"credential": {
|
||||
"form": "none"
|
||||
}
|
||||
}
|
||||
],
|
||||
"claims": [
|
||||
{
|
||||
"name": "container-image-registry",
|
||||
"scope": "mesh",
|
||||
"serves": [
|
||||
"hold",
|
||||
"list",
|
||||
"mirror",
|
||||
"publish",
|
||||
"publish-part",
|
||||
@@ -30,6 +22,7 @@
|
||||
"scope": "mesh",
|
||||
"serves": [
|
||||
"hold",
|
||||
"list",
|
||||
"publish",
|
||||
"publish-part",
|
||||
"release",
|
||||
@@ -134,11 +127,6 @@
|
||||
"broker": "/var/lib/mesh/registry/broker"
|
||||
},
|
||||
"secrets-owner": "distribution",
|
||||
"serves": {
|
||||
"artifact-store": {
|
||||
"port": 5000
|
||||
}
|
||||
},
|
||||
"listens": [
|
||||
{
|
||||
"name": "registry",
|
||||
@@ -269,7 +257,10 @@
|
||||
"binary": "distribution-tools",
|
||||
"loads": [
|
||||
"distribution-tools"
|
||||
]
|
||||
],
|
||||
"env": {
|
||||
"MESH_REGISTRY_CONTAINER": "mesh-registry"
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user