Merge branch 'master' into add_last_version-feature

utils/tarsum.go

@@ -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
+}

utils/utils.go

@@ -248,30 +248,54 @@ func (r *bufReader) Close() error {
 
 type WriteBroadcaster struct {
 	sync.Mutex
-	writers map[io.WriteCloser]struct{}
+	buf     *bytes.Buffer
+	writers map[StreamWriter]bool
 }
 
-func (w *WriteBroadcaster) AddWriter(writer io.WriteCloser) {
+type StreamWriter struct {
+	wc     io.WriteCloser
+	stream string
+}
+
+func (w *WriteBroadcaster) AddWriter(writer io.WriteCloser, stream string) {
 	w.Lock()
-	w.writers[writer] = struct{}{}
+	sw := StreamWriter{wc: writer, stream: stream}
+	w.writers[sw] = true
 	w.Unlock()
 }
 
-// FIXME: Is that function used?
-// FIXME: This relies on the concrete writer type used having equality operator
-func (w *WriteBroadcaster) RemoveWriter(writer io.WriteCloser) {
-	w.Lock()
-	delete(w.writers, writer)
-	w.Unlock()
+type JSONLog struct {
+	Log     string    `json:"log,omitempty"`
+	Stream  string    `json:"stream,omitempty"`
+	Created time.Time `json:"time"`
 }
 
 func (w *WriteBroadcaster) Write(p []byte) (n int, err error) {
 	w.Lock()
 	defer w.Unlock()
-	for writer := range w.writers {
-		if n, err := writer.Write(p); err != nil || n != len(p) {
+	w.buf.Write(p)
+	for sw := range w.writers {
+		lp := p
+		if sw.stream != "" {
+			lp = nil
+			for {
+				line, err := w.buf.ReadString('\n')
+				if err != nil {
+					w.buf.Write([]byte(line))
+					break
+				}
+				b, err := json.Marshal(&JSONLog{Log: line, Stream: sw.stream, Created: time.Now()})
+				if err != nil {
+					// On error, evict the writer
+					delete(w.writers, sw)
+					continue
+				}
+				lp = append(lp, b...)
+			}
+		}
+		if n, err := sw.wc.Write(lp); err != nil || n != len(lp) {
 			// On error, evict the writer
-			delete(w.writers, writer)
+			delete(w.writers, sw)
 		}
 	}
 	return len(p), nil
@@ -280,15 +304,15 @@ func (w *WriteBroadcaster) Write(p []byte) (n int, err error) {
 func (w *WriteBroadcaster) CloseWriters() error {
 	w.Lock()
 	defer w.Unlock()
-	for writer := range w.writers {
-		writer.Close()
+	for sw := range w.writers {
+		sw.wc.Close()
 	}
-	w.writers = make(map[io.WriteCloser]struct{})
+	w.writers = make(map[StreamWriter]bool)
 	return nil
 }
 
 func NewWriteBroadcaster() *WriteBroadcaster {
-	return &WriteBroadcaster{writers: make(map[io.WriteCloser]struct{})}
+	return &WriteBroadcaster{writers: make(map[StreamWriter]bool), buf: bytes.NewBuffer(nil)}
 }
 
 func GetTotalUsedFds() int {
@@ -587,6 +611,24 @@ type JSONMessage struct {
 	Status   string `json:"status,omitempty"`
 	Progress string `json:"progress,omitempty"`
 	Error    string `json:"error,omitempty"`
+	ID       string `json:"id,omitempty"`
+	Time     int64  `json:"time,omitempty"`
+}
+
+func (jm *JSONMessage) Display(out io.Writer) error {
+	if jm.Time != 0 {
+		fmt.Fprintf(out, "[%s] ", time.Unix(jm.Time, 0))
+	}
+	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 {

utils/utils_test.go

@@ -60,9 +60,9 @@ func TestWriteBroadcaster(t *testing.T) {
 
 	// Test 1: Both bufferA and bufferB should contain "foo"
 	bufferA := &dummyWriter{}
-	writer.AddWriter(bufferA)
+	writer.AddWriter(bufferA, "")
 	bufferB := &dummyWriter{}
-	writer.AddWriter(bufferB)
+	writer.AddWriter(bufferB, "")
 	writer.Write([]byte("foo"))
 
 	if bufferA.String() != "foo" {
@@ -76,7 +76,7 @@ func TestWriteBroadcaster(t *testing.T) {
 	// Test2: bufferA and bufferB should contain "foobar",
 	// while bufferC should only contain "bar"
 	bufferC := &dummyWriter{}
-	writer.AddWriter(bufferC)
+	writer.AddWriter(bufferC, "")
 	writer.Write([]byte("bar"))
 
 	if bufferA.String() != "foobar" {
@@ -91,35 +91,22 @@ func TestWriteBroadcaster(t *testing.T) {
 		t.Errorf("Buffer contains %v", bufferC.String())
 	}
 
-	// Test3: Test removal
-	writer.RemoveWriter(bufferB)
-	writer.Write([]byte("42"))
-	if bufferA.String() != "foobar42" {
-		t.Errorf("Buffer contains %v", bufferA.String())
-	}
-	if bufferB.String() != "foobar" {
-		t.Errorf("Buffer contains %v", bufferB.String())
-	}
-	if bufferC.String() != "bar42" {
-		t.Errorf("Buffer contains %v", bufferC.String())
-	}
-
-	// Test4: Test eviction on failure
+	// Test3: Test eviction on failure
 	bufferA.failOnWrite = true
 	writer.Write([]byte("fail"))
-	if bufferA.String() != "foobar42" {
+	if bufferA.String() != "foobar" {
 		t.Errorf("Buffer contains %v", bufferA.String())
 	}
-	if bufferC.String() != "bar42fail" {
+	if bufferC.String() != "barfail" {
 		t.Errorf("Buffer contains %v", bufferC.String())
 	}
 	// Even though we reset the flag, no more writes should go in there
 	bufferA.failOnWrite = false
 	writer.Write([]byte("test"))
-	if bufferA.String() != "foobar42" {
+	if bufferA.String() != "foobar" {
 		t.Errorf("Buffer contains %v", bufferA.String())
 	}
-	if bufferC.String() != "bar42failtest" {
+	if bufferC.String() != "barfailtest" {
 		t.Errorf("Buffer contains %v", bufferC.String())
 	}
 
@@ -141,7 +128,7 @@ func TestRaceWriteBroadcaster(t *testing.T) {
 	writer := NewWriteBroadcaster()
 	c := make(chan bool)
 	go func() {
-		writer.AddWriter(devNullCloser(0))
+		writer.AddWriter(devNullCloser(0), "")
 		c <- true
 	}()
 	writer.Write([]byte("hello"))