logger: TTY logs support (live update)

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

cmd/dagger/cmd/root.go

@@ -24,7 +24,7 @@ var rootCmd = &cobra.Command{
 }
 
 func init() {
-	rootCmd.PersistentFlags().String("log-format", "", "Log format (json, pretty). Defaults to json if the terminal is not a tty")
+	rootCmd.PersistentFlags().String("log-format", "auto", "Log format (auto, plain, tty, json)")
 	rootCmd.PersistentFlags().StringP("log-level", "l", "info", "Log level")
 
 	rootCmd.PersistentFlags().Bool("no-cache", false, "Disable caching")

cmd/dagger/cmd/up.go

@@ -31,7 +31,23 @@ var upCmd = &cobra.Command{
 		}
 	},
 	Run: func(cmd *cobra.Command, args []string) {
-		lg := logger.New()
+		var (
+			lg  = logger.New()
+			tty *logger.TTYOutput
+			err error
+		)
+
+		if f := viper.GetString("log-format"); f == "tty" || f == "auto" && term.IsTerminal(int(os.Stdout.Fd())) {
+			tty, err = logger.NewTTYOutput(os.Stderr)
+			if err != nil {
+				lg.Fatal().Err(err).Msg("failed to initialize TTY logger")
+			}
+			tty.Start()
+			defer tty.Stop()
+
+			lg = lg.Output(tty)
+		}
+
 		ctx := lg.WithContext(cmd.Context())
 
 		project := common.CurrentProject(ctx)
@@ -45,7 +61,7 @@ var upCmd = &cobra.Command{
 
 		cl := common.NewClient(ctx)
 
-		err := cl.Do(ctx, st, func(ctx context.Context, env *environment.Environment, s solver.Solver) error {
+		err = cl.Do(ctx, st, func(ctx context.Context, env *environment.Environment, s solver.Solver) error {
 			// check that all inputs are set
 			if err := checkInputs(ctx, env); err != nil {
 				return err
@@ -60,6 +76,11 @@ var upCmd = &cobra.Command{
 				return err
 			}
 
+			// FIXME: `ListOutput` is printing to Stdout directly which messes
+			// up the TTY logger.
+			if tty != nil {
+				tty.Stop()
+			}
 			return output.ListOutputs(ctx, env, term.IsTerminal(int(os.Stdout.Fd())))
 		})
 

cmd/dagger/logger/logger.go

@@ -21,8 +21,8 @@ func New() zerolog.Logger {
 		Timestamp().
 		Logger()
 
-	if prettyLogs() {
-		logger = logger.Output(&Console{Out: os.Stderr})
+	if !jsonLogs() {
+		logger = logger.Output(&PlainOutput{Out: os.Stderr})
 	} else {
 		logger = logger.With().Timestamp().Caller().Logger()
 	}
@@ -35,14 +35,16 @@ func New() zerolog.Logger {
 	return logger.Level(lvl)
 }
 
-func prettyLogs() bool {
+func jsonLogs() bool {
 	switch f := viper.GetString("log-format"); f {
 	case "json":
-		return false
-	case "pretty":
 		return true
-	case "":
-		return term.IsTerminal(int(os.Stdout.Fd()))
+	case "plain":
+		return false
+	case "tty":
+		return false
+	case "auto":
+		return !term.IsTerminal(int(os.Stdout.Fd()))
 	default:
 		fmt.Fprintf(os.Stderr, "invalid --log-format %q\n", f)
 		os.Exit(1)

cmd/dagger/logger/plain.go

@@ -6,8 +6,8 @@ import (
 	"fmt"
 	"hash/adler32"
 	"io"
+	"sort"
 	"strings"
-	"sync"
 	"time"
 
 	"github.com/mitchellh/colorstring"
@@ -19,38 +19,29 @@ var colorize = colorstring.Colorize{
 	Reset:  true,
 }
 
-type Console struct {
-	Out       io.Writer
-	maxLength int
-	l         sync.Mutex
+type PlainOutput struct {
+	Out io.Writer
 }
 
-func (c *Console) Write(p []byte) (n int, err error) {
+func (c *PlainOutput) Write(p []byte) (int, error) {
 	event := map[string]interface{}{}
 	d := json.NewDecoder(bytes.NewReader(p))
 	if err := d.Decode(&event); err != nil {
-		return n, fmt.Errorf("cannot decode event: %s", err)
+		return 0, fmt.Errorf("cannot decode event: %s", err)
 	}
 
-	source := c.parseSource(event)
+	source := parseSource(event)
 
-	c.l.Lock()
-	if len(source) > c.maxLength {
-		c.maxLength = len(source)
-	}
-	c.l.Unlock()
-
-	return fmt.Fprintln(c.Out,
-		colorize.Color(fmt.Sprintf("%s %s %s%s%s",
-			c.formatTimestamp(event),
-			c.formatLevel(event),
-			c.formatSource(source),
-			c.formatMessage(event),
-			c.formatFields(event),
-		)))
+	return fmt.Fprintln(c.Out, colorize.Color(fmt.Sprintf("%s %s %s%s%s",
+		formatTimestamp(event),
+		formatLevel(event),
+		formatSource(source),
+		formatMessage(event),
+		formatFields(event),
+	)))
 }
 
-func (c *Console) formatLevel(event map[string]interface{}) string {
+func formatLevel(event map[string]interface{}) string {
 	level := zerolog.DebugLevel
 	if l, ok := event[zerolog.LevelFieldName].(string); ok {
 		level, _ = zerolog.ParseLevel(l)
@@ -76,7 +67,7 @@ func (c *Console) formatLevel(event map[string]interface{}) string {
 	}
 }
 
-func (c *Console) formatTimestamp(event map[string]interface{}) string {
+func formatTimestamp(event map[string]interface{}) string {
 	ts, ok := event[zerolog.TimestampFieldName].(string)
 	if !ok {
 		return "???"
@@ -89,7 +80,7 @@ func (c *Console) formatTimestamp(event map[string]interface{}) string {
 	return fmt.Sprintf("[dark_gray]%s[reset]", t.Format(time.Kitchen))
 }
 
-func (c *Console) formatMessage(event map[string]interface{}) string {
+func formatMessage(event map[string]interface{}) string {
 	message, ok := event[zerolog.MessageFieldName].(string)
 	if !ok {
 		return ""
@@ -125,7 +116,7 @@ func (c *Console) formatMessage(event map[string]interface{}) string {
 	}
 }
 
-func (c *Console) parseSource(event map[string]interface{}) string {
+func parseSource(event map[string]interface{}) string {
 	source := "system"
 	if task, ok := event["component"].(string); ok && task != "" {
 		source = task
@@ -133,14 +124,14 @@ func (c *Console) parseSource(event map[string]interface{}) string {
 	return source
 }
 
-func (c *Console) formatSource(source string) string {
+func formatSource(source string) string {
 	return fmt.Sprintf("[%s]%s | [reset]",
 		hashColor(source),
 		source,
 	)
 }
 
-func (c *Console) formatFields(entry map[string]interface{}) string {
+func formatFields(entry map[string]interface{}) string {
 	// these are the fields we don't want to expose, either because they're
 	// already part of the Log structure or because they're internal
 	fieldSkipList := map[string]struct{}{
@@ -149,7 +140,9 @@ func (c *Console) formatFields(entry map[string]interface{}) string {
 		zerolog.TimestampFieldName: {},
 		zerolog.ErrorFieldName:     {},
 		zerolog.CallerFieldName:    {},
+		"environment":              {},
 		"component":                {},
+		"state":                    {},
 	}
 
 	fields := []string{}
@@ -180,7 +173,10 @@ func (c *Console) formatFields(entry map[string]interface{}) string {
 	if len(fields) == 0 {
 		return ""
 	}
-	return fmt.Sprintf("    [dim]%s[reset]", strings.Join(fields, " "))
+	sort.SliceStable(fields, func(i, j int) bool {
+		return fields[i] < fields[j]
+	})
+	return fmt.Sprintf("    [bold]%s[reset]", strings.Join(fields, " "))
 }
 
 // hashColor returns a consistent color for a given string
@@ -195,8 +191,6 @@ func hashColor(text string) string {
 		"light_yellow",
 		"cyan",
 		"light_cyan",
-		"red",
-		"light_red",
 	}
 	h := adler32.Checksum([]byte(text))
 	return colors[int(h)%len(colors)]

cmd/dagger/logger/tty.go

@@ -0,0 +1,356 @@
+package logger
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"os"
+	"strings"
+	"sync"
+	"time"
+	"unicode/utf8"
+
+	"github.com/containerd/console"
+	"github.com/morikuni/aec"
+	"go.dagger.io/dagger/environment"
+)
+
+type Event map[string]interface{}
+
+type Group struct {
+	Name      string
+	State     environment.State
+	Events    []Event
+	Started   *time.Time
+	Completed *time.Time
+}
+
+type Message struct {
+	Event Event
+	Group *Group
+}
+
+type Logs struct {
+	Messages []Message
+
+	groups map[string]*Group
+	l      sync.Mutex
+}
+
+func (l *Logs) Add(event Event) error {
+	l.l.Lock()
+	defer l.l.Unlock()
+
+	component, ok := event["component"].(string)
+	if !ok {
+		l.Messages = append(l.Messages, Message{
+			Event: event,
+		})
+
+		return nil
+	}
+
+	groupKey := strings.Split(component, ".#up")[0]
+	group := l.groups[groupKey]
+
+	// If the group doesn't exist, create it
+	if group == nil {
+		now := time.Now()
+		group = &Group{
+			Name:    groupKey,
+			Started: &now,
+		}
+		l.groups[groupKey] = group
+		l.Messages = append(l.Messages, Message{
+			Group: group,
+		})
+	}
+
+	// Handle state events
+	// For state events, we just want to update the group status -- no need to
+	// dispanything
+	if st, ok := event["state"].(string); ok {
+		// Ignore state updates for "sub" components
+		if component != groupKey {
+			return nil
+		}
+
+		group.State = environment.State(st)
+		if group.State != environment.StateComputing {
+			now := time.Now()
+			group.Completed = &now
+		}
+
+		return nil
+	}
+
+	group.Events = append(group.Events, event)
+
+	return nil
+}
+
+type TTYOutput struct {
+	cons      console.Console
+	logs      *Logs
+	lineCount int
+	l         sync.RWMutex
+
+	stopCh  chan struct{}
+	doneCh  chan struct{}
+	printCh chan struct{}
+}
+
+func NewTTYOutput(w *os.File) (*TTYOutput, error) {
+	cons, err := console.ConsoleFromFile(w)
+	if err != nil {
+		return nil, err
+	}
+
+	c := &TTYOutput{
+		logs: &Logs{
+			groups: make(map[string]*Group),
+		},
+		cons:    cons,
+		stopCh:  make(chan struct{}),
+		doneCh:  make(chan struct{}),
+		printCh: make(chan struct{}, 128),
+	}
+
+	return c, nil
+}
+
+func (c *TTYOutput) Start() {
+	defer close(c.doneCh)
+	go func() {
+		for {
+			select {
+			case <-c.stopCh:
+				return
+			case <-c.printCh:
+				c.print()
+			case <-time.After(100 * time.Millisecond):
+				c.print()
+			}
+		}
+	}()
+}
+
+func (c *TTYOutput) Stop() {
+	c.l.Lock()
+	defer c.l.Unlock()
+
+	if c.doneCh == nil {
+		return
+	}
+	close(c.stopCh)
+	<-c.doneCh
+	c.doneCh = nil
+}
+
+func (c *TTYOutput) Write(p []byte) (n int, err error) {
+	event := Event{}
+	d := json.NewDecoder(bytes.NewReader(p))
+	if err := d.Decode(&event); err != nil {
+		return n, fmt.Errorf("cannot decode event: %s", err)
+	}
+
+	if err := c.logs.Add(event); err != nil {
+		return 0, err
+	}
+
+	c.print()
+
+	return len(p), nil
+}
+
+func (c *TTYOutput) print() {
+	c.l.Lock()
+	defer c.l.Unlock()
+
+	// make sure the printer is not stopped
+	select {
+	case <-c.stopCh:
+		return
+	default:
+	}
+
+	width, height := c.getSize()
+
+	// hide during re-rendering to avoid flickering
+	fmt.Fprint(c.cons, aec.Hide)
+	defer fmt.Fprint(c.cons, aec.Show)
+
+	// rewind to the top
+	b := aec.EmptyBuilder
+	for i := 0; i < c.lineCount; i++ {
+		b = b.Up(1)
+	}
+	fmt.Fprint(c.cons, b.ANSI)
+
+	runningGroups := 0
+	for _, message := range c.logs.Messages {
+		group := message.Group
+		if group == nil || group.State != environment.StateComputing {
+			continue
+		}
+		runningGroups++
+	}
+
+	linesPerGroup := 5
+	if freeLines := (height - len(c.logs.Messages)); freeLines > 0 && runningGroups > 0 {
+		linesPerGroup = (freeLines - 2) / runningGroups
+	}
+
+	lineCount := 0
+	for _, message := range c.logs.Messages {
+		if group := message.Group; group != nil {
+			lineCount += c.printGroup(group, width, linesPerGroup)
+		} else {
+			lineCount += c.printLine(message.Event, width)
+		}
+	}
+
+	if diff := c.lineCount - lineCount; diff > 0 {
+		for i := 0; i < diff; i++ {
+			fmt.Fprintln(c.cons, strings.Repeat(" ", width))
+		}
+		fmt.Fprint(c.cons, aec.EmptyBuilder.Up(uint(diff)).Column(0).ANSI)
+	}
+
+	c.lineCount = lineCount
+}
+
+func (c *TTYOutput) printLine(event Event, width int) int {
+	message := colorize.Color(fmt.Sprintf("%s %s %s%s",
+		formatTimestamp(event),
+		formatLevel(event),
+		formatMessage(event),
+		formatFields(event),
+	))
+
+	// trim
+	for utf8.RuneCountInString(message) > width {
+		message = message[0:len(message)-4] + "…"
+	}
+
+	// pad
+	if delta := width - utf8.RuneCountInString(message); delta > 0 {
+		message += strings.Repeat(" ", delta)
+	}
+	message += "\n"
+
+	// print
+	fmt.Fprint(c.cons, message)
+
+	return 1
+}
+
+func (c *TTYOutput) printGroup(group *Group, width, maxLines int) int {
+	lineCount := 0
+
+	prefix := ""
+	switch group.State {
+	case environment.StateComputing:
+		prefix = "[+]"
+	case environment.StateCanceled:
+		prefix = "[✗]"
+	case environment.StateFailed:
+		prefix = "[✗]"
+	case environment.StateCompleted:
+		prefix = "[✔]"
+	}
+
+	out := prefix + " " + group.Name
+
+	endTime := time.Now()
+	if group.Completed != nil {
+		endTime = *group.Completed
+	}
+
+	dt := endTime.Sub(*group.Started).Seconds()
+	if dt < 0.05 {
+		dt = 0
+	}
+	timer := fmt.Sprintf("%3.1fs", dt)
+
+	// align
+	out += strings.Repeat(" ", width-utf8.RuneCountInString(out)-len(timer))
+	out += timer
+	out += "\n"
+
+	// color
+	switch group.State {
+	case environment.StateComputing:
+		out = aec.Apply(out, aec.LightBlueF)
+	case environment.StateCanceled:
+		out = aec.Apply(out, aec.LightYellowF)
+	case environment.StateFailed:
+		out = aec.Apply(out, aec.LightRedF)
+	case environment.StateCompleted:
+		out = aec.Apply(out, aec.LightGreenF)
+	}
+
+	// Print
+	fmt.Fprint(c.cons, out)
+	lineCount++
+
+	if group.State == environment.StateCompleted {
+		// for completed tasks, don't show any logs
+		return lineCount
+	}
+
+	events := group.Events
+
+	if group.State == environment.StateComputing {
+		// for computing tasks, show only last N
+		if len(events) > maxLines {
+			events = events[len(events)-maxLines:]
+		}
+	}
+
+	// for everything else (error, canceled), show all logs
+
+	for _, event := range events {
+		lineCount += c.printGroupLine(event, width)
+	}
+
+	return lineCount
+}
+
+func (c *TTYOutput) printGroupLine(event Event, width int) int {
+	message := colorize.Color(fmt.Sprintf("%s%s",
+		formatMessage(event),
+		formatFields(event),
+	))
+
+	// trim
+	for utf8.RuneCountInString(message) > width {
+		message = message[0:len(message)-4] + "…"
+	}
+
+	// pad
+	if delta := width - utf8.RuneCountInString(message); delta > 0 {
+		message += strings.Repeat(" ", delta)
+	}
+	message += "\n"
+
+	// color
+	message = aec.Apply(message, aec.Faint)
+
+	// Print
+	fmt.Fprint(c.cons, message)
+
+	return 1
+}
+
+func (c *TTYOutput) getSize() (int, int) {
+	width := 80
+	height := 10
+	size, err := c.cons.Size()
+	if err == nil && size.Width > 0 && size.Height > 0 {
+		width = int(size.Width)
+		height = int(size.Height)
+	}
+
+	return width, height
+}