mirror of
https://github.com/moby/moby.git
synced 2022-11-09 12:21:53 -05:00
Merge pull request #1268 from dotcloud/refactor_checksum
Refactor checksum
This commit is contained in:
Victor Vieux
commit
bcd6ca3685
6 changed files with 298 additions and 272 deletions
|
|
@ -18,7 +18,7 @@ const CONFIGFILE = ".dockercfg"
|
|||
// Only used for user auth + account creation
|
||||
const INDEXSERVER = "https://index.docker.io/v1/"
|
||||
|
||||
//const INDEXSERVER = "http://indexstaging-docker.dotcloud.com/"
|
||||
//const INDEXSERVER = "https://indexstaging-docker.dotcloud.com/v1/"
|
||||
|
||||
var (
|
||||
ErrConfigFileMissing = errors.New("The Auth config file is missing")
|
||||
|
|
|
|||
67
graph.go
67
graph.go
|
|
@ -1,9 +1,7 @@
|
|||
package docker
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"github.com/dotcloud/docker/registry"
|
||||
"github.com/dotcloud/docker/utils"
|
||||
"io"
|
||||
"io/ioutil"
|
||||
|
|
@ -11,17 +9,13 @@ import (
|
|||
"path"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// A Graph is a store for versioned filesystem images and the relationship between them.
|
||||
type Graph struct {
|
||||
Root string
|
||||
idIndex *utils.TruncIndex
|
||||
checksumLock map[string]*sync.Mutex
|
||||
lockSumFile *sync.Mutex
|
||||
lockSumMap *sync.Mutex
|
||||
Root string
|
||||
idIndex *utils.TruncIndex
|
||||
}
|
||||
|
||||
// NewGraph instantiates a new graph at the given root path in the filesystem.
|
||||
|
|
@ -36,11 +30,8 @@ func NewGraph(root string) (*Graph, error) {
|
|||
return nil, err
|
||||
}
|
||||
graph := &Graph{
|
||||
Root: abspath,
|
||||
idIndex: utils.NewTruncIndex(),
|
||||
checksumLock: make(map[string]*sync.Mutex),
|
||||
lockSumFile: &sync.Mutex{},
|
||||
lockSumMap: &sync.Mutex{},
|
||||
Root: abspath,
|
||||
idIndex: utils.NewTruncIndex(),
|
||||
}
|
||||
if err := graph.restore(); err != nil {
|
||||
return nil, err
|
||||
|
|
@ -99,11 +90,6 @@ func (graph *Graph) Get(name string) (*Image, error) {
|
|||
return nil, err
|
||||
}
|
||||
}
|
||||
graph.lockSumMap.Lock()
|
||||
defer graph.lockSumMap.Unlock()
|
||||
if _, exists := graph.checksumLock[img.ID]; !exists {
|
||||
graph.checksumLock[img.ID] = &sync.Mutex{}
|
||||
}
|
||||
return img, nil
|
||||
}
|
||||
|
||||
|
|
@ -123,16 +109,15 @@ func (graph *Graph) Create(layerData Archive, container *Container, comment, aut
|
|||
img.Container = container.ID
|
||||
img.ContainerConfig = *container.Config
|
||||
}
|
||||
if err := graph.Register(layerData, layerData != nil, img); err != nil {
|
||||
if err := graph.Register(nil, layerData, img); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
go img.Checksum()
|
||||
return img, nil
|
||||
}
|
||||
|
||||
// Register imports a pre-existing image into the graph.
|
||||
// FIXME: pass img as first argument
|
||||
func (graph *Graph) Register(layerData Archive, store bool, img *Image) error {
|
||||
func (graph *Graph) Register(jsonData []byte, layerData Archive, img *Image) error {
|
||||
if err := ValidateID(img.ID); err != nil {
|
||||
return err
|
||||
}
|
||||
|
|
@ -145,7 +130,7 @@ func (graph *Graph) Register(layerData Archive, store bool, img *Image) error {
|
|||
if err != nil {
|
||||
return fmt.Errorf("Mktemp failed: %s", err)
|
||||
}
|
||||
if err := StoreImage(img, layerData, tmp, store); err != nil {
|
||||
if err := StoreImage(img, jsonData, layerData, tmp); err != nil {
|
||||
return err
|
||||
}
|
||||
// Commit
|
||||
|
|
@ -154,7 +139,6 @@ func (graph *Graph) Register(layerData Archive, store bool, img *Image) error {
|
|||
}
|
||||
img.graph = graph
|
||||
graph.idIndex.Add(img.ID)
|
||||
graph.checksumLock[img.ID] = &sync.Mutex{}
|
||||
return nil
|
||||
}
|
||||
|
||||
|
|
@ -311,40 +295,3 @@ func (graph *Graph) Heads() (map[string]*Image, error) {
|
|||
func (graph *Graph) imageRoot(id string) string {
|
||||
return path.Join(graph.Root, id)
|
||||
}
|
||||
|
||||
func (graph *Graph) getStoredChecksums() (map[string]string, error) {
|
||||
checksums := make(map[string]string)
|
||||
// FIXME: Store the checksum in memory
|
||||
|
||||
if checksumDict, err := ioutil.ReadFile(path.Join(graph.Root, "checksums")); err == nil {
|
||||
if err := json.Unmarshal(checksumDict, &checksums); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
return checksums, nil
|
||||
}
|
||||
|
||||
func (graph *Graph) storeChecksums(checksums map[string]string) error {
|
||||
checksumJSON, err := json.Marshal(checksums)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := ioutil.WriteFile(path.Join(graph.Root, "checksums"), checksumJSON, 0600); err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (graph *Graph) UpdateChecksums(newChecksums map[string]*registry.ImgData) error {
|
||||
graph.lockSumFile.Lock()
|
||||
defer graph.lockSumFile.Unlock()
|
||||
|
||||
localChecksums, err := graph.getStoredChecksums()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for id, elem := range newChecksums {
|
||||
localChecksums[id] = elem.Checksum
|
||||
}
|
||||
return graph.storeChecksums(localChecksums)
|
||||
}
|
||||
|
|
|
|||
142
image.go
142
image.go
|
|
@ -2,7 +2,6 @@ package docker
|
|||
|
||||
import (
|
||||
"crypto/rand"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
|
|
@ -14,6 +13,7 @@ import (
|
|||
"os/exec"
|
||||
"path"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
|
@ -47,6 +47,19 @@ func LoadImage(root string) (*Image, error) {
|
|||
if err := ValidateID(img.ID); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if buf, err := ioutil.ReadFile(path.Join(root, "layersize")); err != nil {
|
||||
if !os.IsNotExist(err) {
|
||||
return nil, err
|
||||
}
|
||||
} else {
|
||||
if size, err := strconv.Atoi(string(buf)); err != nil {
|
||||
return nil, err
|
||||
} else {
|
||||
img.Size = int64(size)
|
||||
}
|
||||
}
|
||||
|
||||
// Check that the filesystem layer exists
|
||||
if stat, err := os.Stat(layerPath(root)); err != nil {
|
||||
if os.IsNotExist(err) {
|
||||
|
|
@ -59,7 +72,7 @@ func LoadImage(root string) (*Image, error) {
|
|||
return img, nil
|
||||
}
|
||||
|
||||
func StoreImage(img *Image, layerData Archive, root string, store bool) error {
|
||||
func StoreImage(img *Image, jsonData []byte, layerData Archive, root string) error {
|
||||
// Check that root doesn't already exist
|
||||
if _, err := os.Stat(root); err == nil {
|
||||
return fmt.Errorf("Image %s already exists", img.ID)
|
||||
|
|
@ -72,26 +85,6 @@ func StoreImage(img *Image, layerData Archive, root string, store bool) error {
|
|||
return err
|
||||
}
|
||||
|
||||
if store {
|
||||
layerArchive := layerArchivePath(root)
|
||||
file, err := os.OpenFile(layerArchive, os.O_WRONLY|os.O_CREATE, 0600)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
// FIXME: Retrieve the image layer size from here?
|
||||
if _, err := io.Copy(file, layerData); err != nil {
|
||||
return err
|
||||
}
|
||||
// FIXME: Don't close/open, read/write instead of Copy
|
||||
file.Close()
|
||||
|
||||
file, err = os.Open(layerArchive)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer file.Close()
|
||||
layerData = file
|
||||
}
|
||||
// If layerData is not nil, unpack it into the new layer
|
||||
if layerData != nil {
|
||||
start := time.Now()
|
||||
|
|
@ -102,25 +95,36 @@ func StoreImage(img *Image, layerData Archive, root string, store bool) error {
|
|||
utils.Debugf("Untar time: %vs\n", time.Now().Sub(start).Seconds())
|
||||
}
|
||||
|
||||
// If raw json is provided, then use it
|
||||
if jsonData != nil {
|
||||
return ioutil.WriteFile(jsonPath(root), jsonData, 0600)
|
||||
} else { // Otherwise, unmarshal the image
|
||||
jsonData, err := json.Marshal(img)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := ioutil.WriteFile(jsonPath(root), jsonData, 0600); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
return StoreSize(img, root)
|
||||
}
|
||||
|
||||
func StoreSize(img *Image, root string) error {
|
||||
layer := layerPath(root)
|
||||
|
||||
var totalSize int64 = 0
|
||||
filepath.Walk(layer, func(path string, fileInfo os.FileInfo, err error) error {
|
||||
img.Size += fileInfo.Size()
|
||||
totalSize += fileInfo.Size()
|
||||
return nil
|
||||
})
|
||||
img.Size = totalSize
|
||||
|
||||
// Store the json ball
|
||||
jsonData, err := json.Marshal(img)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := ioutil.WriteFile(jsonPath(root), jsonData, 0600); err != nil {
|
||||
return err
|
||||
if err := ioutil.WriteFile(path.Join(root, "layersize"), []byte(strconv.Itoa(int(totalSize))), 0600); err != nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
|
|
@ -128,10 +132,6 @@ func layerPath(root string) string {
|
|||
return path.Join(root, "layer")
|
||||
}
|
||||
|
||||
func layerArchivePath(root string) string {
|
||||
return path.Join(root, "layer.tar.xz")
|
||||
}
|
||||
|
||||
func jsonPath(root string) string {
|
||||
return path.Join(root, "json")
|
||||
}
|
||||
|
|
@ -308,80 +308,6 @@ func (img *Image) layer() (string, error) {
|
|||
return layerPath(root), nil
|
||||
}
|
||||
|
||||
func (img *Image) Checksum() (string, error) {
|
||||
img.graph.checksumLock[img.ID].Lock()
|
||||
defer img.graph.checksumLock[img.ID].Unlock()
|
||||
|
||||
root, err := img.root()
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
checksums, err := img.graph.getStoredChecksums()
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
if checksum, ok := checksums[img.ID]; ok {
|
||||
return checksum, nil
|
||||
}
|
||||
|
||||
layer, err := img.layer()
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
jsonData, err := ioutil.ReadFile(jsonPath(root))
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
var layerData io.Reader
|
||||
|
||||
if file, err := os.Open(layerArchivePath(root)); err != nil {
|
||||
if os.IsNotExist(err) {
|
||||
layerData, err = Tar(layer, Xz)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
} else {
|
||||
return "", err
|
||||
}
|
||||
} else {
|
||||
defer file.Close()
|
||||
layerData = file
|
||||
}
|
||||
|
||||
h := sha256.New()
|
||||
if _, err := h.Write(jsonData); err != nil {
|
||||
return "", err
|
||||
}
|
||||
if _, err := h.Write([]byte("\n")); err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
if _, err := io.Copy(h, layerData); err != nil {
|
||||
return "", err
|
||||
}
|
||||
hash := "sha256:" + hex.EncodeToString(h.Sum(nil))
|
||||
|
||||
// Reload the json file to make sure not to overwrite faster sums
|
||||
img.graph.lockSumFile.Lock()
|
||||
defer img.graph.lockSumFile.Unlock()
|
||||
|
||||
checksums, err = img.graph.getStoredChecksums()
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
checksums[img.ID] = hash
|
||||
|
||||
// Dump the checksums to disc
|
||||
if err := img.graph.storeChecksums(checksums); err != nil {
|
||||
return hash, err
|
||||
}
|
||||
|
||||
return hash, nil
|
||||
}
|
||||
|
||||
func (img *Image) getParentsSize(size int64) int64 {
|
||||
parentImage, err := img.GetParent()
|
||||
if err != nil || parentImage == nil {
|
||||
|
|
|
|||
|
|
@ -17,8 +17,10 @@ import (
|
|||
"strings"
|
||||
)
|
||||
|
||||
var ErrAlreadyExists = errors.New("Image already exists")
|
||||
var ErrInvalidRepositoryName = errors.New("Invalid repository name (ex: \"registry.domain.tld/myrepos\")")
|
||||
var (
|
||||
ErrAlreadyExists = errors.New("Image already exists")
|
||||
ErrInvalidRepositoryName = errors.New("Invalid repository name (ex: \"registry.domain.tld/myrepos\")")
|
||||
)
|
||||
|
||||
func pingRegistryEndpoint(endpoint string) error {
|
||||
if endpoint == auth.IndexServerAddress() {
|
||||
|
|
@ -266,8 +268,11 @@ func (r *Registry) GetRemoteTags(registries []string, repository string, token [
|
|||
}
|
||||
|
||||
func (r *Registry) GetRepositoryData(indexEp, remote string) (*RepositoryData, error) {
|
||||
|
||||
repositoryTarget := fmt.Sprintf("%srepositories/%s/images", indexEp, remote)
|
||||
|
||||
utils.Debugf("[registry] Calling GET %s", repositoryTarget)
|
||||
|
||||
req, err := r.opaqueRequest("GET", repositoryTarget, nil)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
|
|
@ -330,19 +335,54 @@ func (r *Registry) GetRepositoryData(indexEp, remote string) (*RepositoryData, e
|
|||
}, nil
|
||||
}
|
||||
|
||||
func (r *Registry) PushImageChecksumRegistry(imgData *ImgData, registry string, token []string) error {
|
||||
|
||||
utils.Debugf("[registry] Calling PUT %s", registry+"images/"+imgData.ID+"/checksum")
|
||||
|
||||
req, err := http.NewRequest("PUT", registry+"images/"+imgData.ID+"/checksum", nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
req.Header.Set("Authorization", "Token "+strings.Join(token, ","))
|
||||
req.Header.Set("X-Docker-Checksum", imgData.Checksum)
|
||||
|
||||
res, err := doWithCookies(r.client, req)
|
||||
if err != nil {
|
||||
return fmt.Errorf("Failed to upload metadata: %s", err)
|
||||
}
|
||||
defer res.Body.Close()
|
||||
if len(res.Cookies()) > 0 {
|
||||
r.client.Jar.SetCookies(req.URL, res.Cookies())
|
||||
}
|
||||
if res.StatusCode != 200 {
|
||||
errBody, err := ioutil.ReadAll(res.Body)
|
||||
if err != nil {
|
||||
return fmt.Errorf("HTTP code %d while uploading metadata and error when trying to parse response body: %s", res.StatusCode, err)
|
||||
}
|
||||
var jsonBody map[string]string
|
||||
if err := json.Unmarshal(errBody, &jsonBody); err != nil {
|
||||
errBody = []byte(err.Error())
|
||||
} else if jsonBody["error"] == "Image already exists" {
|
||||
return ErrAlreadyExists
|
||||
}
|
||||
return fmt.Errorf("HTTP code %d while uploading metadata: %s", res.StatusCode, errBody)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Push a local image to the registry
|
||||
func (r *Registry) PushImageJSONRegistry(imgData *ImgData, jsonRaw []byte, registry string, token []string) error {
|
||||
// FIXME: try json with UTF8
|
||||
|
||||
utils.Debugf("[registry] Calling PUT %s", registry+"images/"+imgData.ID+"/json")
|
||||
|
||||
req, err := http.NewRequest("PUT", registry+"images/"+imgData.ID+"/json", bytes.NewReader(jsonRaw))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
req.Header.Add("Content-type", "application/json")
|
||||
req.Header.Set("Authorization", "Token "+strings.Join(token, ","))
|
||||
req.Header.Set("X-Docker-Checksum", imgData.Checksum)
|
||||
r.setUserAgent(req)
|
||||
|
||||
utils.Debugf("Setting checksum for %s: %s", imgData.ID, imgData.Checksum)
|
||||
res, err := doWithCookies(r.client, req)
|
||||
if err != nil {
|
||||
return fmt.Errorf("Failed to upload metadata: %s", err)
|
||||
|
|
@ -364,10 +404,15 @@ func (r *Registry) PushImageJSONRegistry(imgData *ImgData, jsonRaw []byte, regis
|
|||
return nil
|
||||
}
|
||||
|
||||
func (r *Registry) PushImageLayerRegistry(imgID string, layer io.Reader, registry string, token []string) error {
|
||||
req, err := http.NewRequest("PUT", registry+"images/"+imgID+"/layer", layer)
|
||||
func (r *Registry) PushImageLayerRegistry(imgID string, layer io.Reader, registry string, token []string, jsonRaw []byte) (checksum string, err error) {
|
||||
|
||||
utils.Debugf("[registry] Calling PUT %s", registry+"images/"+imgID+"/layer")
|
||||
|
||||
tarsumLayer := &utils.TarSum{Reader: layer}
|
||||
|
||||
req, err := http.NewRequest("PUT", registry+"images/"+imgID+"/layer", tarsumLayer)
|
||||
if err != nil {
|
||||
return err
|
||||
return "", err
|
||||
}
|
||||
req.ContentLength = -1
|
||||
req.TransferEncoding = []string{"chunked"}
|
||||
|
|
@ -375,18 +420,18 @@ func (r *Registry) PushImageLayerRegistry(imgID string, layer io.Reader, registr
|
|||
r.setUserAgent(req)
|
||||
res, err := doWithCookies(r.client, req)
|
||||
if err != nil {
|
||||
return fmt.Errorf("Failed to upload layer: %s", err)
|
||||
return "", fmt.Errorf("Failed to upload layer: %s", err)
|
||||
}
|
||||
defer res.Body.Close()
|
||||
|
||||
if res.StatusCode != 200 {
|
||||
errBody, err := ioutil.ReadAll(res.Body)
|
||||
if err != nil {
|
||||
return fmt.Errorf("HTTP code %d while uploading metadata and error when trying to parse response body: %s", res.StatusCode, err)
|
||||
return "", fmt.Errorf("HTTP code %d while uploading metadata and error when trying to parse response body: %s", res.StatusCode, err)
|
||||
}
|
||||
return fmt.Errorf("Received HTTP code %d while uploading layer: %s", res.StatusCode, errBody)
|
||||
return "", fmt.Errorf("Received HTTP code %d while uploading layer: %s", res.StatusCode, errBody)
|
||||
}
|
||||
return nil
|
||||
return tarsumLayer.Sum(jsonRaw), nil
|
||||
}
|
||||
|
||||
func (r *Registry) opaqueRequest(method, urlStr string, body io.Reader) (*http.Request, error) {
|
||||
|
|
@ -424,7 +469,19 @@ func (r *Registry) PushRegistryTag(remote, revision, tag, registry string, token
|
|||
}
|
||||
|
||||
func (r *Registry) PushImageJSONIndex(indexEp, remote string, imgList []*ImgData, validate bool, regs []string) (*RepositoryData, error) {
|
||||
imgListJSON, err := json.Marshal(imgList)
|
||||
cleanImgList := []*ImgData{}
|
||||
|
||||
if validate {
|
||||
for _, elem := range imgList {
|
||||
if elem.Checksum != "" {
|
||||
cleanImgList = append(cleanImgList, elem)
|
||||
}
|
||||
}
|
||||
} else {
|
||||
cleanImgList = imgList
|
||||
}
|
||||
|
||||
imgListJSON, err := json.Marshal(cleanImgList)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
|
@ -434,7 +491,7 @@ func (r *Registry) PushImageJSONIndex(indexEp, remote string, imgList []*ImgData
|
|||
}
|
||||
|
||||
u := fmt.Sprintf("%srepositories/%s/%s", indexEp, remote, suffix)
|
||||
utils.Debugf("PUT %s", u)
|
||||
utils.Debugf("[registry] PUT %s", u)
|
||||
utils.Debugf("Image list pushed to index:\n%s\n", imgListJSON)
|
||||
req, err := r.opaqueRequest("PUT", u, bytes.NewReader(imgListJSON))
|
||||
if err != nil {
|
||||
|
|
|
|||
116
server.go
116
server.go
|
|
@ -438,7 +438,7 @@ func (srv *Server) pullImage(r *registry.Registry, out io.Writer, imgID, endpoin
|
|||
return err
|
||||
}
|
||||
defer layer.Close()
|
||||
if err := srv.runtime.graph.Register(utils.ProgressReader(layer, imgSize, out, sf.FormatProgress("Downloading", "%8v/%v (%v)"), sf), false, img); err != nil {
|
||||
if err := srv.runtime.graph.Register(imgJSON, utils.ProgressReader(layer, imgSize, out, sf.FormatProgress("Downloading", "%8v/%v (%v)"), sf), img); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
|
@ -454,12 +454,6 @@ func (srv *Server) pullRepository(r *registry.Registry, out io.Writer, localName
|
|||
return err
|
||||
}
|
||||
|
||||
utils.Debugf("Updating checksums")
|
||||
// Reload the json file to make sure not to overwrite faster sums
|
||||
if err := srv.runtime.graph.UpdateChecksums(repoData.ImgList); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
utils.Debugf("Retrieving the tag list")
|
||||
tagsList, err := r.GetRemoteTags(repoData.Endpoints, remoteName, repoData.Tokens)
|
||||
if err != nil {
|
||||
|
|
@ -597,41 +591,6 @@ func (srv *Server) ImagePull(localName string, tag string, out io.Writer, sf *ut
|
|||
return nil
|
||||
}
|
||||
|
||||
// Retrieve the checksum of an image
|
||||
// Priority:
|
||||
// - Check on the stored checksums
|
||||
// - Check if the archive exists, if it does not, ask the registry
|
||||
// - If the archive does exists, process the checksum from it
|
||||
// - If the archive does not exists and not found on registry, process checksum from layer
|
||||
func (srv *Server) getChecksum(imageID string) (string, error) {
|
||||
// FIXME: Use in-memory map instead of reading the file each time
|
||||
if sums, err := srv.runtime.graph.getStoredChecksums(); err != nil {
|
||||
return "", err
|
||||
} else if checksum, exists := sums[imageID]; exists {
|
||||
return checksum, nil
|
||||
}
|
||||
|
||||
img, err := srv.runtime.graph.Get(imageID)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
if _, err := os.Stat(layerArchivePath(srv.runtime.graph.imageRoot(imageID))); err != nil {
|
||||
if os.IsNotExist(err) {
|
||||
// TODO: Ask the registry for the checksum
|
||||
// As the archive is not there, it is supposed to come from a pull.
|
||||
} else {
|
||||
return "", err
|
||||
}
|
||||
}
|
||||
|
||||
checksum, err := img.Checksum()
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
return checksum, nil
|
||||
}
|
||||
|
||||
// Retrieve the all the images to be uploaded in the correct order
|
||||
// Note: we can't use a map as it is not ordered
|
||||
func (srv *Server) getImageList(localRepo map[string]string) ([]*registry.ImgData, error) {
|
||||
|
|
@ -648,14 +607,10 @@ func (srv *Server) getImageList(localRepo map[string]string) ([]*registry.ImgDat
|
|||
return nil
|
||||
}
|
||||
imageSet[img.ID] = struct{}{}
|
||||
checksum, err := srv.getChecksum(img.ID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
imgList = append([]*registry.ImgData{{
|
||||
ID: img.ID,
|
||||
Checksum: checksum,
|
||||
Tag: tag,
|
||||
ID: img.ID,
|
||||
Tag: tag,
|
||||
}}, imgList...)
|
||||
return nil
|
||||
})
|
||||
|
|
@ -665,7 +620,7 @@ func (srv *Server) getImageList(localRepo map[string]string) ([]*registry.ImgDat
|
|||
|
||||
func (srv *Server) pushRepository(r *registry.Registry, out io.Writer, localName, remoteName string, localRepo map[string]string, indexEp string, sf *utils.StreamFormatter) error {
|
||||
out = utils.NewWriteFlusher(out)
|
||||
out.Write(sf.FormatStatus("Processing checksums"))
|
||||
|
||||
imgList, err := srv.getImageList(localRepo)
|
||||
if err != nil {
|
||||
return err
|
||||
|
|
@ -689,9 +644,11 @@ func (srv *Server) pushRepository(r *registry.Registry, out io.Writer, localName
|
|||
out.Write(sf.FormatStatus("Image %s already pushed, skipping", elem.ID))
|
||||
continue
|
||||
}
|
||||
if err := srv.pushImage(r, out, remoteName, elem.ID, ep, repoData.Tokens, sf); err != nil {
|
||||
if checksum, err := srv.pushImage(r, out, remoteName, elem.ID, ep, repoData.Tokens, sf); err != nil {
|
||||
// FIXME: Continue on error?
|
||||
return err
|
||||
} else {
|
||||
elem.Checksum = checksum
|
||||
}
|
||||
out.Write(sf.FormatStatus("Pushing tags for rev [%s] on {%s}", elem.ID, ep+"repositories/"+remoteName+"/tags/"+elem.Tag))
|
||||
if err := r.PushRegistryTag(remoteName, elem.ID, elem.Tag, ep, repoData.Tokens); err != nil {
|
||||
|
|
@ -707,64 +664,45 @@ func (srv *Server) pushRepository(r *registry.Registry, out io.Writer, localName
|
|||
return nil
|
||||
}
|
||||
|
||||
func (srv *Server) pushImage(r *registry.Registry, out io.Writer, remote, imgID, ep string, token []string, sf *utils.StreamFormatter) error {
|
||||
func (srv *Server) pushImage(r *registry.Registry, out io.Writer, remote, imgID, ep string, token []string, sf *utils.StreamFormatter) (checksum string, err error) {
|
||||
out = utils.NewWriteFlusher(out)
|
||||
jsonRaw, err := ioutil.ReadFile(path.Join(srv.runtime.graph.Root, imgID, "json"))
|
||||
if err != nil {
|
||||
return fmt.Errorf("Error while retreiving the path for {%s}: %s", imgID, err)
|
||||
return "", fmt.Errorf("Error while retreiving the path for {%s}: %s", imgID, err)
|
||||
}
|
||||
out.Write(sf.FormatStatus("Pushing %s", imgID))
|
||||
|
||||
// Make sure we have the image's checksum
|
||||
checksum, err := srv.getChecksum(imgID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
imgData := ®istry.ImgData{
|
||||
ID: imgID,
|
||||
Checksum: checksum,
|
||||
ID: imgID,
|
||||
}
|
||||
|
||||
// Send the json
|
||||
if err := r.PushImageJSONRegistry(imgData, jsonRaw, ep, token); err != nil {
|
||||
if err == registry.ErrAlreadyExists {
|
||||
out.Write(sf.FormatStatus("Image %s already pushed, skipping", imgData.ID))
|
||||
return nil
|
||||
return "", nil
|
||||
}
|
||||
return err
|
||||
return "", err
|
||||
}
|
||||
|
||||
// Retrieve the tarball to be sent
|
||||
var layerData *TempArchive
|
||||
// If the archive exists, use it
|
||||
file, err := os.Open(layerArchivePath(srv.runtime.graph.imageRoot(imgID)))
|
||||
layerData, err := srv.runtime.graph.TempLayerArchive(imgID, Uncompressed, sf, out)
|
||||
if err != nil {
|
||||
if os.IsNotExist(err) {
|
||||
// If the archive does not exist, create one from the layer
|
||||
layerData, err = srv.runtime.graph.TempLayerArchive(imgID, Xz, sf, out)
|
||||
if err != nil {
|
||||
return fmt.Errorf("Failed to generate layer archive: %s", err)
|
||||
}
|
||||
} else {
|
||||
return err
|
||||
}
|
||||
} else {
|
||||
defer file.Close()
|
||||
st, err := file.Stat()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
layerData = &TempArchive{
|
||||
File: file,
|
||||
Size: st.Size(),
|
||||
}
|
||||
return "", fmt.Errorf("Failed to generate layer archive: %s", err)
|
||||
}
|
||||
|
||||
// Send the layer
|
||||
if err := r.PushImageLayerRegistry(imgData.ID, utils.ProgressReader(layerData, int(layerData.Size), out, sf.FormatProgress("Pushing", "%8v/%v (%v)"), sf), ep, token); err != nil {
|
||||
return err
|
||||
if checksum, err := r.PushImageLayerRegistry(imgData.ID, utils.ProgressReader(layerData, int(layerData.Size), out, sf.FormatProgress("Pushing", "%8v/%v (%v)"), sf), ep, token, jsonRaw); err != nil {
|
||||
return "", err
|
||||
} else {
|
||||
imgData.Checksum = checksum
|
||||
}
|
||||
return nil
|
||||
|
||||
// Send the checksum
|
||||
if err := r.PushImageChecksumRegistry(imgData, ep, token); err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
return imgData.Checksum, nil
|
||||
}
|
||||
|
||||
// FIXME: Allow to interupt current push when new push of same image is done.
|
||||
|
|
@ -802,7 +740,7 @@ func (srv *Server) ImagePush(localName string, out io.Writer, sf *utils.StreamFo
|
|||
|
||||
var token []string
|
||||
out.Write(sf.FormatStatus("The push refers to an image: [%s]", localName))
|
||||
if err := srv.pushImage(r, out, remoteName, img.ID, endpoint, token, sf); err != nil {
|
||||
if _, err := srv.pushImage(r, out, remoteName, img.ID, endpoint, token, sf); err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
|
|
|
|||
158
utils/tarsum.go
Normal file
158
utils/tarsum.go
Normal file
|
|
@ -0,0 +1,158 @@
|
|||
package utils
|
||||
|
||||
import (
|
||||
"archive/tar"
|
||||
"bytes"
|
||||
"compress/gzip"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"hash"
|
||||
"io"
|
||||
"sort"
|
||||
"strconv"
|
||||
)
|
||||
|
||||
type verboseHash struct {
|
||||
hash.Hash
|
||||
}
|
||||
|
||||
func (h verboseHash) Write(buf []byte) (int, error) {
|
||||
Debugf("--->%s<---", buf)
|
||||
return h.Hash.Write(buf)
|
||||
}
|
||||
|
||||
type TarSum struct {
|
||||
io.Reader
|
||||
tarR *tar.Reader
|
||||
tarW *tar.Writer
|
||||
gz *gzip.Writer
|
||||
bufTar *bytes.Buffer
|
||||
bufGz *bytes.Buffer
|
||||
h hash.Hash
|
||||
h2 verboseHash
|
||||
sums []string
|
||||
finished bool
|
||||
first bool
|
||||
}
|
||||
|
||||
func (ts *TarSum) encodeHeader(h *tar.Header) error {
|
||||
for _, elem := range [][2]string{
|
||||
{"name", h.Name},
|
||||
{"mode", strconv.Itoa(int(h.Mode))},
|
||||
{"uid", strconv.Itoa(h.Uid)},
|
||||
{"gid", strconv.Itoa(h.Gid)},
|
||||
{"size", strconv.Itoa(int(h.Size))},
|
||||
{"mtime", strconv.Itoa(int(h.ModTime.UTC().Unix()))},
|
||||
{"typeflag", string([]byte{h.Typeflag})},
|
||||
{"linkname", h.Linkname},
|
||||
{"uname", h.Uname},
|
||||
{"gname", h.Gname},
|
||||
{"devmajor", strconv.Itoa(int(h.Devmajor))},
|
||||
{"devminor", strconv.Itoa(int(h.Devminor))},
|
||||
// {"atime", strconv.Itoa(int(h.AccessTime.UTC().Unix()))},
|
||||
// {"ctime", strconv.Itoa(int(h.ChangeTime.UTC().Unix()))},
|
||||
} {
|
||||
// Debugf("-->%s<-- -->%s<--", elem[0], elem[1])
|
||||
if _, err := ts.h.Write([]byte(elem[0] + elem[1])); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (ts *TarSum) Read(buf []byte) (int, error) {
|
||||
if ts.gz == nil {
|
||||
ts.bufTar = bytes.NewBuffer([]byte{})
|
||||
ts.bufGz = bytes.NewBuffer([]byte{})
|
||||
ts.tarR = tar.NewReader(ts.Reader)
|
||||
ts.tarW = tar.NewWriter(ts.bufTar)
|
||||
ts.gz = gzip.NewWriter(ts.bufGz)
|
||||
ts.h = sha256.New()
|
||||
// ts.h = verboseHash{sha256.New()}
|
||||
ts.h.Reset()
|
||||
ts.first = true
|
||||
}
|
||||
|
||||
if ts.finished {
|
||||
return ts.bufGz.Read(buf)
|
||||
}
|
||||
buf2 := make([]byte, len(buf), cap(buf))
|
||||
|
||||
n, err := ts.tarR.Read(buf2)
|
||||
if err != nil {
|
||||
if err == io.EOF {
|
||||
if _, err := ts.h.Write(buf2[:n]); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
if !ts.first {
|
||||
ts.sums = append(ts.sums, hex.EncodeToString(ts.h.Sum(nil)))
|
||||
ts.h.Reset()
|
||||
} else {
|
||||
ts.first = false
|
||||
}
|
||||
|
||||
currentHeader, err := ts.tarR.Next()
|
||||
if err != nil {
|
||||
if err == io.EOF {
|
||||
if err := ts.gz.Close(); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
ts.finished = true
|
||||
return n, nil
|
||||
}
|
||||
return n, err
|
||||
}
|
||||
if err := ts.encodeHeader(currentHeader); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
if err := ts.tarW.WriteHeader(currentHeader); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
if _, err := ts.tarW.Write(buf2[:n]); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
ts.tarW.Flush()
|
||||
if _, err := io.Copy(ts.gz, ts.bufTar); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
ts.gz.Flush()
|
||||
|
||||
return ts.bufGz.Read(buf)
|
||||
}
|
||||
return n, err
|
||||
}
|
||||
|
||||
// Filling the hash buffer
|
||||
if _, err = ts.h.Write(buf2[:n]); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
// Filling the tar writter
|
||||
if _, err = ts.tarW.Write(buf2[:n]); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
ts.tarW.Flush()
|
||||
|
||||
// Filling the gz writter
|
||||
if _, err = io.Copy(ts.gz, ts.bufTar); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
ts.gz.Flush()
|
||||
|
||||
return ts.bufGz.Read(buf)
|
||||
}
|
||||
|
||||
func (ts *TarSum) Sum(extra []byte) string {
|
||||
sort.Strings(ts.sums)
|
||||
h := sha256.New()
|
||||
if extra != nil {
|
||||
h.Write(extra)
|
||||
}
|
||||
for _, sum := range ts.sums {
|
||||
Debugf("-->%s<--", sum)
|
||||
h.Write([]byte(sum))
|
||||
}
|
||||
checksum := "tarsum+sha256:" + hex.EncodeToString(h.Sum(nil))
|
||||
Debugf("checksum processed: %s", checksum)
|
||||
return checksum
|
||||
}
|
||||
Loading…
Reference in a new issue