mesh-catalog/modules/artifact-store-tools/cmd/artifact-store-tools/read.go

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
}