From 0e71e368a8a781f593b25fdd1318d3882e6d28e5 Mon Sep 17 00:00:00 2001 From: Victor Vieux Date: Wed, 24 Jul 2013 15:41:34 +0000 Subject: [PATCH 1/5] Add ID to JSONMessage in pull Use goroutines to pull in parallel If multiple images pulled at the same time, each progress is displayed on a new line --- commands.go | 2 +- graph.go | 2 +- server.go | 63 ++++++++++++++++++++++++++++++-------------------- utils/utils.go | 13 +++++++---- 4 files changed, 49 insertions(+), 31 deletions(-) diff --git a/commands.go b/commands.go index 2d8ea4efb5..946c83dd11 100644 --- a/commands.go +++ b/commands.go @@ -196,7 +196,7 @@ func (cli *DockerCli) CmdBuild(args ...string) error { // FIXME: ProgressReader shouldn't be this annoyning to use if context != nil { sf := utils.NewStreamFormatter(false) - body = utils.ProgressReader(ioutil.NopCloser(context), 0, cli.err, sf.FormatProgress("Uploading context", "%v bytes%0.0s%0.0s"), sf) + body = utils.ProgressReader(ioutil.NopCloser(context), 0, cli.err, sf.FormatProgress("Uploading context", "%v bytes%0.0s%0.0s", ""), sf) } // Upload the build context v := &url.Values{} diff --git a/graph.go b/graph.go index 42d1bdbd4c..3ae342e7c5 100644 --- a/graph.go +++ b/graph.go @@ -175,7 +175,7 @@ func (graph *Graph) TempLayerArchive(id string, compression Compression, sf *uti if err != nil { return nil, err } - return NewTempArchive(utils.ProgressReader(ioutil.NopCloser(archive), 0, output, sf.FormatProgress("Buffering to disk", "%v/%v (%v)"), sf), tmp.Root) + return NewTempArchive(utils.ProgressReader(ioutil.NopCloser(archive), 0, output, sf.FormatProgress("Buffering to disk", "%v/%v (%v)", ""), sf), tmp.Root) } // Mktemp creates a temporary sub-directory inside the graph's filesystem. diff --git a/server.go b/server.go index 4179a1e160..1a221d3f00 100644 --- a/server.go +++ b/server.go @@ -145,7 +145,7 @@ func (srv *Server) ImageInsert(name, url, path string, out io.Writer, sf *utils. return "", err } - if err := c.Inject(utils.ProgressReader(file.Body, int(file.ContentLength), out, sf.FormatProgress("Downloading", "%8v/%v (%v)"), sf), path); err != nil { + if err := c.Inject(utils.ProgressReader(file.Body, int(file.ContentLength), out, sf.FormatProgress("Downloading", "%8v/%v (%v)", ""), sf), path); err != nil { return "", err } // FIXME: Handle custom repo, tag comment, author @@ -425,7 +425,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(utils.ProgressReader(layer, imgSize, out, sf.FormatProgress("Downloading", "%8v/%v (%v)", id), sf), false, img); err != nil { return err } } @@ -477,30 +477,43 @@ func (srv *Server) pullRepository(r *registry.Registry, out io.Writer, localName repoData.ImgList[id].Tag = askedTag } - for _, img := range repoData.ImgList { - if askedTag != "" && img.Tag != askedTag { - utils.Debugf("(%s) does not match %s (id: %s), skipping", img.Tag, askedTag, img.ID) - continue - } - - if img.Tag == "" { - utils.Debugf("Image (id: %s) present in this repository but untagged, skipping", img.ID) - continue - } - out.Write(sf.FormatStatus("Pulling image %s (%s) from %s", img.ID, img.Tag, localName)) - success := false - for _, ep := range repoData.Endpoints { - if err := srv.pullImage(r, out, img.ID, ep, repoData.Tokens, sf); err != nil { - out.Write(sf.FormatStatus("Error while retrieving image for tag: %s (%s); checking next endpoint", askedTag, err)) - continue + errors := make(chan error) + for _, image := range repoData.ImgList { + go func(img *registry.ImgData) { + if askedTag != "" && img.Tag != askedTag { + utils.Debugf("(%s) does not match %s (id: %s), skipping", img.Tag, askedTag, img.ID) + errors <- nil + return } - success = true - break - } - if !success { - return fmt.Errorf("Could not find repository on any of the indexed registries.") + + if img.Tag == "" { + utils.Debugf("Image (id: %s) present in this repository but untagged, skipping", img.ID) + errors <- nil + return + } + out.Write(sf.FormatStatus("Pulling image %s (%s) from %s", img.ID, img.Tag, localName)) + success := false + for _, ep := range repoData.Endpoints { + if err := srv.pullImage(r, out, img.ID, ep, repoData.Tokens, sf); err != nil { + out.Write(sf.FormatStatus("Error while retrieving image for tag: %s (%s); checking next endpoint", askedTag, err)) + continue + } + success = true + break + } + if !success { + errors <- fmt.Errorf("Could not find repository on any of the indexed registries.") + } + errors <- nil + }(image) + } + + for i := 0; i < len(repoData.ImgList); i++ { + if err := <-errors; err != nil { + return err } } + for tag, id := range tagsList { if askedTag != "" && tag != askedTag { continue @@ -748,7 +761,7 @@ func (srv *Server) pushImage(r *registry.Registry, out io.Writer, remote, imgID, } // 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 { + 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 } return nil @@ -818,7 +831,7 @@ func (srv *Server) ImageImport(src, repo, tag string, in io.Reader, out io.Write if err != nil { return err } - archive = utils.ProgressReader(resp.Body, int(resp.ContentLength), out, sf.FormatProgress("Importing", "%8v/%v (%v)"), sf) + archive = utils.ProgressReader(resp.Body, int(resp.ContentLength), out, sf.FormatProgress("Importing", "%8v/%v (%v)", ""), sf) } img, err := srv.runtime.graph.Create(archive, nil, "Imported from "+src, "", nil) if err != nil { diff --git a/utils/utils.go b/utils/utils.go index acb015becd..ffba2352a7 100644 --- a/utils/utils.go +++ b/utils/utils.go @@ -107,7 +107,7 @@ func (r *progressReader) Close() error { func ProgressReader(r io.ReadCloser, size int, output io.Writer, template []byte, sf *StreamFormatter) *progressReader { tpl := string(template) if tpl == "" { - tpl = string(sf.FormatProgress("", "%8v/%v (%v)")) + tpl = string(sf.FormatProgress("", "%8v/%v (%v)", "")) } return &progressReader{r, NewWriteFlusher(output), size, 0, 0, tpl, sf} } @@ -587,11 +587,14 @@ type NopFlusher struct{} func (f *NopFlusher) Flush() {} type WriteFlusher struct { + sync.Mutex w io.Writer flusher http.Flusher } func (wf *WriteFlusher) Write(b []byte) (n int, err error) { + wf.Lock() + defer wf.Unlock() n, err = wf.w.Write(b) wf.flusher.Flush() return n, err @@ -619,7 +622,9 @@ func (jm *JSONMessage) Display(out io.Writer) (error) { if jm.Time != 0 { fmt.Fprintf(out, "[%s] ", time.Unix(jm.Time, 0)) } - if jm.Progress != "" { + if jm.Progress != "" && jm.ID != ""{ + fmt.Fprintf(out, "\n%s %s %s\r", jm.Status, jm.ID, jm.Progress) + } else if jm.Progress != "" { fmt.Fprintf(out, "%s %s\r", jm.Status, jm.Progress) } else if jm.Error != "" { return fmt.Errorf(jm.Error) @@ -665,10 +670,10 @@ func (sf *StreamFormatter) FormatError(err error) []byte { return []byte("Error: " + err.Error() + "\r\n") } -func (sf *StreamFormatter) FormatProgress(action, str string) []byte { +func (sf *StreamFormatter) FormatProgress(action, str, id string) []byte { sf.used = true if sf.json { - b, err := json.Marshal(&JSONMessage{Status: action, Progress: str}) + b, err := json.Marshal(&JSONMessage{Status: action, Progress: str, ID:id}) if err != nil { return nil } From 8742649aa7f3524bbfa99b68c8d87ffc5aba0af9 Mon Sep 17 00:00:00 2001 From: Victor Vieux Date: Wed, 24 Jul 2013 17:10:59 +0000 Subject: [PATCH 2/5] improve client output --- commands.go | 7 +++++-- graph.go | 2 +- server.go | 44 ++++++++++++++++++++++---------------------- utils/utils.go | 30 ++++++++++++------------------ 4 files changed, 40 insertions(+), 43 deletions(-) diff --git a/commands.go b/commands.go index 946c83dd11..2a672d49e3 100644 --- a/commands.go +++ b/commands.go @@ -196,7 +196,7 @@ func (cli *DockerCli) CmdBuild(args ...string) error { // FIXME: ProgressReader shouldn't be this annoyning to use if context != nil { sf := utils.NewStreamFormatter(false) - body = utils.ProgressReader(ioutil.NopCloser(context), 0, cli.err, sf.FormatProgress("Uploading context", "%v bytes%0.0s%0.0s", ""), sf) + body = utils.ProgressReader(ioutil.NopCloser(context), 0, cli.err, sf.FormatProgress("", "Uploading context", "%v bytes%0.0s%0.0s"), sf) } // Upload the build context v := &url.Values{} @@ -1537,8 +1537,8 @@ func (cli *DockerCli) stream(method, path string, in io.Reader, out io.Writer) e if resp.Header.Get("Content-Type") == "application/json" { dec := json.NewDecoder(resp.Body) + jm := utils.JSONMessage{} for { - var jm utils.JSONMessage if err := dec.Decode(&jm); err == io.EOF { break } else if err != nil { @@ -1546,6 +1546,9 @@ func (cli *DockerCli) stream(method, path string, in io.Reader, out io.Writer) e } jm.Display(out) } + if jm.Progress != "" { + fmt.Fprintf(out, "\n") + } } else { if _, err := io.Copy(out, resp.Body); err != nil { return err diff --git a/graph.go b/graph.go index 3ae342e7c5..5b9162b871 100644 --- a/graph.go +++ b/graph.go @@ -175,7 +175,7 @@ func (graph *Graph) TempLayerArchive(id string, compression Compression, sf *uti if err != nil { return nil, err } - return NewTempArchive(utils.ProgressReader(ioutil.NopCloser(archive), 0, output, sf.FormatProgress("Buffering to disk", "%v/%v (%v)", ""), sf), tmp.Root) + return NewTempArchive(utils.ProgressReader(ioutil.NopCloser(archive), 0, output, sf.FormatProgress("", "Buffering to disk", "%v/%v (%v)"), sf), tmp.Root) } // Mktemp creates a temporary sub-directory inside the graph's filesystem. diff --git a/server.go b/server.go index 1a221d3f00..a1d22d3c44 100644 --- a/server.go +++ b/server.go @@ -145,7 +145,7 @@ func (srv *Server) ImageInsert(name, url, path string, out io.Writer, sf *utils. return "", err } - if err := c.Inject(utils.ProgressReader(file.Body, int(file.ContentLength), out, sf.FormatProgress("Downloading", "%8v/%v (%v)", ""), sf), path); err != nil { + if err := c.Inject(utils.ProgressReader(file.Body, int(file.ContentLength), out, sf.FormatProgress("", "Downloading", "%8v/%v (%v)"), sf), path); err != nil { return "", err } // FIXME: Handle custom repo, tag comment, author @@ -153,7 +153,7 @@ func (srv *Server) ImageInsert(name, url, path string, out io.Writer, sf *utils. if err != nil { return "", err } - out.Write(sf.FormatStatus(img.ID)) + out.Write(sf.FormatStatus("", img.ID)) return img.ShortID(), nil } @@ -407,7 +407,7 @@ func (srv *Server) pullImage(r *registry.Registry, out io.Writer, imgID, endpoin // FIXME: Launch the getRemoteImage() in goroutines for _, id := range history { if !srv.runtime.graph.Exists(id) { - out.Write(sf.FormatStatus("Pulling %s metadata", id)) + out.Write(sf.FormatStatus(utils.TruncateID(id), "Pulling metadata")) imgJSON, imgSize, err := r.GetRemoteImageJSON(id, endpoint, token) if err != nil { // FIXME: Keep goging in case of error? @@ -419,13 +419,13 @@ func (srv *Server) pullImage(r *registry.Registry, out io.Writer, imgID, endpoin } // Get the layer - out.Write(sf.FormatStatus("Pulling %s fs layer", id)) + out.Write(sf.FormatStatus(utils.TruncateID(id), "Pulling fs layer")) layer, err := r.GetRemoteImageLayer(img.ID, endpoint, token) if err != nil { return err } defer layer.Close() - if err := srv.runtime.graph.Register(utils.ProgressReader(layer, imgSize, out, sf.FormatProgress("Downloading", "%8v/%v (%v)", id), sf), false, img); err != nil { + if err := srv.runtime.graph.Register(utils.ProgressReader(layer, imgSize, out, sf.FormatProgress(utils.TruncateID(id), "Downloading", "%8v/%v (%v)"), sf), false, img); err != nil { return err } } @@ -434,7 +434,7 @@ func (srv *Server) pullImage(r *registry.Registry, out io.Writer, imgID, endpoin } func (srv *Server) pullRepository(r *registry.Registry, out io.Writer, localName, remoteName, askedTag, indexEp string, sf *utils.StreamFormatter) error { - out.Write(sf.FormatStatus("Pulling repository %s", localName)) + out.Write(sf.FormatStatus("", "Pulling repository %s", localName)) repoData, err := r.GetRepositoryData(indexEp, remoteName) if err != nil { @@ -491,11 +491,11 @@ func (srv *Server) pullRepository(r *registry.Registry, out io.Writer, localName errors <- nil return } - out.Write(sf.FormatStatus("Pulling image %s (%s) from %s", img.ID, img.Tag, localName)) + out.Write(sf.FormatStatus(utils.TruncateID(img.ID), "Pulling image (%s) from %s", img.Tag, localName)) success := false for _, ep := range repoData.Endpoints { if err := srv.pullImage(r, out, img.ID, ep, repoData.Tokens, sf); err != nil { - out.Write(sf.FormatStatus("Error while retrieving image for tag: %s (%s); checking next endpoint", askedTag, err)) + out.Write(sf.FormatStatus(utils.TruncateID(img.ID), "Error while retrieving image for tag: %s (%s); checking next endpoint", askedTag, err)) continue } success = true @@ -665,12 +665,12 @@ 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")) + out.Write(sf.FormatStatus("", "Processing checksums")) imgList, err := srv.getImageList(localRepo) if err != nil { return err } - out.Write(sf.FormatStatus("Sending image list")) + out.Write(sf.FormatStatus("", "Sending image list")) var repoData *registry.RepositoryData repoData, err = r.PushImageJSONIndex(indexEp, remoteName, imgList, false, nil) @@ -679,21 +679,21 @@ func (srv *Server) pushRepository(r *registry.Registry, out io.Writer, localName } for _, ep := range repoData.Endpoints { - out.Write(sf.FormatStatus("Pushing repository %s (%d tags)", localName, len(localRepo))) + out.Write(sf.FormatStatus("", "Pushing repository %s (%d tags)", localName, len(localRepo))) // For each image within the repo, push them for _, elem := range imgList { if _, exists := repoData.ImgList[elem.ID]; exists { - out.Write(sf.FormatStatus("Image %s already pushed, skipping", elem.ID)) + out.Write(sf.FormatStatus("", "Image %s already pushed, skipping", elem.ID)) continue } else if r.LookupRemoteImage(elem.ID, ep, repoData.Tokens) { - out.Write(sf.FormatStatus("Image %s already pushed, skipping", elem.ID)) + 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 { // FIXME: Continue on error? return err } - out.Write(sf.FormatStatus("Pushing tags for rev [%s] on {%s}", elem.ID, ep+"repositories/"+remoteName+"/tags/"+elem.Tag)) + 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 { return err } @@ -713,7 +713,7 @@ func (srv *Server) pushImage(r *registry.Registry, out io.Writer, remote, imgID, if err != nil { return fmt.Errorf("Error while retreiving the path for {%s}: %s", imgID, err) } - out.Write(sf.FormatStatus("Pushing %s", imgID)) + out.Write(sf.FormatStatus("", "Pushing %s", imgID)) // Make sure we have the image's checksum checksum, err := srv.getChecksum(imgID) @@ -728,7 +728,7 @@ func (srv *Server) pushImage(r *registry.Registry, out io.Writer, remote, 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)) + out.Write(sf.FormatStatus("", "Image %s already pushed, skipping", imgData.ID)) return nil } return err @@ -761,7 +761,7 @@ func (srv *Server) pushImage(r *registry.Registry, out io.Writer, remote, imgID, } // 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 { + 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 } return nil @@ -789,7 +789,7 @@ func (srv *Server) ImagePush(localName string, out io.Writer, sf *utils.StreamFo if err != nil { reposLen := len(srv.runtime.repositories.Repositories[localName]) - out.Write(sf.FormatStatus("The push refers to a repository [%s] (len: %d)", localName, reposLen)) + out.Write(sf.FormatStatus("", "The push refers to a repository [%s] (len: %d)", localName, reposLen)) // If it fails, try to get the repository if localRepo, exists := srv.runtime.repositories.Repositories[localName]; exists { if err := srv.pushRepository(r, out, localName, remoteName, localRepo, endpoint, sf); err != nil { @@ -801,7 +801,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)) + 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 { return err } @@ -824,14 +824,14 @@ func (srv *Server) ImageImport(src, repo, tag string, in io.Reader, out io.Write u.Host = src u.Path = "" } - out.Write(sf.FormatStatus("Downloading from %s", u)) + out.Write(sf.FormatStatus("", "Downloading from %s", u)) // Download with curl (pretty progress bar) // If curl is not available, fallback to http.Get() resp, err = utils.Download(u.String(), out) if err != nil { return err } - archive = utils.ProgressReader(resp.Body, int(resp.ContentLength), out, sf.FormatProgress("Importing", "%8v/%v (%v)", ""), sf) + archive = utils.ProgressReader(resp.Body, int(resp.ContentLength), out, sf.FormatProgress("", "Importing", "%8v/%v (%v)"), sf) } img, err := srv.runtime.graph.Create(archive, nil, "Imported from "+src, "", nil) if err != nil { @@ -843,7 +843,7 @@ func (srv *Server) ImageImport(src, repo, tag string, in io.Reader, out io.Write return err } } - out.Write(sf.FormatStatus(img.ShortID())) + out.Write(sf.FormatStatus("", img.ShortID())) return nil } diff --git a/utils/utils.go b/utils/utils.go index ffba2352a7..6361945eef 100644 --- a/utils/utils.go +++ b/utils/utils.go @@ -94,11 +94,6 @@ func (r *progressReader) Read(p []byte) (n int, err error) { } r.lastUpdate = r.readProgress } - // Send newline when complete - if err != nil { - r.output.Write(r.sf.FormatStatus("")) - } - return read, err } func (r *progressReader) Close() error { @@ -619,24 +614,23 @@ type JSONMessage struct { } func (jm *JSONMessage) Display(out io.Writer) (error) { + if jm.Error != "" { + return fmt.Errorf(jm.Error) + } if jm.Time != 0 { fmt.Fprintf(out, "[%s] ", time.Unix(jm.Time, 0)) } - if jm.Progress != "" && jm.ID != ""{ - fmt.Fprintf(out, "\n%s %s %s\r", jm.Status, jm.ID, jm.Progress) - } else if jm.Progress != "" { + if jm.ID != "" { + fmt.Fprintf(out, "%s: ", jm.ID) + } + if jm.Progress != "" { fmt.Fprintf(out, "%s %s\r", jm.Status, jm.Progress) - } else if jm.Error != "" { - return fmt.Errorf(jm.Error) - } else if jm.ID != "" { - fmt.Fprintf(out, "%s: %s\n", jm.ID, jm.Status) } else { fmt.Fprintf(out, "%s\n", jm.Status) } return nil } - type StreamFormatter struct { json bool used bool @@ -646,11 +640,11 @@ func NewStreamFormatter(json bool) *StreamFormatter { return &StreamFormatter{json, false} } -func (sf *StreamFormatter) FormatStatus(format string, a ...interface{}) []byte { +func (sf *StreamFormatter) FormatStatus(id, format string, a ...interface{}) []byte { sf.used = true str := fmt.Sprintf(format, a...) if sf.json { - b, err := json.Marshal(&JSONMessage{Status: str}) + b, err := json.Marshal(&JSONMessage{ID: id, Status: str}) if err != nil { return sf.FormatError(err) } @@ -670,16 +664,16 @@ func (sf *StreamFormatter) FormatError(err error) []byte { return []byte("Error: " + err.Error() + "\r\n") } -func (sf *StreamFormatter) FormatProgress(action, str, id string) []byte { +func (sf *StreamFormatter) FormatProgress(id, action, progress string) []byte { sf.used = true if sf.json { - b, err := json.Marshal(&JSONMessage{Status: action, Progress: str, ID:id}) + b, err := json.Marshal(&JSONMessage{Status: action, Progress: progress, ID:id}) if err != nil { return nil } return b } - return []byte(action + " " + str + "\r") + return []byte(action + " " + progress + "\r") } func (sf *StreamFormatter) Used() bool { From f1dd299227b15696872822f40a4ab9f1a54098a7 Mon Sep 17 00:00:00 2001 From: Victor Vieux Date: Wed, 24 Jul 2013 20:16:02 +0000 Subject: [PATCH 3/5] Use VT100 escape codes : --- commands.go | 14 +------------- utils/utils.go | 38 +++++++++++++++++++++++++++++++++++++- 2 files changed, 38 insertions(+), 14 deletions(-) diff --git a/commands.go b/commands.go index 2a672d49e3..99d91ac771 100644 --- a/commands.go +++ b/commands.go @@ -1536,19 +1536,7 @@ func (cli *DockerCli) stream(method, path string, in io.Reader, out io.Writer) e } if resp.Header.Get("Content-Type") == "application/json" { - dec := json.NewDecoder(resp.Body) - jm := utils.JSONMessage{} - for { - if err := dec.Decode(&jm); err == io.EOF { - break - } else if err != nil { - return err - } - jm.Display(out) - } - if jm.Progress != "" { - fmt.Fprintf(out, "\n") - } + utils.DisplayJSONMessagesStream(resp.Body, out) } else { if _, err := io.Copy(out, resp.Body); err != nil { return err diff --git a/utils/utils.go b/utils/utils.go index 6361945eef..6081935b41 100644 --- a/utils/utils.go +++ b/utils/utils.go @@ -617,6 +617,7 @@ func (jm *JSONMessage) Display(out io.Writer) (error) { if jm.Error != "" { return fmt.Errorf(jm.Error) } + fmt.Fprintf(out, "%c[2K", 27) if jm.Time != 0 { fmt.Fprintf(out, "[%s] ", time.Unix(jm.Time, 0)) } @@ -626,8 +627,43 @@ func (jm *JSONMessage) Display(out io.Writer) (error) { if jm.Progress != "" { fmt.Fprintf(out, "%s %s\r", jm.Status, jm.Progress) } else { - fmt.Fprintf(out, "%s\n", jm.Status) + fmt.Fprintf(out, "%s\r", jm.Status) } + if jm.ID == "" { + fmt.Fprintf(out, "\n") + } + return nil +} + +func DisplayJSONMessagesStream(in io.Reader, out io.Writer) error { + dec := json.NewDecoder(in) + jm := JSONMessage{} + ids := make(map[string]int) + diff := 0 + for { + if err := dec.Decode(&jm); err == io.EOF { + break + } else if err != nil { + return err + } + if jm.ID != "" { + line, ok := ids[jm.ID] + if !ok { + line = len(ids) + ids[jm.ID] = line + fmt.Fprintf(out, "\n") + diff = 0 + } else { + diff = len(ids) - line + } + fmt.Fprintf(out, "%c[%dA", 27, diff) + } + jm.Display(out) + if jm.ID != "" { + fmt.Fprintf(out, "%c[%dB", 27, diff) + } + } +// fmt.Fprintf(out, "\n") return nil } From 01e98bf0dd26a1de7fd280fd4b0f0a79aedd0cdd Mon Sep 17 00:00:00 2001 From: Victor Vieux Date: Thu, 25 Jul 2013 14:32:46 +0000 Subject: [PATCH 4/5] fix errors --- commands.go | 2 +- utils/utils.go | 6 ++++-- 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/commands.go b/commands.go index 99d91ac771..f0d8f61999 100644 --- a/commands.go +++ b/commands.go @@ -1536,7 +1536,7 @@ func (cli *DockerCli) stream(method, path string, in io.Reader, out io.Writer) e } if resp.Header.Get("Content-Type") == "application/json" { - utils.DisplayJSONMessagesStream(resp.Body, out) + return utils.DisplayJSONMessagesStream(resp.Body, out) } else { if _, err := io.Copy(out, resp.Body); err != nil { return err diff --git a/utils/utils.go b/utils/utils.go index 6081935b41..a66319a1ba 100644 --- a/utils/utils.go +++ b/utils/utils.go @@ -658,12 +658,14 @@ func DisplayJSONMessagesStream(in io.Reader, out io.Writer) error { } fmt.Fprintf(out, "%c[%dA", 27, diff) } - jm.Display(out) + err := jm.Display(out) if jm.ID != "" { fmt.Fprintf(out, "%c[%dB", 27, diff) } + if err != nil { + return err + } } -// fmt.Fprintf(out, "\n") return nil } From 46f59dd9333baa578b184eb25b386ac2f41caf04 Mon Sep 17 00:00:00 2001 From: Victor Vieux Date: Tue, 30 Jul 2013 12:09:07 +0000 Subject: [PATCH 5/5] add parallel pull to 1.4 --- api.go | 2 +- buildfile.go | 2 +- docs/sources/api/docker_remote_api.rst | 4 ++++ runtime_test.go | 2 +- server.go | 24 ++++++++++++++++-------- utils/utils.go | 5 +---- 6 files changed, 24 insertions(+), 15 deletions(-) diff --git a/api.go b/api.go index 4ad2ba461a..cc7482be71 100644 --- a/api.go +++ b/api.go @@ -386,7 +386,7 @@ func postImagesCreate(srv *Server, version float64, w http.ResponseWriter, r *ht } sf := utils.NewStreamFormatter(version > 1.0) if image != "" { //pull - if err := srv.ImagePull(image, tag, w, sf, &auth.AuthConfig{}); err != nil { + if err := srv.ImagePull(image, tag, w, sf, &auth.AuthConfig{}, version > 1.3); err != nil { if sf.Used() { w.Write(sf.FormatError(err)) return nil diff --git a/buildfile.go b/buildfile.go index 736725e915..fda6e5bf41 100644 --- a/buildfile.go +++ b/buildfile.go @@ -55,7 +55,7 @@ func (b *buildFile) CmdFrom(name string) error { if err != nil { if b.runtime.graph.IsNotExist(err) { remote, tag := utils.ParseRepositoryTag(name) - if err := b.srv.ImagePull(remote, tag, b.out, utils.NewStreamFormatter(false), nil); err != nil { + if err := b.srv.ImagePull(remote, tag, b.out, utils.NewStreamFormatter(false), nil, true); err != nil { return err } image, err = b.runtime.repositories.LookupImage(name) diff --git a/docs/sources/api/docker_remote_api.rst b/docs/sources/api/docker_remote_api.rst index 193be501d0..dd017cde8d 100644 --- a/docs/sources/api/docker_remote_api.rst +++ b/docs/sources/api/docker_remote_api.rst @@ -40,6 +40,10 @@ You can still call an old version of the api using What's new ---------- +.. http:post:: /images/create + + **New!** When pull a repo, all images are now downloaded in parallel. + .. http:get:: /containers/(id)/top **New!** You can now use ps args with docker top, like `docker top aux` diff --git a/runtime_test.go b/runtime_test.go index 0b0f62f199..7ecf199e09 100644 --- a/runtime_test.go +++ b/runtime_test.go @@ -101,7 +101,7 @@ func init() { // If the unit test is not found, try to download it. if img, err := globalRuntime.repositories.LookupImage(unitTestImageName); err != nil || img.ID != unitTestImageID { // Retrieve the Image - if err := srv.ImagePull(unitTestImageName, "", os.Stdout, utils.NewStreamFormatter(false), nil); err != nil { + if err := srv.ImagePull(unitTestImageName, "", os.Stdout, utils.NewStreamFormatter(false), nil, true); err != nil { panic(err) } } diff --git a/server.go b/server.go index dc08ee5fbd..7417264e50 100644 --- a/server.go +++ b/server.go @@ -446,7 +446,7 @@ func (srv *Server) pullImage(r *registry.Registry, out io.Writer, imgID, endpoin return nil } -func (srv *Server) pullRepository(r *registry.Registry, out io.Writer, localName, remoteName, askedTag, indexEp string, sf *utils.StreamFormatter) error { +func (srv *Server) pullRepository(r *registry.Registry, out io.Writer, localName, remoteName, askedTag, indexEp string, sf *utils.StreamFormatter, parallel bool) error { out.Write(sf.FormatStatus("", "Pulling repository %s", localName)) repoData, err := r.GetRepositoryData(indexEp, remoteName) @@ -492,7 +492,7 @@ func (srv *Server) pullRepository(r *registry.Registry, out io.Writer, localName errors := make(chan error) for _, image := range repoData.ImgList { - go func(img *registry.ImgData) { + downloadImage := func(img *registry.ImgData) { if askedTag != "" && img.Tag != askedTag { utils.Debugf("(%s) does not match %s (id: %s), skipping", img.Tag, askedTag, img.ID) errors <- nil @@ -518,12 +518,20 @@ func (srv *Server) pullRepository(r *registry.Registry, out io.Writer, localName errors <- fmt.Errorf("Could not find repository on any of the indexed registries.") } errors <- nil - }(image) + } + + if parallel { + go downloadImage(image) + } else { + downloadImage(image) + } } - for i := 0; i < len(repoData.ImgList); i++ { - if err := <-errors; err != nil { - return err + if parallel { + for i := 0; i < len(repoData.ImgList); i++ { + if err := <-errors; err != nil { + return err + } } } @@ -577,7 +585,7 @@ func (srv *Server) poolRemove(kind, key string) error { return nil } -func (srv *Server) ImagePull(localName string, tag string, out io.Writer, sf *utils.StreamFormatter, authConfig *auth.AuthConfig) error { +func (srv *Server) ImagePull(localName string, tag string, out io.Writer, sf *utils.StreamFormatter, authConfig *auth.AuthConfig, parallel bool) error { r, err := registry.NewRegistry(srv.runtime.root, authConfig, srv.versionInfos()...) if err != nil { return err @@ -599,7 +607,7 @@ func (srv *Server) ImagePull(localName string, tag string, out io.Writer, sf *ut } out = utils.NewWriteFlusher(out) - err = srv.pullRepository(r, out, localName, remoteName, tag, endpoint, sf) + err = srv.pullRepository(r, out, localName, remoteName, tag, endpoint, sf, parallel) if err != nil { if err := srv.pullImage(r, out, remoteName, endpoint, nil, sf); err != nil { return err diff --git a/utils/utils.go b/utils/utils.go index 77f701c248..f91fb0663c 100644 --- a/utils/utils.go +++ b/utils/utils.go @@ -635,7 +635,6 @@ func (jm *JSONMessage) Display(out io.Writer) error { return nil } - func DisplayJSONMessagesStream(in io.Reader, out io.Writer) error { dec := json.NewDecoder(in) jm := JSONMessage{} @@ -670,8 +669,6 @@ func DisplayJSONMessagesStream(in io.Reader, out io.Writer) error { return nil } -======= ->>>>>>> master type StreamFormatter struct { json bool used bool @@ -708,7 +705,7 @@ func (sf *StreamFormatter) FormatError(err error) []byte { func (sf *StreamFormatter) FormatProgress(id, action, progress string) []byte { sf.used = true if sf.json { - b, err := json.Marshal(&JSONMessage{Status: action, Progress: progress, ID:id}) + b, err := json.Marshal(&JSONMessage{Status: action, Progress: progress, ID: id}) if err != nil { return nil }