feat(offsite): 异地副本加入 registry 用户镜像，按摘要加密去重，索引按版本保留 14 天，新增 fetch-images 回推恢复
This commit is contained in:
13 files changed
+1626
-51
No files matched your search
@@ -0,0 +1,815 @@
|
||||
package offsite
|
||||
|
||||
// The off-site copy of the platform registry's user images. The installer
|
||||
// pushes everything under felis/ and mirror/ again on any host, so those are
|
||||
// left out; every other repository holds builds that exist nowhere else: a
|
||||
// lost registry volume would otherwise leave each server pinned to one of them
|
||||
// in ImagePullBackOff until someone rebuilds it.
|
||||
//
|
||||
// The bucket holds each blob and manifest once, by digest, encrypted like
|
||||
// everything else, plus an index that says which repository holds which
|
||||
// manifests under which tags. A new index version is written whenever that
|
||||
// changes, and every version stays until ImageHistory after a newer one
|
||||
// replaced it, together with every object it names. So a registry that comes
|
||||
// back empty, a host being rebuilt whose timer runs before anyone restores
|
||||
// anything, only adds a new version: the one before it is still there to
|
||||
// restore from (FetchImages, `felis offsite fetch-images -at`).
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"hash"
|
||||
"io"
|
||||
"maps"
|
||||
"regexp"
|
||||
"slices"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"felis.lolicon.best/internal/registrygate"
|
||||
)
|
||||
|
||||
const (
|
||||
registryDir = "registry/"
|
||||
imageBlobsDir = registryDir + "blobs/"
|
||||
imageManifestsDir = registryDir + "manifests/"
|
||||
imageIndexDir = registryDir + "index/"
|
||||
imageIndexExt = ".json" + objExt
|
||||
imageStampLayout = "20060102T150405Z"
|
||||
|
||||
maxImageIndexBytes = 64 << 20
|
||||
maxImageManifestBytes = 4 << 20
|
||||
)
|
||||
|
||||
// ImageHistory is how long a registry state stays restorable after a newer one
|
||||
// replaced it.
|
||||
const ImageHistory = 14 * 24 * time.Hour
|
||||
|
||||
// ErrImageGone is a manifest or blob the registry lists but no longer holds.
|
||||
var ErrImageGone = errors.New("offsite: not in the registry")
|
||||
|
||||
// ImageSource is the registry the sync reads, anonymously.
|
||||
type ImageSource interface {
|
||||
Repositories(ctx context.Context) ([]string, error)
|
||||
// Revisions lists every manifest repo holds, tagged or not, and its tags.
|
||||
Revisions(ctx context.Context, repo string) (digests []string, tags map[string]string, err error)
|
||||
// Manifest returns the manifest stored under digest and its media type.
|
||||
// A manifest the registry no longer holds is ErrImageGone.
|
||||
Manifest(ctx context.Context, repo, digest string) ([]byte, string, error)
|
||||
// Blob opens a blob; one the registry no longer holds is ErrImageGone.
|
||||
Blob(ctx context.Context, repo, digest string) (io.ReadCloser, error)
|
||||
}
|
||||
|
||||
// ImageTarget is the registry FetchImages pushes into.
|
||||
type ImageTarget interface {
|
||||
// PutBlob uploads the blob open returns unless repo already holds it.
|
||||
PutBlob(ctx context.Context, repo, digest string, size int64, open func() (io.ReadCloser, error)) error
|
||||
// PutManifest stores body under reference, a tag or its digest.
|
||||
PutManifest(ctx context.Context, repo, reference, mediaType string, body []byte) error
|
||||
}
|
||||
|
||||
// ImageIndex is one state of the registry's user images.
|
||||
type ImageIndex struct {
|
||||
Created time.Time `json:"created"`
|
||||
Repositories map[string]ImageRepo `json:"repositories"`
|
||||
Manifests map[string]ImageManifest `json:"manifests"`
|
||||
}
|
||||
|
||||
// ImageRepo is one repository: its manifests, and the tag → digest map.
|
||||
type ImageRepo struct {
|
||||
Manifests []string `json:"manifests"`
|
||||
Tags map[string]string `json:"tags,omitempty"`
|
||||
}
|
||||
|
||||
// ImageManifest describes a stored manifest: the blobs an image manifest names
|
||||
// (config and layers), or the manifests an index names.
|
||||
type ImageManifest struct {
|
||||
MediaType string `json:"media_type"`
|
||||
Size int64 `json:"size"`
|
||||
Blobs []ImageBlob `json:"blobs,omitempty"`
|
||||
Children []string `json:"children,omitempty"`
|
||||
}
|
||||
|
||||
// ImageBlob is one blob a manifest names.
|
||||
type ImageBlob struct {
|
||||
Digest string `json:"digest"`
|
||||
Size int64 `json:"size"`
|
||||
}
|
||||
|
||||
// Images counts the manifests the repositories hold, a manifest in two
|
||||
// repositories counting twice, like two images.
|
||||
func (x *ImageIndex) Images() int {
|
||||
n := 0
|
||||
for _, r := range x.Repositories {
|
||||
n += len(r.Manifests)
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
func newImageIndex(at time.Time) *ImageIndex {
|
||||
return &ImageIndex{Created: at, Repositories: map[string]ImageRepo{}, Manifests: map[string]ImageManifest{}}
|
||||
}
|
||||
|
||||
var imageDigestRE = regexp.MustCompile(`^sha256:[0-9a-f]{64}$`)
|
||||
|
||||
func imageBlobKey(digest string) string {
|
||||
return imageBlobsDir + strings.TrimPrefix(digest, "sha256:") + objExt
|
||||
}
|
||||
|
||||
func imageManifestKey(digest string) string {
|
||||
return imageManifestsDir + strings.TrimPrefix(digest, "sha256:") + objExt
|
||||
}
|
||||
|
||||
func imageIndexKey(stamp string) string { return imageIndexDir + stamp + imageIndexExt }
|
||||
|
||||
func reservedRepo(repo string) bool {
|
||||
root, _, _ := strings.Cut(repo, "/")
|
||||
return slices.Contains(registrygate.ReservedRepoRoots, root)
|
||||
}
|
||||
|
||||
// Manifest media types the copy understands.
|
||||
const (
|
||||
mediaOCIIndex = "application/vnd.oci.image.index.v1+json"
|
||||
mediaDockerList = "application/vnd.docker.distribution.manifest.list.v2+json"
|
||||
mediaOCIManifest = "application/vnd.oci.image.manifest.v1+json"
|
||||
mediaDockerManifest2 = "application/vnd.docker.distribution.manifest.v2+json"
|
||||
)
|
||||
|
||||
// describeManifest reads the blobs or child manifests out of a manifest.
|
||||
func describeManifest(body []byte, mediaType string) (ImageManifest, error) {
|
||||
type desc struct {
|
||||
Digest string `json:"digest"`
|
||||
Size int64 `json:"size"`
|
||||
}
|
||||
var m struct {
|
||||
MediaType string `json:"mediaType"`
|
||||
Config *desc `json:"config"`
|
||||
Layers []desc `json:"layers"`
|
||||
Manifests []desc `json:"manifests"`
|
||||
}
|
||||
if err := json.Unmarshal(body, &m); err != nil {
|
||||
return ImageManifest{}, fmt.Errorf("manifest: %w", err)
|
||||
}
|
||||
if mediaType == "" {
|
||||
mediaType = m.MediaType
|
||||
}
|
||||
im := ImageManifest{MediaType: mediaType, Size: int64(len(body))}
|
||||
check := func(d desc) error {
|
||||
if !imageDigestRE.MatchString(d.Digest) || d.Size < 0 {
|
||||
return fmt.Errorf("manifest names %q (size %d), not a sha256 digest", d.Digest, d.Size)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
switch mediaType {
|
||||
case mediaOCIIndex, mediaDockerList:
|
||||
for _, d := range m.Manifests {
|
||||
if err := check(d); err != nil {
|
||||
return ImageManifest{}, err
|
||||
}
|
||||
if !slices.Contains(im.Children, d.Digest) {
|
||||
im.Children = append(im.Children, d.Digest)
|
||||
}
|
||||
}
|
||||
case mediaOCIManifest, mediaDockerManifest2:
|
||||
if m.Config == nil {
|
||||
return ImageManifest{}, errors.New("image manifest without a config")
|
||||
}
|
||||
for _, d := range append([]desc{*m.Config}, m.Layers...) {
|
||||
if err := check(d); err != nil {
|
||||
return ImageManifest{}, err
|
||||
}
|
||||
if !slices.ContainsFunc(im.Blobs, func(b ImageBlob) bool { return b.Digest == d.Digest }) {
|
||||
im.Blobs = append(im.Blobs, ImageBlob{Digest: d.Digest, Size: d.Size})
|
||||
}
|
||||
}
|
||||
default:
|
||||
return ImageManifest{}, fmt.Errorf("unsupported manifest type %q", mediaType)
|
||||
}
|
||||
return im, nil
|
||||
}
|
||||
|
||||
// imageCopy is one sync pass over the registry.
|
||||
type imageCopy struct {
|
||||
s *Syncer
|
||||
remote map[string]int64
|
||||
prev *ImageIndex
|
||||
next *ImageIndex
|
||||
res *Result
|
||||
fail func(string, ...any)
|
||||
// failed is set by anything that leaves the new index short of the
|
||||
// registry, which rules out pruning in this pass.
|
||||
failed bool
|
||||
}
|
||||
|
||||
// errSkipped marks a manifest left out without failing the pass.
|
||||
var errSkipped = errors.New("skipped")
|
||||
|
||||
func (s *Syncer) syncImages(ctx context.Context, res *Result, fail func(string, ...any)) {
|
||||
if s.Images == nil {
|
||||
return
|
||||
}
|
||||
remote, err := s.listSizes(ctx, registryDir)
|
||||
if err != nil {
|
||||
fail("list %s in the bucket: %v", registryDir, err)
|
||||
return
|
||||
}
|
||||
versions := imageIndexStamps(remote)
|
||||
prev := newImageIndex(time.Time{})
|
||||
if n := len(versions); n > 0 {
|
||||
if prev, err = LoadImageIndex(ctx, s.Bucket, s.Key, versions[n-1]); err != nil {
|
||||
fail("read the registry index %s from the bucket: %v", versions[n-1], err)
|
||||
return
|
||||
}
|
||||
}
|
||||
c := &imageCopy{s: s, remote: remote, prev: prev, next: newImageIndex(s.now().UTC()), res: res, fail: fail}
|
||||
repos, err := s.Images.Repositories(ctx)
|
||||
if err != nil {
|
||||
fail("list the registry's repositories: %v", err)
|
||||
return
|
||||
}
|
||||
sort.Strings(repos)
|
||||
listed := map[string]bool{}
|
||||
for _, repo := range repos {
|
||||
if reservedRepo(repo) {
|
||||
continue
|
||||
}
|
||||
listed[repo] = true
|
||||
if ctx.Err() != nil {
|
||||
c.fail("stopped before %s: %v", repo, ctx.Err())
|
||||
c.failed = true
|
||||
c.carry(repo)
|
||||
continue
|
||||
}
|
||||
c.repo(ctx, repo)
|
||||
}
|
||||
|
||||
stamp := ""
|
||||
if len(versions) > 0 {
|
||||
stamp = versions[len(versions)-1]
|
||||
}
|
||||
if len(versions) == 0 || !sameImages(prev, c.next) {
|
||||
stamp = c.next.Created.Format(imageStampLayout)
|
||||
if err := s.putImageIndex(ctx, stamp, c.next); err != nil {
|
||||
fail("write the registry index: %v", err)
|
||||
return
|
||||
}
|
||||
if len(versions) == 0 || versions[len(versions)-1] != stamp {
|
||||
versions = append(versions, stamp)
|
||||
}
|
||||
s.logf("recorded registry index %s: %d repositories, %d images", stamp, len(c.next.Repositories), c.next.Images())
|
||||
}
|
||||
res.ImageIndex = stamp
|
||||
res.ImageRepos = len(c.next.Repositories)
|
||||
res.Images = c.next.Images()
|
||||
if !c.failed {
|
||||
s.pruneImages(ctx, versions, c.next, remote, res, fail)
|
||||
}
|
||||
for key, size := range remote {
|
||||
if strings.HasPrefix(key, imageBlobsDir) || strings.HasPrefix(key, imageManifestsDir) {
|
||||
res.RemoteImageBytes += size
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// repo copies one repository into c.next. A repository that cannot be read in
|
||||
// full keeps the entry the previous index had for it.
|
||||
func (c *imageCopy) repo(ctx context.Context, repo string) {
|
||||
digests, tags, err := c.s.Images.Revisions(ctx, repo)
|
||||
if err != nil {
|
||||
c.fail("list the images of %s: %v", repo, err)
|
||||
c.failed = true
|
||||
c.carry(repo)
|
||||
return
|
||||
}
|
||||
entry := ImageRepo{Tags: map[string]string{}}
|
||||
short := false
|
||||
for _, d := range digests {
|
||||
switch err := c.manifest(ctx, repo, d); {
|
||||
case errors.Is(err, errSkipped):
|
||||
case err != nil:
|
||||
c.fail("copy %s@%s: %v", repo, d, err)
|
||||
short = true
|
||||
default:
|
||||
entry.Manifests = append(entry.Manifests, d)
|
||||
}
|
||||
}
|
||||
if short {
|
||||
c.failed = true
|
||||
if _, ok := c.prev.Repositories[repo]; ok {
|
||||
c.carry(repo)
|
||||
return
|
||||
}
|
||||
}
|
||||
sort.Strings(entry.Manifests)
|
||||
for tag, d := range tags {
|
||||
if slices.Contains(entry.Manifests, d) {
|
||||
entry.Tags[tag] = d
|
||||
}
|
||||
}
|
||||
if len(entry.Manifests) > 0 {
|
||||
c.next.Repositories[repo] = entry
|
||||
}
|
||||
}
|
||||
|
||||
// carry keeps the previous index's entry for repo, and what it names.
|
||||
func (c *imageCopy) carry(repo string) {
|
||||
entry, ok := c.prev.Repositories[repo]
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
c.next.Repositories[repo] = entry
|
||||
var add func(d string)
|
||||
add = func(d string) {
|
||||
m, ok := c.prev.Manifests[d]
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
c.next.Manifests[d] = m
|
||||
for _, child := range m.Children {
|
||||
add(child)
|
||||
}
|
||||
}
|
||||
for _, d := range entry.Manifests {
|
||||
add(d)
|
||||
}
|
||||
}
|
||||
|
||||
// manifest makes sure digest, everything it names, and then the manifest
|
||||
// itself are in the bucket, and records it in c.next. A manifest or blob the
|
||||
// registry lost is errSkipped, unless an earlier pass already copied the
|
||||
// manifest whole: that copy is then the only complete one and stays.
|
||||
func (c *imageCopy) manifest(ctx context.Context, repo, digest string) error {
|
||||
if _, ok := c.next.Manifests[digest]; ok {
|
||||
return nil
|
||||
}
|
||||
if !imageDigestRE.MatchString(digest) {
|
||||
return fmt.Errorf("%q is not a sha256 digest", digest)
|
||||
}
|
||||
key := imageManifestKey(digest)
|
||||
im, stored := c.prev.Manifests[digest]
|
||||
stored = stored && c.remote[key] == SealedSize(im.Size)
|
||||
var body []byte
|
||||
if !stored {
|
||||
b, mt, err := c.s.Images.Manifest(ctx, repo, digest)
|
||||
if err != nil {
|
||||
return c.gone(repo, digest, err)
|
||||
}
|
||||
if im, err = describeManifest(b, mt); err != nil {
|
||||
return err
|
||||
}
|
||||
body = b
|
||||
}
|
||||
for _, child := range im.Children {
|
||||
if err := c.manifest(ctx, repo, child); err != nil {
|
||||
if errors.Is(err, errSkipped) {
|
||||
return c.gone(repo, digest, fmt.Errorf("%w: child manifest %s", ErrImageGone, child))
|
||||
}
|
||||
return fmt.Errorf("child manifest %s: %w", child, err)
|
||||
}
|
||||
}
|
||||
for _, b := range im.Blobs {
|
||||
bkey := imageBlobKey(b.Digest)
|
||||
if c.remote[bkey] == SealedSize(b.Size) {
|
||||
continue
|
||||
}
|
||||
if err := c.s.putImageBlob(ctx, repo, b); err != nil {
|
||||
return c.gone(repo, digest, fmt.Errorf("blob %s: %w", b.Digest, err))
|
||||
}
|
||||
c.remote[bkey] = SealedSize(b.Size)
|
||||
c.res.ImageBlobsUploaded++
|
||||
c.res.BytesUploaded += b.Size
|
||||
}
|
||||
if body != nil {
|
||||
if err := c.s.putBytes(ctx, key, body); err != nil {
|
||||
return err
|
||||
}
|
||||
c.remote[key] = SealedSize(int64(len(body)))
|
||||
c.res.ImagesUploaded++
|
||||
c.s.logf("copied image %s@%s", repo, digest)
|
||||
}
|
||||
c.next.Manifests[digest] = im
|
||||
return nil
|
||||
}
|
||||
|
||||
// gone turns err into errSkipped when it says the registry lost the manifest
|
||||
// or one of its blobs, keeping the copy an earlier pass made if there is one.
|
||||
func (c *imageCopy) gone(repo, digest string, err error) error {
|
||||
if !errors.Is(err, ErrImageGone) {
|
||||
return err
|
||||
}
|
||||
if _, ok := c.prev.Manifests[digest]; ok && c.prevComplete(digest) {
|
||||
c.res.ImagesIncomplete = append(c.res.ImagesIncomplete, fmt.Sprintf("%s@%s: %v; kept the off-site copy", repo, digest, err))
|
||||
c.addPrev(digest)
|
||||
return nil
|
||||
}
|
||||
c.res.ImagesIncomplete = append(c.res.ImagesIncomplete, fmt.Sprintf("%s@%s: %v; not copied", repo, digest, err))
|
||||
return errSkipped
|
||||
}
|
||||
|
||||
// prevComplete reports whether every object digest needs is in the bucket.
|
||||
func (c *imageCopy) prevComplete(digest string) bool {
|
||||
m, ok := c.prev.Manifests[digest]
|
||||
if !ok || c.remote[imageManifestKey(digest)] != SealedSize(m.Size) {
|
||||
return false
|
||||
}
|
||||
for _, b := range m.Blobs {
|
||||
if c.remote[imageBlobKey(b.Digest)] != SealedSize(b.Size) {
|
||||
return false
|
||||
}
|
||||
}
|
||||
for _, child := range m.Children {
|
||||
if !c.prevComplete(child) {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
func (c *imageCopy) addPrev(digest string) {
|
||||
m := c.prev.Manifests[digest]
|
||||
c.next.Manifests[digest] = m
|
||||
for _, child := range m.Children {
|
||||
c.addPrev(child)
|
||||
}
|
||||
}
|
||||
|
||||
// sameImages reports whether two indexes record the same state.
|
||||
func sameImages(a, b *ImageIndex) bool {
|
||||
ja, err1 := json.Marshal(struct {
|
||||
R map[string]ImageRepo
|
||||
M map[string]ImageManifest
|
||||
}{a.Repositories, a.Manifests})
|
||||
jb, err2 := json.Marshal(struct {
|
||||
R map[string]ImageRepo
|
||||
M map[string]ImageManifest
|
||||
}{b.Repositories, b.Manifests})
|
||||
return err1 == nil && err2 == nil && bytes.Equal(ja, jb)
|
||||
}
|
||||
|
||||
// pruneImages drops the index versions replaced more than ImageHistory ago,
|
||||
// then every blob and manifest no remaining version names.
|
||||
func (s *Syncer) pruneImages(ctx context.Context, versions []string, newest *ImageIndex, remote map[string]int64, res *Result, fail func(string, ...any)) {
|
||||
cutoff := s.now().Add(-ImageHistory)
|
||||
var keep, drop []string
|
||||
for i, v := range versions {
|
||||
if i == len(versions)-1 {
|
||||
break
|
||||
}
|
||||
replaced, err := time.Parse(imageStampLayout, versions[i+1])
|
||||
if err == nil && replaced.Before(cutoff) {
|
||||
drop = append(drop, v)
|
||||
} else {
|
||||
keep = append(keep, v)
|
||||
}
|
||||
}
|
||||
live := map[string]bool{}
|
||||
mark := func(x *ImageIndex) {
|
||||
for d, m := range x.Manifests {
|
||||
live[imageManifestKey(d)] = true
|
||||
for _, b := range m.Blobs {
|
||||
live[imageBlobKey(b.Digest)] = true
|
||||
}
|
||||
}
|
||||
}
|
||||
mark(newest)
|
||||
for _, v := range keep {
|
||||
x, err := LoadImageIndex(ctx, s.Bucket, s.Key, v)
|
||||
if err != nil {
|
||||
fail("read the registry index %s from the bucket: %v; nothing pruned", v, err)
|
||||
return
|
||||
}
|
||||
mark(x)
|
||||
}
|
||||
for _, v := range drop {
|
||||
if err := s.Bucket.Remove(ctx, imageIndexKey(v)); err != nil {
|
||||
fail("prune registry index %s: %v", v, err)
|
||||
return
|
||||
}
|
||||
delete(remote, imageIndexKey(v))
|
||||
res.ImageObjectsPruned++
|
||||
}
|
||||
for _, key := range slices.Sorted(maps.Keys(remote)) {
|
||||
if !strings.HasPrefix(key, imageBlobsDir) && !strings.HasPrefix(key, imageManifestsDir) {
|
||||
continue
|
||||
}
|
||||
if live[key] {
|
||||
continue
|
||||
}
|
||||
if err := s.Bucket.Remove(ctx, key); err != nil {
|
||||
fail("prune %s: %v", key, err)
|
||||
continue
|
||||
}
|
||||
delete(remote, key)
|
||||
res.ImageObjectsPruned++
|
||||
}
|
||||
}
|
||||
|
||||
// putImageBlob streams one blob from the registry into the bucket, checking it
|
||||
// against its digest and size on the way: a blob that does not match fails
|
||||
// the upload instead of storing bytes a restore would push as that digest.
|
||||
func (s *Syncer) putImageBlob(ctx context.Context, repo string, b ImageBlob) error {
|
||||
rc, err := s.Images.Blob(ctx, repo, b.Digest)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer rc.Close()
|
||||
return s.putStream(ctx, imageBlobKey(b.Digest), newDigestReader(rc, b.Digest, b.Size), b.Size)
|
||||
}
|
||||
|
||||
func (s *Syncer) putBytes(ctx context.Context, key string, plain []byte) error {
|
||||
var sealed bytes.Buffer
|
||||
if err := Encrypt(&sealed, bytes.NewReader(plain), s.Key); err != nil {
|
||||
return err
|
||||
}
|
||||
return s.Bucket.Put(ctx, key, &sealed, int64(sealed.Len()))
|
||||
}
|
||||
|
||||
func (s *Syncer) putImageIndex(ctx context.Context, stamp string, x *ImageIndex) error {
|
||||
raw, err := json.Marshal(x)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return s.putBytes(ctx, imageIndexKey(stamp), raw)
|
||||
}
|
||||
|
||||
// digestReader passes a blob through and fails at its end when the bytes do not
|
||||
// hash to digest or are not size long.
|
||||
type digestReader struct {
|
||||
r io.Reader
|
||||
h hash.Hash
|
||||
n int64
|
||||
size int64
|
||||
digest string
|
||||
}
|
||||
|
||||
func newDigestReader(r io.Reader, digest string, size int64) *digestReader {
|
||||
return &digestReader{r: r, h: sha256.New(), size: size, digest: digest}
|
||||
}
|
||||
|
||||
func (d *digestReader) Read(p []byte) (int, error) {
|
||||
n, err := d.r.Read(p)
|
||||
d.h.Write(p[:n])
|
||||
d.n += int64(n)
|
||||
if d.n > d.size {
|
||||
return n, fmt.Errorf("blob %s is longer than the %d bytes its manifest records", d.digest, d.size)
|
||||
}
|
||||
if err == io.EOF {
|
||||
if d.n != d.size {
|
||||
return n, fmt.Errorf("blob %s ended after %d of %d bytes", d.digest, d.n, d.size)
|
||||
}
|
||||
if got := "sha256:" + hex.EncodeToString(d.h.Sum(nil)); got != d.digest {
|
||||
return n, fmt.Errorf("blob %s hashes to %s", d.digest, got)
|
||||
}
|
||||
}
|
||||
return n, err
|
||||
}
|
||||
|
||||
// imageIndexStamps lists the index versions among keys, oldest first.
|
||||
func imageIndexStamps(keys map[string]int64) []string {
|
||||
var out []string
|
||||
for key := range keys {
|
||||
stamp, ok := strings.CutPrefix(key, imageIndexDir)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
stamp, ok = strings.CutSuffix(stamp, imageIndexExt)
|
||||
if _, err := time.Parse(imageStampLayout, stamp); ok && err == nil {
|
||||
out = append(out, stamp)
|
||||
}
|
||||
}
|
||||
sort.Strings(out)
|
||||
return out
|
||||
}
|
||||
|
||||
// ImageIndexes lists the registry index versions in the bucket, oldest first.
|
||||
func ImageIndexes(ctx context.Context, b Bucket) ([]string, error) {
|
||||
objs, err := b.List(ctx, imageIndexDir)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
keys := make(map[string]int64, len(objs))
|
||||
for _, o := range objs {
|
||||
keys[o.Key] = o.Size
|
||||
}
|
||||
return imageIndexStamps(keys), nil
|
||||
}
|
||||
|
||||
// LoadImageIndex reads one index version.
|
||||
func LoadImageIndex(ctx context.Context, b Bucket, key []byte, stamp string) (*ImageIndex, error) {
|
||||
raw, err := getSealed(ctx, b, key, imageIndexKey(stamp), maxImageIndexBytes)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
x := newImageIndex(time.Time{})
|
||||
if err := json.Unmarshal(raw, x); err != nil {
|
||||
return nil, fmt.Errorf("registry index %s: %w", stamp, err)
|
||||
}
|
||||
if x.Repositories == nil {
|
||||
x.Repositories = map[string]ImageRepo{}
|
||||
}
|
||||
if x.Manifests == nil {
|
||||
x.Manifests = map[string]ImageManifest{}
|
||||
}
|
||||
return x, nil
|
||||
}
|
||||
|
||||
// ChooseImageIndex picks the version a restore uses: at when given, otherwise
|
||||
// the newest. A newest version with no images while an older one has some is
|
||||
// what a rebuilt host's first sync records before the restore, so it is
|
||||
// refused with the versions worth choosing instead.
|
||||
func ChooseImageIndex(ctx context.Context, b Bucket, key []byte, at string) (string, *ImageIndex, error) {
|
||||
versions, err := ImageIndexes(ctx, b)
|
||||
if err != nil {
|
||||
return "", nil, err
|
||||
}
|
||||
if len(versions) == 0 {
|
||||
return "", nil, errors.New("the bucket holds no registry index: no sync has copied images yet")
|
||||
}
|
||||
if at != "" {
|
||||
if !slices.Contains(versions, at) {
|
||||
return "", nil, fmt.Errorf("the bucket holds no registry index %s; it holds %s", at, strings.Join(versions, ", "))
|
||||
}
|
||||
x, err := LoadImageIndex(ctx, b, key, at)
|
||||
return at, x, err
|
||||
}
|
||||
newest := versions[len(versions)-1]
|
||||
x, err := LoadImageIndex(ctx, b, key, newest)
|
||||
if err != nil || len(x.Repositories) > 0 {
|
||||
return newest, x, err
|
||||
}
|
||||
var older []string
|
||||
for i := len(versions) - 2; i >= 0; i-- {
|
||||
o, err := LoadImageIndex(ctx, b, key, versions[i])
|
||||
if err != nil {
|
||||
return "", nil, err
|
||||
}
|
||||
if len(o.Repositories) > 0 {
|
||||
older = append(older, fmt.Sprintf("%s (%d repositories, %d images)", versions[i], len(o.Repositories), o.Images()))
|
||||
}
|
||||
}
|
||||
if len(older) == 0 {
|
||||
return newest, x, nil
|
||||
}
|
||||
return "", nil, fmt.Errorf("the newest registry index %s lists no images, which is what a rebuilt host records before its restore; pick the state to restore with -at: %s",
|
||||
newest, strings.Join(older, ", "))
|
||||
}
|
||||
|
||||
// getSealed downloads and decrypts a small object.
|
||||
func getSealed(ctx context.Context, b Bucket, key []byte, objKey string, limit int64) ([]byte, error) {
|
||||
rc, err := b.Get(ctx, objKey)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rc.Close()
|
||||
var out bytes.Buffer
|
||||
if err := Decrypt(&limitWriter{w: &out, left: limit}, rc, key); err != nil {
|
||||
return nil, fmt.Errorf("%s: %w", objKey, err)
|
||||
}
|
||||
return out.Bytes(), nil
|
||||
}
|
||||
|
||||
type limitWriter struct {
|
||||
w io.Writer
|
||||
left int64
|
||||
}
|
||||
|
||||
func (l *limitWriter) Write(p []byte) (int, error) {
|
||||
if int64(len(p)) > l.left {
|
||||
return 0, errors.New("object is larger than expected")
|
||||
}
|
||||
l.left -= int64(len(p))
|
||||
return l.w.Write(p)
|
||||
}
|
||||
|
||||
// FetchImagesResult is what FetchImages did.
|
||||
type FetchImagesResult struct {
|
||||
Repositories int
|
||||
Manifests int
|
||||
Tags int
|
||||
// BlobsPushed and BytesPushed count the blobs the registry lacked.
|
||||
BlobsPushed int
|
||||
BytesPushed int64
|
||||
Failures []string
|
||||
}
|
||||
|
||||
// FetchImages pushes every image x records back into t, each repository's
|
||||
// blobs first, then its manifests by digest (an index after the manifests it
|
||||
// names), then its tags. What t already holds is skipped, so a second run
|
||||
// finishes what the first could not.
|
||||
func FetchImages(ctx context.Context, b Bucket, key []byte, x *ImageIndex, t ImageTarget, log io.Writer) (FetchImagesResult, error) {
|
||||
var res FetchImagesResult
|
||||
bodies := map[string][]byte{}
|
||||
body := func(d string) ([]byte, error) {
|
||||
if raw, ok := bodies[d]; ok {
|
||||
return raw, nil
|
||||
}
|
||||
raw, err := getSealed(ctx, b, key, imageManifestKey(d), maxImageManifestBytes)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
sum := sha256.Sum256(raw)
|
||||
if got := "sha256:" + hex.EncodeToString(sum[:]); got != d {
|
||||
return nil, fmt.Errorf("the stored manifest %s hashes to %s", d, got)
|
||||
}
|
||||
bodies[d] = raw
|
||||
return raw, nil
|
||||
}
|
||||
pushedBlobs := map[string]bool{}
|
||||
for _, repo := range slices.Sorted(maps.Keys(x.Repositories)) {
|
||||
entry := x.Repositories[repo]
|
||||
done := map[string]bool{}
|
||||
var push func(d string) error
|
||||
push = func(d string) error {
|
||||
if done[d] {
|
||||
return nil
|
||||
}
|
||||
m, ok := x.Manifests[d]
|
||||
if !ok {
|
||||
return fmt.Errorf("the index names %s but does not describe it", d)
|
||||
}
|
||||
for _, child := range m.Children {
|
||||
if err := push(child); err != nil {
|
||||
return fmt.Errorf("child %s: %w", child, err)
|
||||
}
|
||||
}
|
||||
for _, bl := range m.Blobs {
|
||||
open := func() (io.ReadCloser, error) {
|
||||
if !pushedBlobs[repo+"@"+bl.Digest] {
|
||||
pushedBlobs[repo+"@"+bl.Digest] = true
|
||||
res.BlobsPushed++
|
||||
res.BytesPushed += bl.Size
|
||||
}
|
||||
return openSealed(ctx, b, key, imageBlobKey(bl.Digest))
|
||||
}
|
||||
if err := t.PutBlob(ctx, repo, bl.Digest, bl.Size, open); err != nil {
|
||||
return fmt.Errorf("blob %s: %w", bl.Digest, err)
|
||||
}
|
||||
}
|
||||
raw, err := body(d)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := t.PutManifest(ctx, repo, d, m.MediaType, raw); err != nil {
|
||||
return err
|
||||
}
|
||||
done[d] = true
|
||||
return nil
|
||||
}
|
||||
ok := true
|
||||
for _, d := range entry.Manifests {
|
||||
if err := ctx.Err(); err != nil {
|
||||
return res, err
|
||||
}
|
||||
if err := push(d); err != nil {
|
||||
res.Failures = append(res.Failures, fmt.Sprintf("%s@%s: %v", repo, d, err))
|
||||
ok = false
|
||||
continue
|
||||
}
|
||||
res.Manifests++
|
||||
}
|
||||
for _, tag := range slices.Sorted(maps.Keys(entry.Tags)) {
|
||||
d := entry.Tags[tag]
|
||||
if !done[d] {
|
||||
continue
|
||||
}
|
||||
if err := t.PutManifest(ctx, repo, tag, x.Manifests[d].MediaType, bodies[d]); err != nil {
|
||||
res.Failures = append(res.Failures, fmt.Sprintf("%s:%s: %v", repo, tag, err))
|
||||
ok = false
|
||||
continue
|
||||
}
|
||||
res.Tags++
|
||||
}
|
||||
if ok {
|
||||
res.Repositories++
|
||||
if log != nil {
|
||||
fmt.Fprintf(log, "felis offsite: restored %s (%d images, %d tags)\n", repo, len(entry.Manifests), len(entry.Tags))
|
||||
}
|
||||
}
|
||||
}
|
||||
if len(res.Failures) > 0 {
|
||||
return res, fmt.Errorf("%d images or tags failed; first: %s", len(res.Failures), res.Failures[0])
|
||||
}
|
||||
return res, nil
|
||||
}
|
||||
|
||||
// openSealed streams the decrypted content of objKey. Decryption authenticates
|
||||
// every segment; the registry checks the whole against its digest.
|
||||
func openSealed(ctx context.Context, b Bucket, key []byte, objKey string) (io.ReadCloser, error) {
|
||||
rc, err := b.Get(ctx, objKey)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
pr, pw := io.Pipe()
|
||||
go func() {
|
||||
err := Decrypt(pw, rc, key)
|
||||
rc.Close()
|
||||
pw.CloseWithError(err)
|
||||
}()
|
||||
return pr, nil
|
||||
}
|
||||
@@ -0,0 +1,454 @@
|
||||
package offsite
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// ---- fake registry ---------------------------------------------------------
|
||||
|
||||
type fakeRepo struct {
|
||||
digests []string
|
||||
tags map[string]string
|
||||
}
|
||||
|
||||
type fakeManifest struct {
|
||||
body []byte
|
||||
mediaType string
|
||||
}
|
||||
|
||||
// fakeRegistry is the registry as the sync reads it and as a restore writes
|
||||
// it: one blob and manifest store shared by every repository, as
|
||||
// distribution keeps it, and per-repository links.
|
||||
type fakeRegistry struct {
|
||||
repos map[string]*fakeRepo
|
||||
blobs map[string][]byte
|
||||
manifests map[string]fakeManifest
|
||||
// linked records which blobs each repository holds, for a restore.
|
||||
linked map[string]map[string]bool
|
||||
revErr map[string]error
|
||||
gone map[string]bool
|
||||
corrupt map[string]bool
|
||||
blobGets int
|
||||
// log is what a restore did, in order.
|
||||
log []string
|
||||
}
|
||||
|
||||
func newFakeRegistry() *fakeRegistry {
|
||||
return &fakeRegistry{
|
||||
repos: map[string]*fakeRepo{}, blobs: map[string][]byte{}, manifests: map[string]fakeManifest{},
|
||||
linked: map[string]map[string]bool{}, revErr: map[string]error{}, gone: map[string]bool{}, corrupt: map[string]bool{},
|
||||
}
|
||||
}
|
||||
|
||||
func digestOf(b []byte) string {
|
||||
sum := sha256.Sum256(b)
|
||||
return "sha256:" + hex.EncodeToString(sum[:])
|
||||
}
|
||||
|
||||
func (r *fakeRegistry) blob(data []byte) map[string]any {
|
||||
d := digestOf(data)
|
||||
r.blobs[d] = data
|
||||
return map[string]any{"mediaType": "application/octet-stream", "digest": d, "size": len(data)}
|
||||
}
|
||||
|
||||
func (r *fakeRegistry) link(repo, digest string, tags ...string) {
|
||||
rp := r.repos[repo]
|
||||
if rp == nil {
|
||||
rp = &fakeRepo{tags: map[string]string{}}
|
||||
r.repos[repo] = rp
|
||||
}
|
||||
if !slices.Contains(rp.digests, digest) {
|
||||
rp.digests = append(rp.digests, digest)
|
||||
}
|
||||
for _, t := range tags {
|
||||
rp.tags[t] = digest
|
||||
}
|
||||
}
|
||||
|
||||
func (r *fakeRegistry) putManifest(mediaType string, m map[string]any) string {
|
||||
m["schemaVersion"] = 2
|
||||
m["mediaType"] = mediaType
|
||||
body, _ := json.Marshal(m)
|
||||
d := digestOf(body)
|
||||
r.manifests[d] = fakeManifest{body: body, mediaType: mediaType}
|
||||
return d
|
||||
}
|
||||
|
||||
// image stores an image manifest of the given layers in repo and returns its digest.
|
||||
func (r *fakeRegistry) image(repo, tag string, layers ...[]byte) string {
|
||||
var ls []any
|
||||
for _, l := range layers {
|
||||
ls = append(ls, r.blob(l))
|
||||
}
|
||||
cfg := r.blob([]byte(fmt.Sprintf(`{"architecture":"arm64","repo":%q,"tag":%q}`, repo, tag)))
|
||||
d := r.putManifest(mediaDockerManifest2, map[string]any{"config": cfg, "layers": ls})
|
||||
r.link(repo, d, tag)
|
||||
return d
|
||||
}
|
||||
|
||||
// index stores an image index over children in repo.
|
||||
func (r *fakeRegistry) index(repo, tag string, children ...string) string {
|
||||
var ms []any
|
||||
for _, c := range children {
|
||||
ms = append(ms, map[string]any{"mediaType": mediaOCIManifest, "digest": c, "size": len(r.manifests[c].body)})
|
||||
}
|
||||
d := r.putManifest(mediaOCIIndex, map[string]any{"manifests": ms})
|
||||
r.link(repo, d, tag)
|
||||
return d
|
||||
}
|
||||
|
||||
func (r *fakeRegistry) Repositories(context.Context) ([]string, error) {
|
||||
var out []string
|
||||
for name := range r.repos {
|
||||
out = append(out, name)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (r *fakeRegistry) Revisions(_ context.Context, repo string) ([]string, map[string]string, error) {
|
||||
if err := r.revErr[repo]; err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
rp := r.repos[repo]
|
||||
return slices.Clone(rp.digests), rp.tags, nil
|
||||
}
|
||||
|
||||
func (r *fakeRegistry) Manifest(_ context.Context, repo, digest string) ([]byte, string, error) {
|
||||
m, ok := r.manifests[digest]
|
||||
if !ok || r.gone[digest] {
|
||||
return nil, "", fmt.Errorf("%w: manifest %s", ErrImageGone, digest)
|
||||
}
|
||||
return m.body, m.mediaType, nil
|
||||
}
|
||||
|
||||
func (r *fakeRegistry) Blob(_ context.Context, repo, digest string) (io.ReadCloser, error) {
|
||||
r.blobGets++
|
||||
b, ok := r.blobs[digest]
|
||||
if !ok || r.gone[digest] {
|
||||
return nil, fmt.Errorf("%w: blob %s", ErrImageGone, digest)
|
||||
}
|
||||
if r.corrupt[digest] {
|
||||
b = append(bytes.Clone(b), 'x')
|
||||
}
|
||||
return io.NopCloser(bytes.NewReader(b)), nil
|
||||
}
|
||||
|
||||
func (r *fakeRegistry) PutBlob(_ context.Context, repo, digest string, size int64, open func() (io.ReadCloser, error)) error {
|
||||
if r.linked[repo][digest] {
|
||||
return nil
|
||||
}
|
||||
rc, err := open()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer rc.Close()
|
||||
data, err := io.ReadAll(rc)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if int64(len(data)) != size || digestOf(data) != digest {
|
||||
return fmt.Errorf("blob upload does not match %s", digest)
|
||||
}
|
||||
r.blobs[digest] = data
|
||||
if r.linked[repo] == nil {
|
||||
r.linked[repo] = map[string]bool{}
|
||||
}
|
||||
r.linked[repo][digest] = true
|
||||
r.log = append(r.log, "blob "+repo+" "+digest)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *fakeRegistry) PutManifest(_ context.Context, repo, reference, mediaType string, body []byte) error {
|
||||
d := digestOf(body)
|
||||
if strings.HasPrefix(reference, "sha256:") && reference != d {
|
||||
return fmt.Errorf("manifest %s hashes to %s", reference, d)
|
||||
}
|
||||
// Like distribution: everything a manifest names must be in the repository.
|
||||
im, err := describeManifest(body, mediaType)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, b := range im.Blobs {
|
||||
if !r.linked[repo][b.Digest] {
|
||||
return fmt.Errorf("blob unknown: %s", b.Digest)
|
||||
}
|
||||
}
|
||||
for _, c := range im.Children {
|
||||
if r.repos[repo] == nil || !slices.Contains(r.repos[repo].digests, c) {
|
||||
return fmt.Errorf("manifest unknown: %s", c)
|
||||
}
|
||||
}
|
||||
r.manifests[d] = fakeManifest{body: body, mediaType: mediaType}
|
||||
if strings.HasPrefix(reference, "sha256:") {
|
||||
r.link(repo, d)
|
||||
} else {
|
||||
r.link(repo, d, reference)
|
||||
}
|
||||
r.log = append(r.log, "manifest "+repo+" "+reference)
|
||||
return nil
|
||||
}
|
||||
|
||||
func layer(n int) []byte {
|
||||
b := make([]byte, n)
|
||||
rand.Read(b)
|
||||
return b
|
||||
}
|
||||
|
||||
func newImageSyncer(t *testing.T, reg *fakeRegistry) (*Syncer, *memBucket, *time.Time) {
|
||||
t.Helper()
|
||||
b := newMemBucket()
|
||||
clock := now
|
||||
return &Syncer{
|
||||
Bucket: b, Catalog: &fakeCatalog{}, Key: testKey(t), Images: reg,
|
||||
Now: func() time.Time { return clock },
|
||||
}, b, &clock
|
||||
}
|
||||
|
||||
func keysUnder(b *memBucket, prefix string) []string {
|
||||
objs, _ := b.List(context.Background(), prefix)
|
||||
var out []string
|
||||
for _, o := range objs {
|
||||
out = append(out, o.Key)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// ---- tests -----------------------------------------------------------------
|
||||
|
||||
// TestSyncImagesCopiesUserImages: user repositories are copied blob by blob,
|
||||
// each blob once however many images share it, and the platform's reserved
|
||||
// repositories are left alone. A second run with nothing new sends nothing.
|
||||
func TestSyncImagesCopiesUserImages(t *testing.T) {
|
||||
reg := newFakeRegistry()
|
||||
shared := layer(3000)
|
||||
a := reg.image("user-uploads/sub-a", "latest", shared, layer(70000))
|
||||
child := reg.image("e2e/multi", "arm64", shared)
|
||||
idx := reg.index("e2e/multi", "latest", child)
|
||||
reg.image("felis/felis", "v1", layer(5000))
|
||||
reg.image("mirror/trivy-db", "2", layer(5000))
|
||||
|
||||
s, b, _ := newImageSyncer(t, reg)
|
||||
res, err := s.Run(context.Background())
|
||||
if err != nil {
|
||||
t.Fatalf("Run: %v", err)
|
||||
}
|
||||
if res.ImageRepos != 2 || res.Images != 3 || res.ImagesUploaded != 3 {
|
||||
t.Fatalf("result = %+v, want 2 repositories, 3 images, 3 uploaded", res)
|
||||
}
|
||||
// shared layer once, one unique layer, two configs
|
||||
if got := keysUnder(b, imageBlobsDir); len(got) != 4 || res.ImageBlobsUploaded != 4 {
|
||||
t.Fatalf("blobs in bucket = %v (uploaded %d), want 4", got, res.ImageBlobsUploaded)
|
||||
}
|
||||
for _, d := range []string{a, child, idx} {
|
||||
if _, ok := b.objs[imageManifestKey(d)]; !ok {
|
||||
t.Errorf("manifest %s not in the bucket", d)
|
||||
}
|
||||
}
|
||||
x, err := LoadImageIndex(context.Background(), b, s.Key, res.ImageIndex)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, ok := x.Repositories["felis/felis"]; ok {
|
||||
t.Fatal("reserved repository felis/felis was copied")
|
||||
}
|
||||
if got := x.Repositories["e2e/multi"]; got.Tags["latest"] != idx || got.Tags["arm64"] != child {
|
||||
t.Fatalf("e2e/multi tags = %v", got.Tags)
|
||||
}
|
||||
|
||||
puts, gets := b.puts, reg.blobGets
|
||||
res, err = s.Run(context.Background())
|
||||
if err != nil {
|
||||
t.Fatalf("second Run: %v", err)
|
||||
}
|
||||
if b.puts != puts || reg.blobGets != gets {
|
||||
t.Fatalf("second run sent %d objects and read %d blobs, want none", b.puts-puts, reg.blobGets-gets)
|
||||
}
|
||||
if len(keysUnder(b, imageIndexDir)) != 1 {
|
||||
t.Fatalf("an unchanged registry wrote another index: %v", keysUnder(b, imageIndexDir))
|
||||
}
|
||||
}
|
||||
|
||||
// TestSyncImagesRejectsCorruptBlob: bytes that do not hash to the digest never
|
||||
// become that digest's object, the image stays out of the index, and the run
|
||||
// fails so the next one tries again.
|
||||
func TestSyncImagesRejectsCorruptBlob(t *testing.T) {
|
||||
reg := newFakeRegistry()
|
||||
bad := layer(100)
|
||||
reg.image("user/a", "v1", bad)
|
||||
reg.corrupt[digestOf(bad)] = true
|
||||
|
||||
s, b, _ := newImageSyncer(t, reg)
|
||||
res, err := s.Run(context.Background())
|
||||
if err == nil || !strings.Contains(err.Error(), "longer than") {
|
||||
t.Fatalf("Run = %v, want the corrupt blob to fail it", err)
|
||||
}
|
||||
if _, ok := b.objs[imageBlobKey(digestOf(bad))]; ok {
|
||||
t.Fatal("the corrupt blob was stored")
|
||||
}
|
||||
if len(keysUnder(b, imageManifestsDir)) != 0 || res.Images != 0 {
|
||||
t.Fatalf("manifest stored without its blob: %v, images=%d", keysUnder(b, imageManifestsDir), res.Images)
|
||||
}
|
||||
}
|
||||
|
||||
// TestSyncImagesSkipsWhatTheRegistryLost: a manifest whose blob the registry no
|
||||
// longer has is reported, not copied, and does not fail the run; one copied
|
||||
// earlier keeps its off-site copy.
|
||||
func TestSyncImagesSkipsWhatTheRegistryLost(t *testing.T) {
|
||||
reg := newFakeRegistry()
|
||||
keptLayer := layer(100)
|
||||
kept := reg.image("user/a", "v1", keptLayer)
|
||||
s, b, _ := newImageSyncer(t, reg)
|
||||
if _, err := s.Run(context.Background()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
lostLayer := layer(100)
|
||||
lost := reg.image("user/a", "v2", lostLayer)
|
||||
reg.gone[digestOf(lostLayer)] = true
|
||||
res, err := s.Run(context.Background())
|
||||
if err != nil {
|
||||
t.Fatalf("Run = %v, want a lost blob to be reported only", err)
|
||||
}
|
||||
if len(res.ImagesIncomplete) != 1 || !strings.Contains(res.ImagesIncomplete[0], lost) {
|
||||
t.Fatalf("incomplete = %v, want %s", res.ImagesIncomplete, lost)
|
||||
}
|
||||
x, _ := LoadImageIndex(context.Background(), b, s.Key, res.ImageIndex)
|
||||
if got := x.Repositories["user/a"]; !slices.Equal(got.Manifests, []string{kept}) || got.Tags["v2"] != "" {
|
||||
t.Fatalf("user/a = %+v, want only the complete image", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestSyncImagesKeepsReplacedStates: a registry that comes back empty (a host
|
||||
// being rebuilt) writes a new index version but prunes nothing; the objects of
|
||||
// the old state go only ImageHistory after it was replaced.
|
||||
func TestSyncImagesKeepsReplacedStates(t *testing.T) {
|
||||
reg := newFakeRegistry()
|
||||
reg.image("user/a", "v1", layer(100))
|
||||
s, b, clock := newImageSyncer(t, reg)
|
||||
if _, err := s.Run(context.Background()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
objects := len(keysUnder(b, imageBlobsDir)) + len(keysUnder(b, imageManifestsDir))
|
||||
|
||||
reg.repos = map[string]*fakeRepo{}
|
||||
*clock = clock.Add(time.Hour)
|
||||
res, err := s.Run(context.Background())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if res.ImageRepos != 0 || len(keysUnder(b, imageIndexDir)) != 2 || res.ImageObjectsPruned != 0 {
|
||||
t.Fatalf("after emptying: repos=%d versions=%v pruned=%d", res.ImageRepos, keysUnder(b, imageIndexDir), res.ImageObjectsPruned)
|
||||
}
|
||||
if got := len(keysUnder(b, imageBlobsDir)) + len(keysUnder(b, imageManifestsDir)); got != objects {
|
||||
t.Fatalf("objects = %d, want the old state's %d kept", got, objects)
|
||||
}
|
||||
// A restore now refuses the empty newest state and names the old one.
|
||||
if _, _, err := ChooseImageIndex(context.Background(), b, s.Key, ""); err == nil || !strings.Contains(err.Error(), "-at") {
|
||||
t.Fatalf("ChooseImageIndex = %v, want it to point at -at", err)
|
||||
}
|
||||
|
||||
*clock = clock.Add(ImageHistory - 2*time.Hour)
|
||||
if res, _ := s.Run(context.Background()); res.ImageObjectsPruned != 0 {
|
||||
t.Fatalf("pruned %d objects inside the history window", res.ImageObjectsPruned)
|
||||
}
|
||||
*clock = clock.Add(3 * time.Hour)
|
||||
res, err = s.Run(context.Background())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if res.ImageObjectsPruned != objects+1 || len(keysUnder(b, imageBlobsDir)) != 0 || len(keysUnder(b, imageIndexDir)) != 1 {
|
||||
t.Fatalf("pruned %d, left blobs %v and versions %v", res.ImageObjectsPruned, keysUnder(b, imageBlobsDir), keysUnder(b, imageIndexDir))
|
||||
}
|
||||
}
|
||||
|
||||
// TestSyncImagesUnreadableRepoKeepsItsEntry: a repository the registry fails to
|
||||
// list keeps its last recorded state, and the failed run prunes nothing.
|
||||
func TestSyncImagesUnreadableRepoKeepsItsEntry(t *testing.T) {
|
||||
reg := newFakeRegistry()
|
||||
a := reg.image("user/a", "v1", layer(100))
|
||||
s, b, clock := newImageSyncer(t, reg)
|
||||
if _, err := s.Run(context.Background()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
reg.revErr["user/a"] = errors.New("gate restarting")
|
||||
reg.image("user/b", "v1", layer(100))
|
||||
*clock = clock.Add(ImageHistory * 2)
|
||||
res, err := s.Run(context.Background())
|
||||
if err == nil {
|
||||
t.Fatal("Run succeeded with an unreadable repository")
|
||||
}
|
||||
x, _ := LoadImageIndex(context.Background(), b, s.Key, res.ImageIndex)
|
||||
if got := x.Repositories["user/a"]; !slices.Equal(got.Manifests, []string{a}) {
|
||||
t.Fatalf("user/a = %+v, want its last recorded state", got)
|
||||
}
|
||||
if _, ok := x.Repositories["user/b"]; !ok || res.ImageObjectsPruned != 0 {
|
||||
t.Fatalf("user/b missing or pruned %d", res.ImageObjectsPruned)
|
||||
}
|
||||
}
|
||||
|
||||
// TestFetchImagesRestoresRegistry: a fresh registry gets back every user image
|
||||
// with the same digests and tags, blobs before the manifests that name them and
|
||||
// an index after its children; a second run pushes nothing.
|
||||
func TestFetchImagesRestoresRegistry(t *testing.T) {
|
||||
reg := newFakeRegistry()
|
||||
shared := layer(3000)
|
||||
a := reg.image("user/a", "v1", shared, layer(9000))
|
||||
child := reg.image("user/multi", "arm64", shared)
|
||||
idx := reg.index("user/multi", "latest", child)
|
||||
s, b, _ := newImageSyncer(t, reg)
|
||||
if _, err := s.Run(context.Background()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
stamp, x, err := ChooseImageIndex(context.Background(), b, s.Key, "")
|
||||
if err != nil || stamp == "" {
|
||||
t.Fatalf("ChooseImageIndex = %q, %v", stamp, err)
|
||||
}
|
||||
fresh := newFakeRegistry()
|
||||
res, err := FetchImages(context.Background(), b, s.Key, x, fresh, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("FetchImages: %v", err)
|
||||
}
|
||||
if res.Repositories != 2 || res.Manifests != 3 || res.Tags != 3 || res.BlobsPushed != 5 {
|
||||
t.Fatalf("result = %+v", res)
|
||||
}
|
||||
for repo, want := range map[string]map[string]string{
|
||||
"user/a": {"v1": a},
|
||||
"user/multi": {"arm64": child, "latest": idx},
|
||||
} {
|
||||
for tag, d := range want {
|
||||
if got := fresh.repos[repo].tags[tag]; got != d {
|
||||
t.Errorf("%s:%s = %s, want %s", repo, tag, got, d)
|
||||
}
|
||||
}
|
||||
}
|
||||
if !bytes.Equal(fresh.manifests[idx].body, reg.manifests[idx].body) {
|
||||
t.Fatal("restored index differs from the original")
|
||||
}
|
||||
pos := func(entry string) int { return slices.Index(fresh.log, entry) }
|
||||
if pos("manifest user/multi "+child) > pos("manifest user/multi "+idx) {
|
||||
t.Fatalf("index pushed before its child: %v", fresh.log)
|
||||
}
|
||||
|
||||
n := len(fresh.log)
|
||||
if _, err := FetchImages(context.Background(), b, s.Key, x, fresh, nil); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, e := range fresh.log[n:] {
|
||||
if strings.HasPrefix(e, "blob ") {
|
||||
t.Fatalf("second restore uploaded a blob again: %s", e)
|
||||
}
|
||||
}
|
||||
}
|
||||
+39
-16
@@ -1,8 +1,10 @@
|
||||
// Package offsite keeps a second copy of what a lost node would take with it:
|
||||
// every world archive (world_backups) and the newest control-plane database
|
||||
// bundles (internal/dbbackup), encrypted, in an S3-compatible bucket off the
|
||||
// machine. `felis offsite sync` runs it from felis-offsite.timer on the host,
|
||||
// which is where both the archive volume and the bundle directory live.
|
||||
// every world archive (world_backups), the newest control-plane database
|
||||
// bundles (internal/dbbackup) and the user images in the platform registry
|
||||
// (images.go), encrypted, in an S3-compatible bucket off the machine. `felis
|
||||
// offsite sync` runs it from felis-offsite.timer on the host, which is where
|
||||
// the archive volume and the bundle directory live and where the registry
|
||||
// answers on its loopback hostPort.
|
||||
//
|
||||
// The database records the copy: world_backups.offsite_at is set once an
|
||||
// archive's object is in the bucket, and with [offsite] configured the reaper
|
||||
@@ -89,6 +91,9 @@ type Syncer struct {
|
||||
// bucket keeps.
|
||||
DBDir string
|
||||
DBKeep int
|
||||
// Images is the platform registry whose user images are copied (images.go);
|
||||
// nil copies none.
|
||||
Images ImageSource
|
||||
Now func() time.Time
|
||||
Log io.Writer
|
||||
}
|
||||
@@ -110,7 +115,19 @@ type Result struct {
|
||||
DBPruned int `json:"db_pruned"`
|
||||
RemoteDB int `json:"remote_db"`
|
||||
NewestDB string `json:"newest_db,omitempty"`
|
||||
Errors []string `json:"errors,omitempty"`
|
||||
// ImageIndex is the newest registry index version in the bucket, which
|
||||
// lists ImageRepos repositories holding Images images.
|
||||
ImageIndex string `json:"image_index,omitempty"`
|
||||
ImageRepos int `json:"image_repos"`
|
||||
Images int `json:"images"`
|
||||
ImagesUploaded int `json:"images_uploaded"`
|
||||
ImageBlobsUploaded int `json:"image_blobs_uploaded"`
|
||||
ImageObjectsPruned int `json:"image_objects_pruned"`
|
||||
RemoteImageBytes int64 `json:"remote_image_bytes"`
|
||||
// ImagesIncomplete are manifests the registry lists without holding all
|
||||
// of them, so there was nothing whole to copy.
|
||||
ImagesIncomplete []string `json:"images_incomplete,omitempty"`
|
||||
Errors []string `json:"errors,omitempty"`
|
||||
}
|
||||
|
||||
func (s *Syncer) now() time.Time {
|
||||
@@ -126,9 +143,9 @@ func (s *Syncer) logf(format string, args ...any) {
|
||||
}
|
||||
}
|
||||
|
||||
// Run does one pass: world archives, then database bundles, then expiry. A
|
||||
// failure on one item is recorded and the pass carries on; the returned error
|
||||
// is non-nil when anything failed.
|
||||
// Run does one pass: world archives, database bundles, registry images, then
|
||||
// expiry. A failure on one item is recorded and the pass carries on; the
|
||||
// returned error is non-nil when anything failed.
|
||||
func (s *Syncer) Run(ctx context.Context) (Result, error) {
|
||||
var res Result
|
||||
fail := func(format string, args ...any) {
|
||||
@@ -143,6 +160,7 @@ func (s *Syncer) Run(ctx context.Context) (Result, error) {
|
||||
}
|
||||
s.syncWorlds(ctx, remoteWorlds, &res, fail)
|
||||
s.syncDB(ctx, &res, fail)
|
||||
s.syncImages(ctx, &res, fail)
|
||||
s.expireWorlds(ctx, remoteWorlds, &res, fail)
|
||||
|
||||
for _, size := range remoteWorlds {
|
||||
@@ -327,23 +345,28 @@ func (s *Syncer) expireWorlds(ctx context.Context, remote map[string]int64, res
|
||||
}
|
||||
}
|
||||
|
||||
// putFile encrypts the file at p into key. The sealed size is known in
|
||||
// advance, so the upload streams: nothing larger than one part is buffered.
|
||||
// putFile encrypts the file at p into key.
|
||||
func (s *Syncer) putFile(ctx context.Context, key, p string, size int64) error {
|
||||
f, err := os.Open(p)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer f.Close()
|
||||
// A file that changed size under us would not match the declared length;
|
||||
// LimitReader keeps the stream to the size we announced and the bucket's
|
||||
// length check catches a short one.
|
||||
return s.putStream(ctx, key, io.LimitReader(f, size), size)
|
||||
}
|
||||
|
||||
// putStream encrypts size bytes of src into key. The sealed size is known in
|
||||
// advance, so the upload streams: nothing larger than one part is buffered.
|
||||
// An error from src fails the upload.
|
||||
func (s *Syncer) putStream(ctx context.Context, key string, src io.Reader, size int64) error {
|
||||
pr, pw := io.Pipe()
|
||||
go func() {
|
||||
// A file that changed size under us would not match the declared
|
||||
// length; LimitReader keeps the stream to the size we announced and
|
||||
// the length check below catches a short one.
|
||||
err := Encrypt(pw, io.LimitReader(f, size), s.Key)
|
||||
pw.CloseWithError(err)
|
||||
pw.CloseWithError(Encrypt(pw, src, s.Key))
|
||||
}()
|
||||
err = s.Bucket.Put(ctx, key, pr, SealedSize(size))
|
||||
err := s.Bucket.Put(ctx, key, pr, SealedSize(size))
|
||||
pr.CloseWithError(errors.New("upload finished"))
|
||||
return err
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user