pkg/progressui: customization callbacks

Signed-off-by: Andrea Luzzardi <aluzzardi@gmail.com>

.golangci.yml

@@ -1,6 +1,10 @@
+run:
+  skip-dirs:
+    # progressui is a modified 3rd party library from buildkit
+    - pkg/progressui
+
 linters:
   disable-all: true
-  timeout: 30m
   enable:
     - bodyclose
     - deadcode

pkg/progressui/display.go

@@ -20,21 +20,90 @@ import (
 	"golang.org/x/time/rate"
 )
 
-func DisplaySolveStatus(ctx context.Context, phase string, c console.Console, w io.Writer, ch chan *client.SolveStatus) error {
+const (
+	defaultTickerTimeout  = 150 * time.Millisecond
+	defaultDisplayTimeout = 100 * time.Millisecond
+)
 
+type VertexPrintFunc func(v *client.Vertex, index int)
+type StatusPrintFunc func(v *client.Vertex, format string, a ...interface{})
+type LogPrintFunc func(v *client.Vertex, stream int, partial bool, format string, a ...interface{})
+
+func PrintSolveStatus(ctx context.Context, ch chan *client.SolveStatus, vertexPrintCb VertexPrintFunc, statusPrintCb StatusPrintFunc, logPrintCb LogPrintFunc) error {
+	printer := &textMux{
+		vertexPrintCb: vertexPrintCb,
+		statusPrintCb: statusPrintCb,
+		logPrintCb:    logPrintCb,
+	}
+
+	t := newTrace(false)
+
+	var done bool
+	ticker := time.NewTicker(defaultTickerTimeout)
+	defer ticker.Stop()
+
+	displayLimiter := rate.NewLimiter(rate.Every(defaultDisplayTimeout), 1)
+
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-ticker.C:
+		case ss, ok := <-ch:
+			if ok {
+				t.update(ss, 80)
+			} else {
+				done = true
+			}
+		}
+
+		if done || displayLimiter.Allow() {
+			printer.print(t)
+			if done {
+				t.printErrorLogs(statusPrintCb)
+				return nil
+			}
+			ticker.Stop()
+			ticker = time.NewTicker(defaultTickerTimeout)
+		}
+	}
+}
+
+func DisplaySolveStatus(ctx context.Context, phase string, c console.Console, w io.Writer, ch chan *client.SolveStatus) error {
 	modeConsole := c != nil
 
-	disp := &display{c: c, phase: phase}
+	if !modeConsole {
-	printer := &textMux{w: w}
+		vertexPrintCb := func(v *client.Vertex, index int) {
+			if os.Getenv("PROGRESS_NO_TRUNC") == "0" {
+				fmt.Fprintf(w, "#%d %s\n", index, limitString(v.Name, 72))
+			} else {
+				fmt.Fprintf(w, "#%d %s\n", index, v.Name)
+				fmt.Fprintf(w, "#%d %s\n", index, v.Digest)
+			}
+		}
+		statusPrintCb := func(v *client.Vertex, format string, a ...interface{}) {
+			fmt.Fprintf(w, fmt.Sprintf("%s\n", format), a...)
+		}
+		logPrintCb := func(v *client.Vertex, stream int, partial bool, format string, a ...interface{}) {
+			if partial {
+				fmt.Fprintf(w, format, a...)
+			} else {
+				fmt.Fprintf(w, fmt.Sprintf("%s\n", format), a...)
+			}
+		}
 
+		return PrintSolveStatus(ctx, ch, vertexPrintCb, statusPrintCb, logPrintCb)
+	}
+
+	disp := &display{c: c, phase: phase}
 	if disp.phase == "" {
 		disp.phase = "Building"
 	}
 
-	t := newTrace(w, modeConsole)
+	t := newTrace(true)
 
-	tickerTimeout := 150 * time.Millisecond
+	tickerTimeout := defaultTickerTimeout
-	displayTimeout := 100 * time.Millisecond
+	displayTimeout := defaultDisplayTimeout
 
 	if v := os.Getenv("TTY_DISPLAY_RATE"); v != "" {
 		if r, err := strconv.ParseInt(v, 10, 64); err == nil {
@@ -64,28 +133,18 @@ func DisplaySolveStatus(ctx context.Context, phase string, c console.Console, w
 			}
 		}
 
-		if modeConsole {
 		width, height = disp.getSize()
 		if done {
 			disp.print(t.displayInfo(), width, height, true)
-				t.printErrorLogs(c)
+			t.printErrorLogs(func(v *client.Vertex, format string, a ...interface{}) {
+				fmt.Fprintf(w, format, a...)
+			})
 			return nil
 		} else if displayLimiter.Allow() {
 			ticker.Stop()
 			ticker = time.NewTicker(tickerTimeout)
 			disp.print(t.displayInfo(), width, height, false)
 		}
-		} else {
-			if done || displayLimiter.Allow() {
-				printer.print(t)
-				if done {
-					t.printErrorLogs(w)
-					return nil
-				}
-				ticker.Stop()
-				ticker = time.NewTicker(tickerTimeout)
-			}
-		}
 	}
 }
 
