Merge pull request #1880 from AkihiroSuda/refactor-importer
importer: refactor and fix GC
This commit is contained in:
Derek McGowan
21
images/importexport.go
Normal file
21
images/importexport.go
Normal file
@@ -0,0 +1,21 @@
|
||||
package images
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
|
||||
"github.com/containerd/containerd/content"
|
||||
ocispec "github.com/opencontainers/image-spec/specs-go/v1"
|
||||
)
|
||||
|
||||
// Importer is the interface for image importer.
|
||||
type Importer interface {
|
||||
// Import imports an image from a tar stream.
|
||||
Import(ctx context.Context, store content.Store, reader io.Reader) ([]Image, error)
|
||||
}
|
||||
|
||||
// Exporter is the interface for image exporter.
|
||||
type Exporter interface {
|
||||
// Export exports an image to a tar stream.
|
||||
Export(ctx context.Context, store content.Store, desc ocispec.Descriptor, writer io.Writer) error
|
||||
}
|
||||
188
images/oci/exporter.go
Normal file
188
images/oci/exporter.go
Normal file
@@ -0,0 +1,188 @@
|
||||
package oci
|
||||
|
||||
import (
|
||||
"archive/tar"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"io"
|
||||
"sort"
|
||||
|
||||
"github.com/containerd/containerd/content"
|
||||
"github.com/containerd/containerd/images"
|
||||
"github.com/containerd/containerd/platforms"
|
||||
ocispecs "github.com/opencontainers/image-spec/specs-go"
|
||||
ocispec "github.com/opencontainers/image-spec/specs-go/v1"
|
||||
"github.com/pkg/errors"
|
||||
)
|
||||
|
||||
// V1Exporter implements OCI Image Spec v1.
|
||||
// It is up to caller to put "org.opencontainers.image.ref.name" annotation to desc.
|
||||
//
|
||||
// TODO(AkihiroSuda): add V1Exporter{TranslateMediaTypes: true} that transforms media types,
|
||||
// e.g. application/vnd.docker.image.rootfs.diff.tar.gzip
|
||||
// -> application/vnd.oci.image.layer.v1.tar+gzip
|
||||
type V1Exporter struct {
|
||||
}
|
||||
|
||||
// Export implements Exporter.
|
||||
func (oe *V1Exporter) Export(ctx context.Context, store content.Store, desc ocispec.Descriptor, writer io.Writer) error {
|
||||
tw := tar.NewWriter(writer)
|
||||
defer tw.Close()
|
||||
|
||||
records := []tarRecord{
|
||||
ociLayoutFile(""),
|
||||
ociIndexRecord(desc),
|
||||
}
|
||||
|
||||
algorithms := map[string]struct{}{}
|
||||
exportHandler := func(ctx context.Context, desc ocispec.Descriptor) ([]ocispec.Descriptor, error) {
|
||||
records = append(records, blobRecord(store, desc))
|
||||
algorithms[desc.Digest.Algorithm().String()] = struct{}{}
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
handlers := images.Handlers(
|
||||
images.ChildrenHandler(store, platforms.Default()),
|
||||
images.HandlerFunc(exportHandler),
|
||||
)
|
||||
|
||||
// Walk sequentially since the number of fetchs is likely one and doing in
|
||||
// parallel requires locking the export handler
|
||||
if err := images.Walk(ctx, handlers, desc); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if len(algorithms) > 0 {
|
||||
records = append(records, directoryRecord("blobs/", 0755))
|
||||
for alg := range algorithms {
|
||||
records = append(records, directoryRecord("blobs/"+alg+"/", 0755))
|
||||
}
|
||||
}
|
||||
|
||||
return writeTar(ctx, tw, records)
|
||||
}
|
||||
|
||||
type tarRecord struct {
|
||||
Header *tar.Header
|
||||
CopyTo func(context.Context, io.Writer) (int64, error)
|
||||
}
|
||||
|
||||
func blobRecord(cs content.Store, desc ocispec.Descriptor) tarRecord {
|
||||
path := "blobs/" + desc.Digest.Algorithm().String() + "/" + desc.Digest.Hex()
|
||||
return tarRecord{
|
||||
Header: &tar.Header{
|
||||
Name: path,
|
||||
Mode: 0444,
|
||||
Size: desc.Size,
|
||||
Typeflag: tar.TypeReg,
|
||||
},
|
||||
CopyTo: func(ctx context.Context, w io.Writer) (int64, error) {
|
||||
r, err := cs.ReaderAt(ctx, desc.Digest)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
defer r.Close()
|
||||
|
||||
// Verify digest
|
||||
dgstr := desc.Digest.Algorithm().Digester()
|
||||
|
||||
n, err := io.Copy(io.MultiWriter(w, dgstr.Hash()), content.NewReader(r))
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
if dgstr.Digest() != desc.Digest {
|
||||
return 0, errors.Errorf("unexpected digest %s copied", dgstr.Digest())
|
||||
}
|
||||
return n, nil
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func directoryRecord(name string, mode int64) tarRecord {
|
||||
return tarRecord{
|
||||
Header: &tar.Header{
|
||||
Name: name,
|
||||
Mode: mode,
|
||||
Typeflag: tar.TypeDir,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func ociLayoutFile(version string) tarRecord {
|
||||
if version == "" {
|
||||
version = ocispec.ImageLayoutVersion
|
||||
}
|
||||
layout := ocispec.ImageLayout{
|
||||
Version: version,
|
||||
}
|
||||
|
||||
b, err := json.Marshal(layout)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
return tarRecord{
|
||||
Header: &tar.Header{
|
||||
Name: ocispec.ImageLayoutFile,
|
||||
Mode: 0444,
|
||||
Size: int64(len(b)),
|
||||
Typeflag: tar.TypeReg,
|
||||
},
|
||||
CopyTo: func(ctx context.Context, w io.Writer) (int64, error) {
|
||||
n, err := w.Write(b)
|
||||
return int64(n), err
|
||||
},
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
func ociIndexRecord(manifests ...ocispec.Descriptor) tarRecord {
|
||||
index := ocispec.Index{
|
||||
Versioned: ocispecs.Versioned{
|
||||
SchemaVersion: 2,
|
||||
},
|
||||
Manifests: manifests,
|
||||
}
|
||||
|
||||
b, err := json.Marshal(index)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
return tarRecord{
|
||||
Header: &tar.Header{
|
||||
Name: "index.json",
|
||||
Mode: 0644,
|
||||
Size: int64(len(b)),
|
||||
Typeflag: tar.TypeReg,
|
||||
},
|
||||
CopyTo: func(ctx context.Context, w io.Writer) (int64, error) {
|
||||
n, err := w.Write(b)
|
||||
return int64(n), err
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func writeTar(ctx context.Context, tw *tar.Writer, records []tarRecord) error {
|
||||
sort.Slice(records, func(i, j int) bool {
|
||||
return records[i].Header.Name < records[j].Header.Name
|
||||
})
|
||||
|
||||
for _, record := range records {
|
||||
if err := tw.WriteHeader(record.Header); err != nil {
|
||||
return err
|
||||
}
|
||||
if record.CopyTo != nil {
|
||||
n, err := record.CopyTo(ctx, tw)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if n != record.Header.Size {
|
||||
return errors.Errorf("unexpected copy size for %s", record.Header.Name)
|
||||
}
|
||||
} else if record.Header.Size > 0 {
|
||||
return errors.Errorf("no content to write to record with non-zero size for %s", record.Header.Name)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
188
images/oci/importer.go
Normal file
188
images/oci/importer.go
Normal file
@@ -0,0 +1,188 @@
|
||||
// Package oci provides the importer and the exporter for OCI Image Spec.
|
||||
package oci
|
||||
|
||||
import (
|
||||
"archive/tar"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"io/ioutil"
|
||||
"path"
|
||||
"strings"
|
||||
|
||||
"github.com/containerd/containerd/content"
|
||||
"github.com/containerd/containerd/errdefs"
|
||||
"github.com/containerd/containerd/images"
|
||||
digest "github.com/opencontainers/go-digest"
|
||||
ocispec "github.com/opencontainers/image-spec/specs-go/v1"
|
||||
"github.com/pkg/errors"
|
||||
)
|
||||
|
||||
// V1Importer implements OCI Image Spec v1.
|
||||
type V1Importer struct {
|
||||
// ImageName is preprended to either `:` + OCI ref name or `@` + digest (for anonymous refs).
|
||||
// This field is mandatory atm, but may change in the future. maybe ref map[string]string as in moby/moby#33355
|
||||
ImageName string
|
||||
}
|
||||
|
||||
var _ images.Importer = &V1Importer{}
|
||||
|
||||
// Import implements Importer.
|
||||
func (oi *V1Importer) Import(ctx context.Context, store content.Store, reader io.Reader) ([]images.Image, error) {
|
||||
if oi.ImageName == "" {
|
||||
return nil, errors.New("ImageName not set")
|
||||
}
|
||||
tr := tar.NewReader(reader)
|
||||
var imgrecs []images.Image
|
||||
foundIndexJSON := false
|
||||
for {
|
||||
hdr, err := tr.Next()
|
||||
if err == io.EOF {
|
||||
break
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if hdr.Typeflag != tar.TypeReg && hdr.Typeflag != tar.TypeRegA {
|
||||
continue
|
||||
}
|
||||
hdrName := path.Clean(hdr.Name)
|
||||
if hdrName == "index.json" {
|
||||
if foundIndexJSON {
|
||||
return nil, errors.New("duplicated index.json")
|
||||
}
|
||||
foundIndexJSON = true
|
||||
imgrecs, err = onUntarIndexJSON(tr, oi.ImageName)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
continue
|
||||
}
|
||||
if strings.HasPrefix(hdrName, "blobs/") {
|
||||
if err := onUntarBlob(ctx, tr, store, hdrName, hdr.Size); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
}
|
||||
if !foundIndexJSON {
|
||||
return nil, errors.New("no index.json found")
|
||||
}
|
||||
for _, img := range imgrecs {
|
||||
err := setGCRefContentLabels(ctx, store, img.Target)
|
||||
if err != nil {
|
||||
return imgrecs, err
|
||||
}
|
||||
}
|
||||
// FIXME(AkihiroSuda): set GC labels for unreferrenced blobs (i.e. with unknown media types)?
|
||||
return imgrecs, nil
|
||||
}
|
||||
|
||||
func onUntarIndexJSON(r io.Reader, imageName string) ([]images.Image, error) {
|
||||
b, err := ioutil.ReadAll(r)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var idx ocispec.Index
|
||||
if err := json.Unmarshal(b, &idx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var imgrecs []images.Image
|
||||
for _, m := range idx.Manifests {
|
||||
ref, err := normalizeImageRef(imageName, m)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
imgrecs = append(imgrecs, images.Image{
|
||||
Name: ref,
|
||||
Target: m,
|
||||
})
|
||||
}
|
||||
return imgrecs, nil
|
||||
}
|
||||
|
||||
func normalizeImageRef(imageName string, manifest ocispec.Descriptor) (string, error) {
|
||||
digest := manifest.Digest
|
||||
if digest == "" {
|
||||
return "", errors.Errorf("manifest with empty digest: %v", manifest)
|
||||
}
|
||||
ociRef := manifest.Annotations[ocispec.AnnotationRefName]
|
||||
if ociRef == "" {
|
||||
return imageName + "@" + digest.String(), nil
|
||||
}
|
||||
return imageName + ":" + ociRef, nil
|
||||
}
|
||||
|
||||
func onUntarBlob(ctx context.Context, r io.Reader, store content.Store, name string, size int64) error {
|
||||
// name is like "blobs/sha256/deadbeef"
|
||||
split := strings.Split(name, "/")
|
||||
if len(split) != 3 {
|
||||
return errors.Errorf("unexpected name: %q", name)
|
||||
}
|
||||
algo := digest.Algorithm(split[1])
|
||||
if !algo.Available() {
|
||||
return errors.Errorf("unsupported algorithm: %s", algo)
|
||||
}
|
||||
dgst := digest.NewDigestFromHex(algo.String(), split[2])
|
||||
return content.WriteBlob(ctx, store, "unknown-"+dgst.String(), r, size, dgst)
|
||||
}
|
||||
|
||||
// GetChildrenDescriptors returns children blob descriptors for the following supported types:
|
||||
// - images.MediaTypeDockerSchema2Manifest, ocispec.MediaTypeImageManifest
|
||||
// - images.MediaTypeDockerSchema2ManifestList, ocispec.MediaTypeImageIndex
|
||||
func GetChildrenDescriptors(r io.Reader, desc ocispec.Descriptor) ([]ocispec.Descriptor, error) {
|
||||
switch desc.MediaType {
|
||||
case images.MediaTypeDockerSchema2Manifest, ocispec.MediaTypeImageManifest:
|
||||
var manifest ocispec.Manifest
|
||||
if err := json.NewDecoder(r).Decode(&manifest); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return append([]ocispec.Descriptor{manifest.Config}, manifest.Layers...), nil
|
||||
case images.MediaTypeDockerSchema2ManifestList, ocispec.MediaTypeImageIndex:
|
||||
var index ocispec.Index
|
||||
if err := json.NewDecoder(r).Decode(&index); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return index.Manifests, nil
|
||||
}
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func setGCRefContentLabels(ctx context.Context, store content.Store, desc ocispec.Descriptor) error {
|
||||
info, err := store.Info(ctx, desc.Digest)
|
||||
if err != nil {
|
||||
if errdefs.IsNotFound(err) {
|
||||
// when the archive is created from multi-arch image,
|
||||
// it may contain only blobs for a certain platform.
|
||||
// So ErrNotFound (on manifest list) is expected here.
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
ra, err := store.ReaderAt(ctx, desc.Digest)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer ra.Close()
|
||||
r := content.NewReader(ra)
|
||||
children, err := GetChildrenDescriptors(r, desc)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if info.Labels == nil {
|
||||
info.Labels = map[string]string{}
|
||||
}
|
||||
for i, child := range children {
|
||||
// Note: child blob is not guaranteed to be written to the content store. (multi-arch)
|
||||
info.Labels[fmt.Sprintf("containerd.io/gc.ref.content.%d", i)] = child.Digest.String()
|
||||
}
|
||||
if _, err := store.Update(ctx, info, "labels"); err != nil {
|
||||
return err
|
||||
}
|
||||
for _, child := range children {
|
||||
if err := setGCRefContentLabels(ctx, store, child); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
37
images/oci/importer_test.go
Normal file
37
images/oci/importer_test.go
Normal file
@@ -0,0 +1,37 @@
|
||||
package oci
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/opencontainers/go-digest"
|
||||
ocispec "github.com/opencontainers/image-spec/specs-go/v1"
|
||||
"github.com/stretchr/testify/assert"
|
||||
)
|
||||
|
||||
func TestNormalizeImageRef(t *testing.T) {
|
||||
imageBaseName := "foo/bar"
|
||||
for _, test := range []struct {
|
||||
input ocispec.Descriptor
|
||||
expect string
|
||||
}{
|
||||
{
|
||||
input: ocispec.Descriptor{
|
||||
Digest: digest.Digest("sha256:e22e93af8657d43d7f204b93d69604aeacf273f71d2586288cde312808c0ec77"),
|
||||
},
|
||||
expect: "foo/bar@sha256:e22e93af8657d43d7f204b93d69604aeacf273f71d2586288cde312808c0ec77",
|
||||
},
|
||||
{
|
||||
input: ocispec.Descriptor{
|
||||
Digest: digest.Digest("sha256:e22e93af8657d43d7f204b93d69604aeacf273f71d2586288cde312808c0ec77"),
|
||||
Annotations: map[string]string{
|
||||
ocispec.AnnotationRefName: "latest",
|
||||
},
|
||||
},
|
||||
expect: "foo/bar:latest", // no @digest for simplicity
|
||||
},
|
||||
} {
|
||||
normalized, err := normalizeImageRef(imageBaseName, test.input)
|
||||
assert.NoError(t, err)
|
||||
assert.Equal(t, test.expect, normalized)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user