clearlinux/rkt
2
0
Fork 0
mirror of https://github.com/clearlinux/rkt.git synced 2026-09-06 22:01:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #90 from philips/fetch-from-disk

Browse Source 
Fetch from disk
This commit is contained in:
Brandon Philips
2014-11-28 10:06:41 -08:00
parent 6eed7af4d2 883c12ba34
commit f368965106
7 changed files with 168 additions and 132 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app-container/schema/types/hash.go
+8
View File
@@ -1,6 +1,7 @@
package types
import (
"crypto/sha256"
"encoding/json"
"errors"
"fmt"
@@ -69,3 +70,10 @@ func (h Hash) MarshalJSON() ([]byte, error) {
}
return json.Marshal(h.String())
}
func NewHashSHA256(b []byte) *Hash {
h := sha256.New()
h.Write(b)
nh, _ := NewHash(fmt.Sprintf("sha256-%x", h.Sum(nil)))
return nh
}
cas/cas.go
+105 -31
View File
@@ -1,30 +1,28 @@
package cas
import (
"bytes"
"crypto/sha256"
"fmt"
"io"
"path/filepath"
"github.com/coreos-inc/rkt/app-container/aci"
"github.com/coreos-inc/rkt/Godeps/_workspace/src/github.com/peterbourgon/diskv"
)
// TODO(philips): use a database for the secondary indexes like remoteType and
// appType. This is OK for now though.
const (
remoteType int64 = iota
objectType
downloadType
blobType int64 = iota
remoteType
tmpType
)
var otmap = [...]string{
"blob",
"remote",
"object",
"download",
}
type Blob interface {
Hash() string
Marshal() []byte
Unmarshal([]byte)
Type() int64
"tmp",
}
type Store struct {
@@ -46,6 +44,101 @@ func NewStore(base string) *Store {
return ds
}
func (ds Store) ReadStream(key string) (io.ReadCloser, error) {
return ds.stores[blobType].ReadStream(key, false)
}
func (ds Store) WriteStream(key string, r io.Reader) error {
return ds.stores[blobType].WriteStream(key, r, true)
}
func (ds Store) WriteACI(tmpKey string, orig io.Reader) (string, error) {
var b bytes.Buffer
// TODO(philips): use go routines to parallelize this pipeline and make
// the file type detection happen without a second stream
_, err := io.Copy(&b, orig)
if err != nil {
return "", err
}
err = ds.stores[tmpType].WriteStream(tmpKey, &b, true)
if err != nil {
return "", err
}
// Detect the filetype
rs, err := ds.stores[tmpType].ReadStream(tmpKey, false)
if err != nil {
return "", err
}
defer rs.Close()
typ, err := aci.DetectFileType(rs)
if err != nil {
return "", err
}
rs, err = ds.stores[tmpType].ReadStream(tmpKey, false)
if err != nil {
return "", err
}
defer rs.Close()
// Generate the hash of the decompressed tar
dr, err := decompress(rs, typ)
if err != nil {
return "", err
}
hash := sha256.New()
_, err = io.Copy(hash, dr)
if err != nil {
return "", err
}
// Store the decompressed tar
rs, err = ds.stores[tmpType].ReadStream(tmpKey, false)
if err != nil {
return "", err
}
defer rs.Close()
dr, err = decompress(rs, typ)
if err != nil {
return "", err
}
key := fmt.Sprintf("sha256-%x", hash.Sum(nil))
err = ds.stores[blobType].WriteStream(key, dr, true)
if err != nil {
return "", err
}
ds.stores[tmpType].Erase(tmpKey)
return key, nil
}
type Index interface {
Hash() string
Marshal() []byte
Unmarshal([]byte)
Type() int64
}
func (ds Store) WriteIndex(i Index) {
ds.stores[i.Type()].Write(i.Hash(), i.Marshal())
}
func (ds Store) ReadIndex(i Index) error {
buf, err := ds.stores[i.Type()].Read(i.Hash())
if err != nil {
return err
}
i.Unmarshal(buf)
return nil
}
func (ds Store) Dump(hex bool) {
for _, s := range ds.stores {
var keyCount int
@@ -67,22 +160,3 @@ func (ds Store) Dump(hex bool) {
fmt.Printf("%d total keys\n", keyCount)
}
}
func (ds Store) Store(b Blob) {
ds.stores[b.Type()].Write(b.Hash(), b.Marshal())
}
func (ds Store) ObjectStream(file string) (io.ReadCloser, error) {
return ds.stores[objectType].ReadStream(file, false)
}
func (ds Store) Get(b Blob) error {
buf, err := ds.stores[b.Type()].Read(b.Hash())
if err != nil {
return err
}
b.Unmarshal(buf)
return nil
}
cas/remote.go
+6 -89
View File
@@ -1,17 +1,11 @@
package cas
import (
"bytes"
"compress/bzip2"
"compress/gzip"
"crypto/sha256"
"encoding/json"
"fmt"
"io"
"net/http"
"os"
"github.com/coreos-inc/rkt/app-container/aci"
"github.com/coreos-inc/rkt/app-container/schema/types"
)
func NewRemote(name string, mirrors []string) *Remote {
@@ -26,7 +20,7 @@ type Remote struct {
Name string
Mirrors []string
ETag string
File string
Blob string
}
func (r Remote) Marshal() []byte {
@@ -42,41 +36,15 @@ func (r *Remote) Unmarshal(data []byte) {
}
func (r Remote) Hash() string {
return sha256sum(r.Name)
return types.NewHashSHA256([]byte(r.Name)).String()
}
func (r Remote) Type() int64 {
return remoteType
}
func decompress(rs io.Reader, typ aci.FileType) (io.Reader, error) {
var (
dr io.Reader
err error
)
switch typ {
case aci.TypeGzip:
dr, err = gzip.NewReader(rs)
if err != nil {
return nil, err
}
case aci.TypeBzip2:
dr = bzip2.NewReader(rs)
case aci.TypeXz:
dr = aci.XzReader(rs)
case aci.TypeUnknown:
fmt.Fprintf(os.Stderr, "error: unknown image filetype\n")
default:
// should never happen
panic("no type returned from DetectFileType?")
}
return dr, nil
}
// TODO: add locking
func (r Remote) Download(ds Store) (*Remote, error) {
var b bytes.Buffer
res, err := http.Get(r.Name)
if err != nil {
return nil, err
@@ -88,64 +56,13 @@ func (r Remote) Download(ds Store) (*Remote, error) {
return nil, fmt.Errorf("bad HTTP status code: %d", res.StatusCode)
}
// TODO(philips): use go routines to parallelize this pipeline and make
// the file type detection happen without a second stream
_, err = io.Copy(&b, res.Body)
if err != nil {
return nil, err
}
err = ds.stores[downloadType].WriteStream(r.Hash(), &b, true)
key, err := ds.WriteACI(r.Hash(), res.Body)
if err != nil {
return nil, err
}
// Detect the filetype
rs, err := ds.stores[downloadType].ReadStream(r.Hash(), false)
if err != nil {
return nil, err
}
defer rs.Close()
typ, err := aci.DetectFileType(rs)
if err != nil {
return nil, err
}
rs, err = ds.stores[downloadType].ReadStream(r.Hash(), false)
if err != nil {
return nil, err
}
defer rs.Close()
// Generate the hash of the decompressed tar
dr, err := decompress(rs, typ)
if err != nil {
return nil, err
}
hash := sha256.New()
_, err = io.Copy(hash, dr)
if err != nil {
return nil, err
}
// Store the decompressed tar
rs, err = ds.stores[downloadType].ReadStream(r.Hash(), false)
if err != nil {
return nil, err
}
defer rs.Close()
dr, err = decompress(rs, typ)
if err != nil {
return nil, err
}
key := fmt.Sprintf("sha256-%x", hash.Sum(nil))
err = ds.stores[objectType].WriteStream(key, dr, true)
if err != nil {
return nil, err
}
ds.stores[downloadType].Erase(r.Hash())
r.File = key
ds.stores[remoteType].Write(r.Hash(), r.Marshal())
r.Blob = key
ds.WriteIndex(&r)
return &r, nil
}
cas/utils.go
+28 -8
View File
@@ -1,11 +1,14 @@
package cas
import (
"crypto/sha256"
"fmt"
"compress/bzip2"
"compress/gzip"
"errors"
"io"
"net/url"
"strings"
"github.com/coreos-inc/rkt/app-container/aci"
)
// copy the default of git which is a two byte prefix. We will likely want to
@@ -19,13 +22,30 @@ func blockTransform(s string) []string {
return pathSlice
}
func sha256sum(s string) string {
h := sha256.New()
io.WriteString(h, s)
return fmt.Sprintf("sha256-%x", h.Sum(nil))
}
func parseAlways(s string) *url.URL {
u, _ := url.Parse(s)
return u
}
func decompress(rs io.Reader, typ aci.FileType) (io.Reader, error) {
var (
dr io.Reader
err error
)
switch typ {
case aci.TypeGzip:
dr, err = gzip.NewReader(rs)
if err != nil {
return nil, err
}
case aci.TypeBzip2:
dr = bzip2.NewReader(rs)
case aci.TypeXz:
dr = aci.XzReader(rs)
case aci.TypeUnknown:
return nil, errors.New("error: unknown image filetype")
default:
return nil, errors.New("no type returned from DetectFileType?")
}
return dr, nil
}
cmd/fetch.go
+3 -3
View File
@@ -23,14 +23,14 @@ var (
func fetchURL(img string, ds *cas.Store) (string, error) {
rem := cas.NewRemote(img, []string{})
err := ds.Get(rem)
if err != nil && rem.File == "" {
err := ds.ReadIndex(rem)
if err != nil && rem.Blob == "" {
rem, err = rem.Download(*ds)
if err != nil {
return "", fmt.Errorf("downloading: %v\n", err)
}
}
return rem.File, nil
return rem.Blob, nil
}
func runFetch(args []string) (exit int) {
cmd/run.go
+17
View File
@@ -23,6 +23,8 @@ var (
Name: "run",
Summary: "Run image(s) in an application container in rocket",
Usage: "[--volume LABEL:SOURCE] IMAGE...",
Description: `IMAGE should be a string referencing an image; either a hash, local file on disk, or URL.
They will be checked in that order and the first match will be used.`,
Run: runRun,
}
)
@@ -43,6 +45,21 @@ func findImages(args []string, ds *cas.Store) (out []string, err error) {
if err == nil {
continue
}
// import the local file if it exists
file, err := os.Open(img)
if err == nil {
hash := types.NewHashSHA256([]byte(img)).String()
key, err := ds.WriteACI(hash, file)
file.Close()
if err != nil {
return nil, fmt.Errorf("%s: %v", img, err)
}
out[i] = key
continue
}
// download if it is a URL
u, err := url.Parse(img)
if err != nil {
return nil, fmt.Errorf("%s: not a valid URL or hash", img)
stage0/run.go
+1 -1
View File
@@ -264,7 +264,7 @@ func unpackBuiltinRootfs(dir string) error {
func setupImage(cfg Config, img string, h types.Hash, dir string) (*schema.AppManifest, error) {
log.Println("Loading image", img)
rs, err := cfg.Store.ObjectStream(img)
rs, err := cfg.Store.ReadStream(img)
if err != nil {
return nil, err
}