@@ -111,7 +170,6 @@ type job struct {
 }
 
 type trace struct {
-	w             io.Writer
 	localTimeDiff time.Duration
 	vertexes      []*vertex
 	byDigest      map[digest.Digest]*vertex
@@ -120,6 +178,12 @@ type trace struct {
 	modeConsole   bool
 }
 
+type log struct {
+	index  int
+	line   []byte
+	stream int
+}
+
 type vertex struct {
 	*client.Vertex
 	statuses []*status
@@ -127,7 +191,7 @@ type vertex struct {
 	indent   string
 	index    int
 
-	logs          [][]byte
+	logs          []log
 	logsPartial   bool
 	logsOffset    int
 	prev          *client.Vertex
@@ -156,11 +220,10 @@ type status struct {
 	*client.VertexStatus
 }
 
-func newTrace(w io.Writer, modeConsole bool) *trace {
+func newTrace(modeConsole bool) *trace {
 	return &trace{
 		byDigest:    make(map[digest.Digest]*vertex),
 		updates:     make(map[digest.Digest]struct{}),
-		w:           w,
 		modeConsole: modeConsole,
 	}
 }
@@ -252,6 +315,7 @@ func (t *trace) update(s *client.SolveStatus, termWidth int) {
 		v.update(1)
 	}
 	for _, l := range s.Logs {
+		l := l
 		v, ok := t.byDigest[l.Vertex]
 		if !ok {
 			continue // shouldn't happen
@@ -266,8 +330,8 @@ func (t *trace) update(s *client.SolveStatus, termWidth int) {
 		}
 		i := 0
 		complete := split(l.Data, byte('\n'), func(dt []byte) {
-			if v.logsPartial && len(v.logs) != 0 && i == 0 {
+			if v.logsPartial && len(v.logs) != 0 && i == 0 && v.logs[len(v.logs)-1].stream == l.Stream {
-				v.logs[len(v.logs)-1] = append(v.logs[len(v.logs)-1], dt...)
+				v.logs[len(v.logs)-1].line = append(v.logs[len(v.logs)-1].line, dt...)
 			} else {
 				ts := time.Duration(0)
 				if v.Started != nil {
@@ -280,7 +344,11 @@ func (t *trace) update(s *client.SolveStatus, termWidth int) {
 				} else if sec < 100 {
 					prec = 2
 				}
-				v.logs = append(v.logs, []byte(fmt.Sprintf("#%d %s %s", v.index, fmt.Sprintf("%.[2]*[1]f", sec, prec), dt)))
+				v.logs = append(v.logs, log{
+					line:   []byte(fmt.Sprintf("#%d %s %s", v.index, fmt.Sprintf("%.[2]*[1]f", sec, prec), dt)),
+					stream: l.Stream,
+					index:  v.index,
+				})
 			}
 			i++
 		})
@@ -290,16 +358,15 @@ func (t *trace) update(s *client.SolveStatus, termWidth int) {
 	}
 }
 
-func (t *trace) printErrorLogs(f io.Writer) {
+func (t *trace) printErrorLogs(printCb StatusPrintFunc) {
 	for _, v := range t.vertexes {
 		if v.Error != "" && !strings.HasSuffix(v.Error, context.Canceled.Error()) {
-			fmt.Fprintln(f, "------")
+			printCb(v.Vertex, "------")
-			fmt.Fprintf(f, " > %s:\n", v.Name)
+			printCb(v.Vertex, " > %s:", v.Name)
 			for _, l := range v.logs {
-				f.Write(l)
+				printCb(v.Vertex, "%s", l.line)
-				fmt.Fprintln(f)
 			}
-			fmt.Fprintln(f, "------")
+			printCb(v.Vertex, "------")
 		}
 	}
 }
@@ -307,7 +374,7 @@ func (t *trace) printErrorLogs(f io.Writer) {
 func (t *trace) displayInfo() (d displayInfo) {
 	d.startTime = time.Now()
 	if t.localTimeDiff != 0 {
-		d.startTime = (*t.vertexes[0].Started).Add(t.localTimeDiff)
+		d.startTime = t.vertexes[0].Started.Add(t.localTimeDiff)
 	}
 	d.countTotal = len(t.byDigest)
 	for _, v := range t.byDigest {
@@ -325,7 +392,7 @@ func (t *trace) displayInfo() (d displayInfo) {
 		j := &job{
 			startTime:     addTime(v.Started, t.localTimeDiff),
 			completedTime: addTime(v.Completed, t.localTimeDiff),
-			name:          strings.Replace(v.Name, "\t", " ", -1),
+			name:          strings.ReplaceAll(v.Name, "\t", " "),
 			vertex:        v,
 		}
 		if v.Error != "" {
@@ -385,7 +452,7 @@ func addTime(tm *time.Time, d time.Duration) *time.Time {
 	if tm == nil {
 		return nil
 	}
-	t := (*tm).Add(d)
+	t := tm.Add(d)
 	return &t
 }
 
@@ -580,7 +647,8 @@ func wrapHeight(j []*job, limit int) []*job {
 				l--
 			}
 			freespace := len(wrapped) - len(rewrapped)
-			wrapped = append(invisible[len(invisible)-freespace:], rewrapped...)
+			invisible = append(invisible[len(invisible)-freespace:], rewrapped...)
+			wrapped = invisible
 		}
 	}
 	return wrapped

pkg/progressui/printer.go

@@ -3,8 +3,6 @@ package progressui
 import (
 	"context"
 	"fmt"
-	"io"
-	"os"
 	"sort"
 	"strings"
 	"time"
@@ -24,7 +22,10 @@ type lastStatus struct {
 }
 
 type textMux struct {
-	w        io.Writer
+	vertexPrintCb VertexPrintFunc
+	statusPrintCb StatusPrintFunc
+	logPrintCb    LogPrintFunc
+
 	current  digest.Digest
 	last     map[string]lastStatus
 	notFirst bool
@@ -44,33 +45,27 @@ func (p *textMux) printVtx(t *trace, dgst digest.Digest) {
 		if p.current != "" {
 			old := t.byDigest[p.current]
 			if old.logsPartial {
-				fmt.Fprintln(p.w, "")
+				p.statusPrintCb(v.Vertex, "")
 			}
 			old.logsOffset = 0
 			old.count = 0
-			fmt.Fprintf(p.w, "#%d ...\n", old.index)
+			p.statusPrintCb(v.Vertex, "#%d ...", old.index)
 		}
 
 		if p.notFirst {
-			fmt.Fprintln(p.w, "")
+			p.statusPrintCb(v.Vertex, "")
 		} else {
 			p.notFirst = true
 		}
 
-		if os.Getenv("PROGRESS_NO_TRUNC") == "0" {
+		p.vertexPrintCb(v.Vertex, v.index)
-			fmt.Fprintf(p.w, "#%d %s\n", v.index, limitString(v.Name, 72))
-		} else {
-			fmt.Fprintf(p.w, "#%d %s\n", v.index, v.Name)
-			fmt.Fprintf(p.w, "#%d %s\n", v.index, v.Digest)
-		}
-
 	}
 
 	if len(v.events) != 0 {
 		v.logsOffset = 0
 	}
 	for _, ev := range v.events {
-		fmt.Fprintf(p.w, "#%d %s\n", v.index, ev)
+		p.statusPrintCb(v.Vertex, "#%d %s", v.index, ev)
 	}
 	v.events = v.events[:0]
 
@@ -118,25 +113,24 @@ func (p *textMux) printVtx(t *trace, dgst digest.Digest) {
 			if s.Completed != nil {
 				tm += " done"
 			}
-			fmt.Fprintf(p.w, "#%d %s%s%s\n", v.index, s.ID, bytes, tm)
+			p.statusPrintCb(v.Vertex, "#%d %s%s%s", v.index, s.ID, bytes, tm)
 		}
 	}
 	v.statusUpdates = map[string]struct{}{}
 
 	for i, l := range v.logs {
+		line := l.line
 		if i == 0 {
-			l = l[v.logsOffset:]
+			line = line[v.logsOffset:]
-		}
-		fmt.Fprintf(p.w, "%s", []byte(l))
-		if i != len(v.logs)-1 || !v.logsPartial {
-			fmt.Fprintln(p.w, "")
 		}
+		complete := i != len(v.logs)-1 || !v.logsPartial
+		p.logPrintCb(v.Vertex, l.stream, !complete, "%s", line)
 	}
 
 	if len(v.logs) > 0 {
 		if v.logsPartial {
 			v.logs = v.logs[len(v.logs)-1:]
-			v.logsOffset = len(v.logs[0])
+			v.logsOffset = len(v.logs[0].line)
 		} else {
 			v.logs = nil
 			v.logsOffset = 0
@@ -150,23 +144,22 @@ func (p *textMux) printVtx(t *trace, dgst digest.Digest) {
 
 		if v.Error != "" {
 			if v.logsPartial {
-				fmt.Fprintln(p.w, "")
+				p.statusPrintCb(v.Vertex, "")
 			}
 			if strings.HasSuffix(v.Error, context.Canceled.Error()) {
-				fmt.Fprintf(p.w, "#%d CANCELED\n", v.index)
+				p.statusPrintCb(v.Vertex, "#%d CANCELED", v.index)
 			} else {
-				fmt.Fprintf(p.w, "#%d ERROR: %s\n", v.index, v.Error)
+				p.statusPrintCb(v.Vertex, "#%d ERROR: %s", v.index, v.Error)
 			}
 		} else if v.Cached {
-			fmt.Fprintf(p.w, "#%d CACHED\n", v.index)
+			p.statusPrintCb(v.Vertex, "#%d CACHED", v.index)
 		} else {
 			tm := ""
 			if v.Started != nil {
 				tm = fmt.Sprintf(" %.1fs", v.Completed.Sub(*v.Started).Seconds())
 			}
-			fmt.Fprintf(p.w, "#%d DONE%s\n", v.index, tm)
+			p.statusPrintCb(v.Vertex, "#%d DONE%s", v.index, tm)
 		}
-
 	}
 
 	delete(t.updates, dgst)