containerd/content/writer.go
Stephen J Day f9cd9be61a
dist: expand functionality of the dist tool
With this change, we add the following commands to the dist tool:

- `ingest`: verify and accept content into storage
- `active`: display active ingest processes
- `list`: list content in storage
- `path`: provide a path to a blob by digest
- `delete`: remove a piece of content from storage

We demonstrate the utility with the following shell pipeline:

```
$ ./dist fetch docker.io/library/redis latest mediatype:application/vnd.docker.distribution.manifest.v2+json | \
    jq -r '.layers[] | "./dist fetch docker.io/library/redis "+.digest + "| ./dist ingest --expected-digest "+.digest+" --expected-size "+(.size | tostring) +" docker.io/library/redis@"+.digest' | xargs -I{} -P10 -n1 sh -c "{}"
```

The above fetches a manifest, pipes it to jq, which assembles a shell
pipeline to ingest each layer into the content store. Because the
transactions are keyed by their digest, concurrent downloads and
downloads of repeated content are ignored. Each process is then executed
parallel using xargs.

Put shortly, this is a parallel layer download.

In a separate shell session, could monitor the active downloads with the
following:

```
$ watch -n0.2 ./dist active
```

For now, the content is downloaded into `.content` in the current
working directory. To watch the contents of this directory, you can use
the following:

```
$ watch -n0.2 tree .content
```

This will help to understand what is going on internally.

To get access to the layers, you can use the path command:

```
$./dist path sha256:010c454d55e53059beaba4044116ea4636f8dd8181e975d893931c7e7204fffa
sha256:010c454d55e53059beaba4044116ea4636f8dd8181e975d893931c7e7204fffa /home/sjd/go/src/github.com/docker/containerd/.content/blobs/sha256/010c454d55e53059beaba4044116ea4636f8dd8181e975d893931c7e7204fffa
```

When you are done, you can clear out the content with the classic xargs
pipeline:

```
$ ./dist list -q | xargs ./dist delete
```

Note that this is mostly a POC. Things like failed downloads and
abandoned download cleanup aren't quite handled. We'll probably make
adjustments around how content store transactions are handled to address
this.

From here, we'll build out full image pull and create tooling to get
runtime bundles from the fetched content.

Signed-off-by: Stephen J Day <stephen.day@docker.com>
2017-01-27 10:29:10 -08:00

132 lines
3.2 KiB
Go

package content
import (
"log"
"os"
"path/filepath"
"github.com/nightlyone/lockfile"
"github.com/opencontainers/go-digest"
"github.com/pkg/errors"
)
// Writer represents a write transaction against the blob store.
type Writer struct {
cs *Store
fp *os.File // opened data file
lock lockfile.Lockfile
path string // path to writer dir
ref string // ref key
offset int64
digester digest.Digester
}
func (cw *Writer) Ref() string {
return cw.ref
}
// Size returns the current size written.
//
// Cannot be called concurrently with `Write`. If you need need concurrent
// status, query it with `Store.Stat`.
func (cw *Writer) Size() int64 {
return cw.offset
}
// Digest returns the current digest of the content, up to the current write.
//
// Cannot be called concurrently with `Write`.
func (cw *Writer) Digest() digest.Digest {
return cw.digester.Digest()
}
// Write p to the transaction.
//
// Note that writes are unbuffered to the backing file. When writing, it is
// recommended to wrap in a bufio.Writer or, preferably, use io.CopyBuffer.
func (cw *Writer) Write(p []byte) (n int, err error) {
n, err = cw.fp.Write(p)
cw.digester.Hash().Write(p[:n])
cw.offset += int64(len(p))
return n, err
}
func (cw *Writer) Commit(size int64, expected digest.Digest) error {
if err := cw.fp.Sync(); err != nil {
return errors.Wrap(err, "sync failed")
}
fi, err := cw.fp.Stat()
if err != nil {
return errors.Wrap(err, "stat on ingest file failed")
}
// change to readonly, more important for read, but provides _some_
// protection from this point on. We use the existing perms with a mask
// only allowing reads honoring the umask on creation.
//
// This removes write and exec, only allowing read per the creation umask.
if err := cw.fp.Chmod((fi.Mode() & os.ModePerm) &^ 0333); err != nil {
return errors.Wrap(err, "failed to change ingest file permissions")
}
if size > 0 && size != fi.Size() {
return errors.Errorf("failed size validation: %v != %v", fi.Size(), size)
}
if err := cw.fp.Close(); err != nil {
return errors.Wrap(err, "failed closing ingest")
}
dgst := cw.digester.Digest()
if expected != "" && expected != dgst {
return errors.Errorf("unexpected digest: %v != %v", dgst, expected)
}
var (
ingest = filepath.Join(cw.path, "data")
target = cw.cs.blobPath(dgst)
)
// make sure parent directories of blob exist
if err := os.MkdirAll(filepath.Dir(target), 0755); err != nil {
return err
}
// clean up!!
defer os.RemoveAll(cw.path)
if err := os.Rename(ingest, target); err != nil {
if os.IsExist(err) {
// collision with the target file!
return nil
}
return err
}
unlock(cw.lock)
cw.fp = nil
return nil
}
// Close the writer, flushing any unwritten data and leaving the progress in
// tact.
//
// If one needs to resume the transaction, a new writer can be obtained from
// `ContentStore.Resume` using the same key. The write can then be continued
// from it was left off.
//
// To abandon a transaction completely, first call close then `Store.Remove` to
// clean up the associated resources.
func (cw *Writer) Close() (err error) {
if err := unlock(cw.lock); err != nil {
log.Printf("unlock failed: %v", err)
}
if cw.fp != nil {
cw.fp.Sync()
return cw.fp.Close()
}
return nil
}