stream

README

stream

The stream package contains an implementation of v1.Layer that supports streaming access, i.e. the layer contents are read once and not buffered.

Usage

package main

import (
	"os"

	"github.com/google/go-containerregistry/pkg/name"
	"github.com/google/go-containerregistry/pkg/v1/remote"
	"github.com/google/go-containerregistry/pkg/v1/stream"
)

// upload the contents of stdin as a layer to a local registry
func main() {
	repo, err := name.NewRepository("localhost:5000/stream")
	if err != nil {
		panic(err)
	}

	layer := stream.NewLayer(os.Stdin)

	if err := remote.WriteLayer(repo, layer); err != nil {
		panic(err)
	}
}

Structure

This implements the layer portion of an image upload. We launch a goroutine that is responsible for hashing the uncompressed contents to compute the DiffID, gzipping them to produce the Compressed contents, and hashing/counting the bytes to produce the Digest/Size. This goroutine writes to an io.PipeWriter, which blocks until Compressed reads the gzipped contents from the corresponding io.PipeReader.

Caveats

This assumes that you have an uncompressed layer (i.e. a tarball) and would like to compress it. Calling Uncompressed is always an error. Likewise, other methods are invalid until the contents of Compressed have been completely consumed and Closed.

Using a stream.Layer will likely not work without careful consideration. For example, in the mutate package, we defer computing the manifest and config file until they are actually called. This allows you to mutate.Append a streaming layer to an image without accidentally consuming it. Similarly, in remote.Write, if calling Digest on a layer fails, we attempt to upload the layer anyway, understanding that we may be dealing with a stream.Layer whose contents need to be uploaded before we can upload the config file.

Given the structure of how this is implemented, forgetting to Close a stream.Layer will leak a goroutine.

Documentation

Index

• Variables
• type Layer
  • func NewLayer(rc io.ReadCloser, opts ...LayerOption) *Layer
  • func (l *Layer) Compressed() (io.ReadCloser, error)
  • func (l *Layer) DiffID() (v1.Hash, error)
  • func (l *Layer) Digest() (v1.Hash, error)
  • func (l *Layer) MediaType() (types.MediaType, error)
  • func (l *Layer) Size() (int64, error)
  • func (l *Layer) Uncompressed() (io.ReadCloser, error)
• type LayerOption
  • func WithCompressionLevel(level int) LayerOption

Variables

var (
	// ErrNotComputed is returned when the requested value is not yet
	// computed because the stream has not been consumed yet.
	ErrNotComputed = errors.New("value not computed until stream is consumed")

	// ErrConsumed is returned by Compressed when the underlying stream has
	// already been consumed and closed.
	ErrConsumed = errors.New("stream was already consumed")
)

Types

type Layer

type Layer struct {
	// contains filtered or unexported fields
}

Layer is a streaming implementation of v1.Layer.

func NewLayer

func NewLayer(rc io.ReadCloser, opts ...LayerOption) *Layer

NewLayer creates a Layer from an io.ReadCloser.

func (*Layer) Compressed

func (l *Layer) Compressed() (io.ReadCloser, error)

Compressed implements v1.Layer.

func (*Layer) DiffID

func (l *Layer) DiffID() (v1.Hash, error)

DiffID implements v1.Layer.

func (*Layer) Digest

func (l *Layer) Digest() (v1.Hash, error)

Digest implements v1.Layer.

func (*Layer) MediaType

func (l *Layer) MediaType() (types.MediaType, error)

MediaType implements v1.Layer

func (*Layer) Size

func (l *Layer) Size() (int64, error)

Size implements v1.Layer.

func (*Layer) Uncompressed

func (l *Layer) Uncompressed() (io.ReadCloser, error)

Uncompressed implements v1.Layer.

type LayerOption

type LayerOption func(*Layer)

LayerOption applies options to layer

func WithCompressionLevel

func WithCompressionLevel(level int) LayerOption

WithCompressionLevel sets the gzip compression. See `gzip.NewWriterLevel` for possible values.