wip

api/swagger-spec/openshift-openapi-spec.json

@@ -48201,6 +48201,24 @@
      }
     }
    },
+   "v1.BuildStatusOutput": {
+    "description": "BuildStatusOutput contains the status of the built image.",
+    "properties": {
+     "to": {
+      "description": "to describes the status of the built image being pushed to a registry.",
+      "$ref": "#/definitions/v1.BuildStatusOutputTo"
+     }
+    }
+   },
+   "v1.BuildStatusOutputTo": {
+    "description": "BuildStatusOutputTo describes the status of the built image with regards to image registry to which it was supposed to be pushed.",
+    "properties": {
+     "imageDigest": {
+      "description": "imageDigest is the digest of the built Docker image. The digest uniquely identifies the image in the registry to which it was pushed. Please note that this field may not always set even if the push completes successfully.",
+      "type": "string"
+     }
+    }
+   },
    "v1.BuildStrategy": {
     "description": "BuildStrategy contains the details of how to perform a build.",
     "required": [
@@ -48229,24 +48247,6 @@
      }
     }
    },
-   "v1.BuildStatusOutput": {
-    "description": "BuildStatusOutput contains the status of the built image.",
-    "properties": {
-     "to": {
-      "description": "to describes the status of the built image being pushed to a registry.",
-      "$ref": "#/definitions/v1.BuildStatusOutputTo"
-     }
-    }
-   },
-   "v1.BuildStatusOutputTo": {
-    "description": "BuildStatusOutputTo describes the status of the built image with regards to image registry to which it was supposed to be pushed.",
-    "properties": {
-     "imageDigest": {
-      "description": "imageDigest is the digest of the built Docker image. The digest uniquely identifies the image in the registry to which it was pushed. Please note that this field may not always set even if the push completes successfully.",
-      "type": "string"
-     }
-    }
-   },
    "v1.BuildTriggerCause": {
     "description": "BuildTriggerCause holds information about a triggered build. It is used for displaying build trigger data for each build and build configuration in oc describe. It is also used to describe which triggers led to the most recent update in the build configuration.",
     "properties": {

pkg/build/api/types.go

@@ -341,7 +341,7 @@ const (
 // BuildStatusOutput contains the status of the built image.
 type BuildStatusOutput struct {
 	// To describes the status of the built image being pushed to a registry.
-	To *BuildStatusOutputTo `json:"to,omitempty" protobuf:"bytes,1,opt,name=to"`
+	To *BuildStatusOutputTo
 }
 
 // BuildStatusOutputTo describes the status of the built image with regards to
@@ -351,7 +351,7 @@ type BuildStatusOutputTo struct {
 	// identifies the image in the registry to which it was pushed. Please note
 	// that this field may not always set even if the push completes
 	// successfully.
-	ImageDigest string `json:"imageDigest,omitempty" protobuf:"bytes,1,opt,name=imageDigest"`
+	ImageDigest string
 }
 
 // BuildSource is the input used for the build.

pkg/build/api/v1/zz_generated.conversion.go

@@ -806,15 +806,7 @@ func Convert_api_BuildStatus_To_v1_BuildStatus(in *api.BuildStatus, out *BuildSt
 }
 
 func autoConvert_v1_BuildStatusOutput_To_api_BuildStatusOutput(in *BuildStatusOutput, out *api.BuildStatusOutput, s conversion.Scope) error {
-	if in.To != nil {
-		in, out := &in.To, &out.To
-		*out = new(api.BuildStatusOutputTo)
-		if err := Convert_v1_BuildStatusOutputTo_To_api_BuildStatusOutputTo(*in, *out, s); err != nil {
-			return err
-		}
-	} else {
-		out.To = nil
-	}
+	out.To = (*api.BuildStatusOutputTo)(unsafe.Pointer(in.To))
 	return nil
 }
 
@@ -823,15 +815,7 @@ func Convert_v1_BuildStatusOutput_To_api_BuildStatusOutput(in *BuildStatusOutput
 }
 
 func autoConvert_api_BuildStatusOutput_To_v1_BuildStatusOutput(in *api.BuildStatusOutput, out *BuildStatusOutput, s conversion.Scope) error {
-	if in.To != nil {
-		in, out := &in.To, &out.To
-		*out = new(BuildStatusOutputTo)
-		if err := Convert_api_BuildStatusOutputTo_To_v1_BuildStatusOutputTo(*in, *out, s); err != nil {
-			return err
-		}
-	} else {
-		out.To = nil
-	}
+	out.To = (*BuildStatusOutputTo)(unsafe.Pointer(in.To))
 	return nil
 }
 

pkg/build/builder/dockerutil.go

@@ -3,11 +3,12 @@ package builder
 import (
 	"bytes"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
+	"io/ioutil"
 	"os"
 	"strings"
-	"sync"
 	"time"
 
 	docker "github.com/fsouza/go-dockerclient"
@@ -81,16 +82,14 @@ func pushImage(client DockerClient, name string, authConfig docker.AuthConfigura
 
 	var progressWriter io.Writer
 	if glog.Is(5) {
-		simpleWriter := newSimpleWriter(os.Stderr)
-		defer simpleWriter.Close()
-		progressWriter = simpleWriter
+		progressWriter = newSimpleWriter(os.Stderr)
 	} else {
 		logProgress := func(s string) {
 			glog.V(0).Infof("%s", s)
 		}
 		progressWriter = imageprogress.NewPushWriter(logProgress)
 	}
-	digestWriter := &digestWriter{}
+	digestWriter := newDigestWriter()
 
 	opts := docker.PushImageOptions{
 		Name:          repository,
@@ -104,7 +103,7 @@ func pushImage(client DockerClient, name string, authConfig docker.AuthConfigura
 	for retries := 0; retries <= DefaultPushRetryCount; retries++ {
 		err = client.PushImage(opts, authConfig)
 		if err == nil {
-			return digestWriter.Digest(), nil
+			return digestWriter.Digest, nil
 		}
 
 		errMsg := fmt.Sprintf("%s", err)
@@ -235,110 +234,99 @@ type progressAux struct {
 	Size   int64  `json:"Size"`
 }
 
-// digestWriter consumes stream of json messages from docker client push
-// operation and looks for digest of the pushed image.
-type digestWriter struct {
-	bytes.Buffer
-	digest string
+type pushWriterCallback func(progressLine) error
+
+// pushWriter is an io.Writer which consumes a stream of json messages returned
+// by docker client when it pushes image to registry. It calls the provided
+// callback for each decoded JSON object.
+type pushWriter struct {
+	buf      *bytes.Buffer
+	callback pushWriterCallback
 }
 
-// Digest returns the digest of the pushed image. Call this method after all
-// jsons messages have been written to the digestWriter.
-func (d *digestWriter) Digest() string {
-	if len(d.digest) > 0 {
-		return d.digest
+func newPushWriter(cb pushWriterCallback) *pushWriter {
+	return &pushWriter{
+		buf:      &bytes.Buffer{},
+		callback: cb,
 	}
+}
 
-	decoder := json.NewDecoder(d)
-	for decoder.More() {
+func (t *pushWriter) Write(data []byte) (int, error) {
+	n, err := t.buf.Write(data)
+	if err != nil {
+		return n, err
+	}
+	dec := json.NewDecoder(t.buf)
+
+	for {
+		// save the not yet parsed input so we can restore it in case it
+		// contains part of valid JSON
+		savedBuf, err := ioutil.ReadAll(dec.Buffered())
+		savedBuf = append(savedBuf, t.buf.Bytes()...)
+
+		// try decoding a value
 		line := &progressLine{}
-		err := decoder.Decode(line)
-		if err != nil {
-			break
-		}
+		err = dec.Decode(line)
 
-		if len(line.Aux.Digest) > 0 {
-			d.digest = line.Aux.Digest
-			break
+		switch err {
+		// decoded a value, pass it to callback
+		case nil:
+			if callbackErr := t.callback(*line); callbackErr != nil {
+				return n, callbackErr
+			}
+		// no more values
+		case io.EOF:
+			return n, nil
+		// there's no whole JSON but we consumed bytes that might be part of
+		// one - restore the saved buffer
+		case io.ErrUnexpectedEOF:
+			t.buf = bytes.NewBuffer(savedBuf)
+			return n, nil
+		// actual error happened
+		default:
+			return n, err
 		}
 	}
-
-	return d.digest
 }
 
 // simpleProgressWriter is an io.Writer which consumes a stream of json
 // messages returned by docker client when it pushes image to registry. It
 // writes simple human-readable indication of the push progress to the output
 // io.Writer. The output format mimics what go-dockerclient writes when called
 // with RawJSONStream=false.
-type simpleProgressWriter struct {
-	mutex          *sync.Mutex
-	internalWriter *io.PipeWriter
-	output         io.Writer
-}
-
-func newSimpleWriter(output io.Writer) *simpleProgressWriter {
-	writer := &simpleProgressWriter{
-		mutex:  &sync.Mutex{},
-		output: output,
-	}
-	return writer
-}
-
-func (w *simpleProgressWriter) Write(data []byte) (int, error) {
-	w.mutex.Lock()
-	defer w.mutex.Unlock()
-	if w.internalWriter == nil {
-		var pipeIn *io.PipeReader
-		pipeIn, w.internalWriter = io.Pipe()
-		decoder := json.NewDecoder(pipeIn)
-		go func() {
-			err := w.readProgress(decoder)
-			if err != nil {
-				pipeIn.CloseWithError(err)
-			}
-		}()
-	}
-	return w.internalWriter.Write(data)
-}
-
-func (w *simpleProgressWriter) readProgress(decoder *json.Decoder) error {
-	for {
-		line := &progressLine{}
-		err := decoder.Decode(line)
-		if err == io.EOF {
-			break
-		}
-		if err != nil {
-			return err
+func newSimpleWriter(output io.Writer) io.Writer {
+	return newPushWriter(func(line progressLine) error {
+		if line.Stream != "" {
+			fmt.Fprint(output, line.Stream)
+		} else if line.Progress != "" {
+			fmt.Fprintf(output, "%s %s\r", line.Status, line.Progress)
+		} else if line.Error != "" {
+			return errors.New(line.Error)
 		}
-		err = w.processLine(line)
-		if err != nil {
-			return err
+		if line.Status != "" {
+			fmt.Fprintln(output, line.Status)
 		}
-	}
-	return nil
+		return nil
+	})
 }
 
-func (w *simpleProgressWriter) processLine(line *progressLine) error {
-	if line.Stream != "" {
-		fmt.Fprint(w.output, line.Stream)
-	} else if line.Progress != "" {
-		fmt.Fprintf(w.output, "%s %s\r", line.Status, line.Progress)
-	} else if line.Error != "" {
-		fmt.Fprintf(w.output, "error: %s\r", line.Error)
-	}
-	if line.Status != "" {
-		fmt.Fprintln(w.output, line.Status)
-	}
-	return nil
+// digestWriter consumes stream of json messages from docker client push
+// operation and looks for digest of the pushed image.
+type digestWriter struct {
+	*pushWriter
+	Digest string
 }
 
-func (w *simpleProgressWriter) Close() error {
-	w.mutex.Lock()
-	defer w.mutex.Unlock()
-	if w.internalWriter != nil {
-		return w.internalWriter.Close()
-	}
-	return nil
+func newDigestWriter() *digestWriter {
+	dw := digestWriter{}
+	dw.pushWriter = newPushWriter(func(line progressLine) error {
+		if line.Error != "" {
+			return errors.New(line.Error)
+		}
+		if len(dw.Digest) == 0 && len(line.Aux.Digest) > 0 {
+			dw.Digest = line.Aux.Digest
+		}
+		return nil
+	})
+	return &dw
 }