@@ -14,7 +14,7 @@ import (

 	"github.com/golang/glog"

 	starterrors "github.com/openshift/origin/pkg/bootstrap/docker/errors"

-	"github.com/openshift/origin/pkg/cmd/util/pullprogress"

+	"github.com/openshift/origin/pkg/util/docker/dockerfile/builder/imageprogress"

 )

 const openShiftInsecureCIDR = "172.30.0.0/16"

@@ -118,31 +118,11 @@ func (h *Helper) CheckAndPull(image string, out io.Writer) error {

 	}

 	glog.V(5).Infof("Image %q not found. Pulling", image)

 	fmt.Fprintf(out, "Pulling image %s\n", image)

-	extracting := false

-	var outputStream io.Writer

-	writeProgress := func(r *pullprogress.ProgressReport) {

-		if extracting {

-			return

-		}

-		if r.Downloading == 0 && r.Waiting == 0 && r.Extracting > 0 {

-			fmt.Fprintf(out, "Extracting\n")

-			extracting = true

-			return

-		}

-		plural := "s"

-		if r.Downloading == 1 {

-			plural = " "

-		}

-		fmt.Fprintf(out, "Downloading %d layer%s (%3.0f%%)", r.Downloading, plural, r.DownloadPct)

-		if r.Waiting > 0 {

-			fmt.Fprintf(out, ", %d waiting\n", r.Waiting)

-		} else {

-			fmt.Fprintf(out, "\n")

-		}

+	logProgress := func(s string) {

+		fmt.Fprintf(out, "%s\n", s)

 	}

-	if !glog.V(5) {

-		outputStream = pullprogress.NewPullProgressWriter(writeProgress)

-	} else {

+	outputStream := imageprogress.NewPullWriter(logProgress)

+	if glog.V(5) {

 		outputStream = out

 	}

 	err = h.client.PullImage(docker.PullImageOptions{

@@ -153,7 +133,7 @@ func (h *Helper) CheckAndPull(image string, out io.Writer) error {

 	if err != nil {

 		return starterrors.NewError("error pulling Docker image %s", image).WithCause(err)

 	}

-	fmt.Fprintf(out, "Image pull comlete\n")

+	fmt.Fprintf(out, "Image pull complete\n")

 	return nil

 }

@@ -7,13 +7,15 @@ import (

 	"strings"

 	"time"

-	s2iapi "github.com/openshift/source-to-image/pkg/api"

+	"github.com/docker/docker/pkg/parsers"

+	docker "github.com/fsouza/go-dockerclient"

 	"k8s.io/kubernetes/pkg/util/interrupt"

 	utilruntime "k8s.io/kubernetes/pkg/util/runtime"

-	"github.com/docker/docker/pkg/parsers"

-	docker "github.com/fsouza/go-dockerclient"

+	s2iapi "github.com/openshift/source-to-image/pkg/api"

 	"github.com/openshift/source-to-image/pkg/tar"

+

+	"github.com/openshift/origin/pkg/util/docker/dockerfile/builder/imageprogress"

 )

 var (

@@ -55,12 +57,18 @@ type DockerClient interface {

 // If any other scenario the push will fail, without retries.

 func pushImage(client DockerClient, name string, authConfig docker.AuthConfiguration) error {

 	repository, tag := docker.ParseRepositoryTag(name)

+	logProgress := func(s string) {

+		glog.V(1).Infof("%s", s)

+	}

 	opts := docker.PushImageOptions{

-		Name: repository,

-		Tag:  tag,

+		Name:          repository,

+		Tag:           tag,

+		OutputStream:  imageprogress.NewPushWriter(logProgress),

+		RawJSONStream: true,

 	}

 	if glog.Is(5) {

 		opts.OutputStream = os.Stderr

+		opts.RawJSONStream = false

 	}

 	var err error

 	var retriableError = false

@@ -265,7 +265,7 @@ func (s *S2IBuilder) Build() error {

 		} else {

 			glog.V(2).Infof("No push secret provided")

 		}

-		glog.V(1).Infof("Pushing %s image ...", pushTag)

+		glog.V(1).Infof("Pushing image %s ...", pushTag)

 		if err := pushImage(s.dockerClient, pushTag, pushAuthConfig); err != nil {

 			// write extended error message to assist in problem resolution

 			msg := fmt.Sprintf("Failed to push image. Response from registry is: %v", err)

@@ -281,7 +281,7 @@ func (s *S2IBuilder) Build() error {

 			}

 			return errors.New(msg)

 		}

-		glog.V(1).Infof("Successfully pushed %s", pushTag)

+		glog.V(1).Infof("Push successful")

 	}

 	return nil

 }

deleted file mode 100644

@@ -1,223 +0,0 @@

-package pullprogress

-

-import (

-	"encoding/json"

-	"io"

-	"strings"

-	"time"

-)

-

-const (

-	defaultCountTimeThreshhold    = 10 * time.Second

-	defaultProgressTimeThreshhold = 45 * time.Second

-	defaultStableThreshhold       = 3

-)

-

-// NewPullProgressWriter creates a writer that periodically reports

-// on pull progress of an image. It only reports when the state of the

-// different layers has changed and uses time thresholds to limit the

-// rate of the reports.

-func NewPullProgressWriter(reportFn func(*ProgressReport)) io.Writer {

-	pipeIn, pipeOut := io.Pipe()

-	progressWriter := &pullProgressWriter{

-		Writer:                 pipeOut,

-		decoder:                json.NewDecoder(pipeIn),

-		layerStatus:            map[string]Status{},

-		reportFn:               reportFn,

-		countTimeThreshhold:    defaultCountTimeThreshhold,

-		progressTimeThreshhold: defaultProgressTimeThreshhold,

-		stableThreshhold:       defaultStableThreshhold,

-	}

-	go func() {

-		err := progressWriter.readProgress()

-		if err != nil {

-			pipeIn.CloseWithError(err)

-		}

-	}()

-	return progressWriter

-}

-

-// Status is a structure representation of a Docker pull progress line

-type Status struct {

-	ID             string `json:"id"`

-	Status         string `json:"status"`

-	ProgressDetail Detail `json:"progressDetail"`

-	Progress       string `json:"progress"`

-}

-

-// Detail is the progressDetail structure in a Docker pull progress line

-type Detail struct {

-	Current int64 `json:"current"`

-	Total   int64 `json:"total"`

-}

-

-// ProgressReport is a report of the progress of an image pull. It provides counts

-// of layers in a given state. It also provides a percentage of downloaded data

-// of those layers that are currently getting downloaded

-type ProgressReport struct {

-	Waiting     int

-	Downloading int

-	Extracting  int

-	Complete    int

-

-	DownloadPct float32

-}

-

-type statusType int

-

-const (

-	statusWaiting statusType = iota

-	statusDownloading

-	statusExtracting

-	statusComplete

-)

-

-type pullProgressWriter struct {

-	io.Writer

-	decoder                *json.Decoder

-	layerStatus            map[string]Status

-	lastLayerCount         int

-	stableLines            int

-	stableThreshhold       int

-	countTimeThreshhold    time.Duration

-	progressTimeThreshhold time.Duration

-	lastReport             *ProgressReport

-	lastReportTime         time.Time

-	reportFn               func(*ProgressReport)

-}

-

-func (w *pullProgressWriter) readProgress() error {

-	for {

-		status := &Status{}

-		err := w.decoder.Decode(status)

-		if err == io.EOF {

-			break

-		}

-		if err != nil {

-			return err

-		}

-		err = w.processStatus(status)

-		if err != nil {

-			return err

-		}

-	}

-	return nil

-}

-

-func (w *pullProgressWriter) isStableLayerCount() bool {

-	// If the number of layers has changed since last status, we're not stable

-	if w.lastLayerCount != len(w.layerStatus) {

-		w.lastLayerCount = len(w.layerStatus)

-		w.stableLines = 0

-		return false

-	}

-	// Only proceed after we've received status for the same number

-	// of layers at least 3 times. If not, they're still increasing

-	w.stableLines++

-	if w.stableLines < w.stableThreshhold {

-		// We're not stable enough yet

-		return false

-	}

-

-	return true

-}

-

-func (w *pullProgressWriter) processStatus(status *Status) error {

-	// determine if it's a status we want to process

-	if !isLayerStatus(status) {

-		return nil

-	}

-

-	w.layerStatus[status.ID] = *status

-

-	// if the number of layers has not stabilized yet, return and wait for more

-	// progress

-	if !w.isStableLayerCount() {

-		return nil

-	}

-

-	report := createProgressReport(w.layerStatus)

-

-	// check if the count of layers in each state has changed

-	if countsChanged(report, w.lastReport) {

-		// only report on changed counts if the change occurs after

-		// a predefined set of seconds (10 sec by default). This prevents

-		// multiple reports in rapid succession

-		if time.Since(w.lastReportTime) > w.countTimeThreshhold {

-			w.lastReport = report

-			w.lastReportTime = time.Now()

-			w.reportFn(report)

-		}

-	} else {

-		// If counts haven't changed, but enough time has passed (45 sec by default),

-		// at least report on download progress

-		if time.Since(w.lastReportTime) > w.progressTimeThreshhold {

-			w.lastReport = report

-			w.lastReportTime = time.Now()

-			w.reportFn(report)

-		}

-	}

-	return nil

-}

-

-func countsChanged(new, old *ProgressReport) bool {

-	if old == nil {

-		return true

-	}

-	return new.Waiting != old.Waiting ||

-		new.Downloading != old.Downloading ||

-		new.Extracting != old.Extracting ||

-		new.Complete != old.Complete

-}

-

-func layerStatusToPullStatus(str string) statusType {

-	switch str {

-	case "Downloading":

-		return statusDownloading

-	case "Extracting", "Verifying Checksum", "Download complete":

-		return statusExtracting

-	case "Pull complete", "Already exists":

-		return statusComplete

-	default: // "Pull fs layer" or "Waiting"

-		return statusWaiting

-	}

-}

-

-func isLayerStatus(status *Status) bool {

-	// ignore status lines with no layer id

-	if len(status.ID) == 0 {

-		return false

-	}

-	// ignore status lines with the initial named layer

-	if strings.HasPrefix(status.Status, "Pulling from") {

-		return false

-	}

-

-	return true

-}

-

-func createProgressReport(layerStatus map[string]Status) *ProgressReport {

-	report := &ProgressReport{}

-	var totalDownload, totalCurrent int64

-	for _, status := range layerStatus {

-		pullStatus := layerStatusToPullStatus(status.Status)

-		switch pullStatus {

-		case statusWaiting:

-			report.Waiting++

-		case statusDownloading:

-			report.Downloading++

-			totalDownload += status.ProgressDetail.Total

-			totalCurrent += status.ProgressDetail.Current

-		case statusExtracting:

-			report.Extracting++

-		case statusComplete:

-			report.Complete++

-		}

-	}

-	if totalDownload == 0 {

-		report.DownloadPct = 0

-	} else {

-		report.DownloadPct = float32(totalCurrent) / float32(totalDownload) * 100.0

-	}

-	return report

-}

@@ -17,6 +17,8 @@ import (

 	"github.com/fsouza/go-dockerclient/external/github.com/docker/docker/pkg/archive"

 	"github.com/fsouza/go-dockerclient/external/github.com/docker/docker/pkg/fileutils"

 	"github.com/golang/glog"

+

+	"github.com/openshift/origin/pkg/util/docker/dockerfile/builder/imageprogress"

 )

 // ClientExecutor can run Docker builds from a Docker client.

@@ -298,10 +300,22 @@ func (e *ClientExecutor) LoadImage(from string) (*docker.Image, error) {

 	}

 	var lastErr error

+	outputProgress := func(s string) {

+		e.LogFn("%s", s)

+	}

 	for _, config := range auth {

 		// TODO: handle IDs?

-		// TODO: use RawJSONStream:true and handle the output nicely

-		if err = e.Client.PullImage(docker.PullImageOptions{Repository: from, OutputStream: e.Out, Tag: tag}, config); err == nil {

+		pullImageOptions := docker.PullImageOptions{

+			Repository:    from,

+			Tag:           tag,

+			OutputStream:  imageprogress.NewPullWriter(outputProgress),

+			RawJSONStream: true,

+		}

+		if glog.V(5) {

+			pullImageOptions.OutputStream = os.Stderr

+			pullImageOptions.RawJSONStream = false

+		}

+		if err = e.Client.PullImage(pullImageOptions, config); err == nil {

 			break

 		}

 		lastErr = err

new file mode 100644

@@ -0,0 +1,234 @@

+package imageprogress

+

+import (

+	"encoding/json"

+	"io"

+	"regexp"

+	"time"

+)

+

+const (

+	defaultProgressTimeThreshhold = 30 * time.Second

+	defaultStableThreshhold       = 10

+)

+

+// progressLine is a structure representation of a Docker pull progress line

+type progressLine struct {

+	ID     string          `json:"id"`

+	Status string          `json:"status"`

+	Detail *progressDetail `json:"progressDetail"`

+}

+

+// progressDetail is the progressDetail structure in a Docker pull progress line

+type progressDetail struct {

+	Current int64 `json:"current"`

+	Total   int64 `json:"total"`

+}

+

+// layerDetail is layer information associated with a specific layerStatus

+type layerDetail struct {

+	Count   int

+	Current int64

+	Total   int64

+}

+

+// layerStatus is one of different possible status for layers detected by

+// the ProgressWriter

+type layerStatus int

+

+const (

+	statusPending layerStatus = iota

+	statusDownloading

+	statusExtracting

+	statusComplete

+	statusPushing

+)

+

+// layerStatusFromDockerString translates a string in a Docker status

+// line to a layerStatus

+func layerStatusFromDockerString(dockerStatus string) layerStatus {

+	switch dockerStatus {

+	case "Pushing":

+		return statusPushing

+	case "Downloading":

+		return statusDownloading

+	case "Extracting", "Verifying Checksum", "Download complete":

+		return statusExtracting

+	case "Pull complete", "Already exists", "Pushed":

+		return statusComplete

+	default:

+		return statusPending

+	}

+}

+

+type report map[layerStatus]*layerDetail

+

+func (r report) count(status layerStatus) int {

+	detail, ok := r[status]

+	if !ok {

+		return 0

+	}

+	return detail.Count

+}

+

+func (r report) percentProgress(status layerStatus) float32 {

+	detail, ok := r[status]

+	if !ok {

+		return 0

+	}

+	if detail.Total == 0 {

+		return 0

+	}

+	pct := float32(detail.Current) / float32(detail.Total) * 100.0

+	if pct > 100.0 {

+		pct = 100.0

+	}

+	return pct

+}

+

+func (r report) totalCount() int {

+	cnt := 0

+	for _, detail := range r {

+		cnt += detail.Count

+	}

+	return cnt

+}

+

+// newWriter creates a writer that periodically reports

+// on pull/push progress of a Docker image. It only reports when the state of the

+// different layers has changed and uses time thresholds to limit the

+// rate of the reports.

+func newWriter(reportFn func(report), layersChangedFn func(report, report) bool) io.Writer {

+	pipeIn, pipeOut := io.Pipe()

+	writer := &imageProgressWriter{

+		Writer:                 pipeOut,

+		decoder:                json.NewDecoder(pipeIn),

+		layerStatus:            map[string]progressLine{},

+		reportFn:               reportFn,

+		layersChangedFn:        layersChangedFn,

+		progressTimeThreshhold: defaultProgressTimeThreshhold,

+		stableThreshhold:       defaultStableThreshhold,

+	}

+	go func() {

+		err := writer.readProgress()

+		if err != nil {

+			pipeIn.CloseWithError(err)

+		}

+	}()

+	return writer

+}

+

+type imageProgressWriter struct {

+	io.Writer

+	decoder                *json.Decoder

+	layerStatus            map[string]progressLine

+	lastLayerCount         int

+	stableLines            int

+	stableThreshhold       int

+	progressTimeThreshhold time.Duration

+	lastReport             report

+	lastReportTime         time.Time

+	reportFn               func(report)

+	layersChangedFn        func(report, report) bool

+}

+

+func (w *imageProgressWriter) readProgress() error {

+	for {

+		line := &progressLine{}

+		err := w.decoder.Decode(line)

+		if err == io.EOF {

+			break

+		}

+		if err != nil {

+			return err

+		}

+		err = w.processLine(line)

+		if err != nil {

+			return err

+		}

+	}

+	return nil

+}

+

+func (w *imageProgressWriter) processLine(line *progressLine) error {

+	// determine if it's a line we want to process

+	if !islayerStatus(line) {

+		return nil

+	}

+

+	w.layerStatus[line.ID] = *line

+

+	// if the number of layers has not stabilized yet, return and wait for more

+	// progress

+	if !w.isStableLayerCount() {

+		return nil

+	}

+

+	r := createReport(w.layerStatus)

+

+	// check if the count of layers in each state has changed

+	if w.layersChangedFn(w.lastReport, r) {

+		w.lastReport = r

+		w.lastReportTime = time.Now()

+		w.reportFn(r)

+		return nil

+	}

+	// If layer counts haven't changed, but enough time has passed (30 sec by default),

+	// at least report on download/push progress

+	if time.Since(w.lastReportTime) > w.progressTimeThreshhold {

+		w.lastReport = r

+		w.lastReportTime = time.Now()

+		w.reportFn(r)

+	}

+	return nil

+}

+

+func (w *imageProgressWriter) isStableLayerCount() bool {

+	// If the number of layers has changed since last status, we're not stable

+	if w.lastLayerCount != len(w.layerStatus) {

+		w.lastLayerCount = len(w.layerStatus)

+		w.stableLines = 0

+		return false

+	}

+	// Only proceed after we've received status for the same number

+	// of layers at least stableThreshhold times. If not, they're still increasing

+	w.stableLines++

+	if w.stableLines < w.stableThreshhold {

+		// We're not stable enough yet

+		return false

+	}

+

+	return true

+}

+

+var layerIDRegexp = regexp.MustCompile("^[a-f,0-9]*$")

+

+func islayerStatus(line *progressLine) bool {

+	// ignore status lines with no layer id

+	if len(line.ID) == 0 {

+		return false

+	}

+	// ignore layer ids that are not hex string

+	if !layerIDRegexp.MatchString(line.ID) {

+		return false

+	}

+	return true

+}

+

+func createReport(dockerProgress map[string]progressLine) report {

+	r := report{}

+	for _, line := range dockerProgress {

+		layerStatus := layerStatusFromDockerString(line.Status)

+		detail, exists := r[layerStatus]

+		if !exists {

+			detail = &layerDetail{}

+			r[layerStatus] = detail

+		}

+		detail.Count++

+		if line.Detail != nil {

+			detail.Current += line.Detail.Current

+			detail.Total += line.Detail.Total

+		}

+	}

+	return r

+}

new file mode 100644

@@ -0,0 +1,45 @@

+package imageprogress

+

+import (

+	"fmt"

+	"io"

+)

+

+// NewPullWriter creates a writer that periodically reports

+// on pull progress of a Docker image. It only reports when the state of the

+// different layers has changed and uses time thresholds to limit the

+// rate of the reports.

+func NewPullWriter(printFn func(string)) io.Writer {

+	return newWriter(pullReporter(printFn), pullLayersChanged)

+}

+

+func pullReporter(printFn func(string)) func(report) {

+	extracting := false

+	return func(r report) {

+		if extracting {

+			return

+		}

+		if r.count(statusDownloading) == 0 &&

+			r.count(statusPending) == 0 &&

+			r.count(statusExtracting) > 0 {

+

+			printFn(fmt.Sprintf("Pulled %[1]d/%[1]d layers, 100%% complete", r.totalCount()))

+			printFn("Extracting")

+			extracting = true

+			return

+		}

+

+		completeCount := r.count(statusComplete) + r.count(statusExtracting)

+		var pctComplete float32 = 0.0

+		pctComplete += float32(completeCount) / float32(r.totalCount())

+		pctComplete += float32(r.count(statusDownloading)) / float32(r.totalCount()) * r.percentProgress(statusDownloading) / 100.0

+		pctComplete *= 100.0

+		printFn(fmt.Sprintf("Pulled %d/%d layers, %.0f%% complete", completeCount, r.totalCount(), pctComplete))

+	}

+}

+

+func pullLayersChanged(older, newer report) bool {

+	olderCompleteCount := older.count(statusComplete) + older.count(statusExtracting)

+	newerCompleteCount := newer.count(statusComplete) + newer.count(statusExtracting)

+	return olderCompleteCount != newerCompleteCount

+}

new file mode 100644

@@ -0,0 +1,29 @@

+package imageprogress

+

+import (

+	"fmt"

+	"io"

+)

+

+// NewPushWriter creates a writer that periodically reports

+// on push progress of a Docker image. It only reports when the state of the

+// different layers has changed and uses time thresholds to limit the

+// rate of the reports.

+func NewPushWriter(printFn func(string)) io.Writer {

+	return newWriter(pushReporter(printFn), pushLayersChanged)

+}

+

+func pushReporter(printFn func(string)) func(report) {

+	return func(r report) {

+		var pctComplete float32 = 0.0

+		pctComplete += float32(r.count(statusComplete)) / float32(r.totalCount())

+		pctComplete += float32(r.count(statusPushing)) / float32(r.totalCount()) * r.percentProgress(statusPushing) / 100.0

+		pctComplete *= 100.0

+

+		printFn(fmt.Sprintf("Pushed %d/%d layers, %.0f%% complete", r.count(statusComplete), r.totalCount(), pctComplete))

+	}

+}

+

+func pushLayersChanged(older, newer report) bool {

+	return older.count(statusComplete) != newer.count(statusComplete)

+}