Move state to event for whether they support inplace progress logging (#339)

## Changes
Adds a IsInplaceSupported() function to the event interface. Any event
that now uses the progress logger has to declare whether they support in
place logging

## Tests
Manually

bundle/deploy/terraform/destroy.go

@@ -41,6 +41,10 @@ func (c *PlanResourceChange) String() string {
 	return result.String()
 }
 
+func (c *PlanResourceChange) IsInplaceSupported() bool {
+	return false
+}
+
 func logDestroyPlan(l *cmdio.Logger, changes []*tfjson.ResourceChange) error {
 	// TODO: remove once we have mutator logging in place
 	fmt.Fprintln(os.Stderr, "The following resources will be removed: ")

bundle/run/pipeline.go

@@ -11,7 +11,6 @@ import (
 	"github.com/databricks/bricks/bundle/run/output"
 	"github.com/databricks/bricks/bundle/run/progress"
 	"github.com/databricks/bricks/libs/cmdio"
-	"github.com/databricks/bricks/libs/flags"
 	"github.com/databricks/bricks/libs/log"
 	"github.com/databricks/databricks-sdk-go/service/pipelines"
 	flag "github.com/spf13/pflag"
@@ -167,10 +166,6 @@ func (r *pipelineRunner) Run(ctx context.Context, opts *Options) (output.RunOutp
 	if !ok {
 		return nil, fmt.Errorf("no progress logger found")
 	}
-	// Inplace logger mode is not supported for pipelines right now
-	if progressLogger.Mode == flags.ModeInplace {
-		progressLogger.Mode = flags.ModeAppend
-	}
 
 	// Log the pipeline update URL as soon as it is available.
 	progressLogger.Log(progress.NewUpdateUrlEvent(w.Config.Host, updateID, pipelineID))

bundle/run/progress/job.go

@@ -33,3 +33,7 @@ func (event *JobProgressEvent) String() string {
 	result.WriteString(event.RunPageURL)
 	return result.String()
 }
+
+func (event *JobProgressEvent) IsInplaceSupported() bool {
+	return true
+}

bundle/run/progress/pipeline.go

@@ -25,6 +25,10 @@ func (event *ProgressEvent) String() string {
 	return result.String()
 }
 
+func (event *ProgressEvent) IsInplaceSupported() bool {
+	return false
+}
+
 // TODO: Add inplace logging to pipelines. https://github.com/databricks/bricks/issues/280
 type UpdateTracker struct {
 	UpdateId             string

bundle/run/progress/pipeline_events.go

@@ -21,3 +21,7 @@ func NewUpdateUrlEvent(host, updateId, pipelineId string) *UpdateUrlEvent {
 func (event *UpdateUrlEvent) String() string {
 	return fmt.Sprintf("The update can be found at %s\n", event.Url)
 }
+
+func (event *UpdateUrlEvent) IsInplaceSupported() bool {
+	return false
+}

libs/cmdio/event.go

@@ -1,5 +1,9 @@
 package cmdio
 
 type Event interface {
+	// convert event into human readable string
 	String() string
+
+	// true if event supports inplace logging, return false otherwise
+	IsInplaceSupported() bool
 }

libs/cmdio/logger.go

@@ -42,40 +42,57 @@ func (l *Logger) Ask(question string) (bool, error) {
 	}
 }
 
+func (l *Logger) writeJson(event Event) {
+	b, err := json.MarshalIndent(event, "", "  ")
+	if err != nil {
+		// we panic because there we cannot catch this in jobs.RunNowAndWait
+		panic(err)
+	}
+	l.Writer.Write([]byte(b))
+	l.Writer.Write([]byte("\n"))
+}
+
+func (l *Logger) writeAppend(event Event) {
+	l.Writer.Write([]byte(event.String()))
+	l.Writer.Write([]byte("\n"))
+}
+
+func (l *Logger) writeInplace(event Event) {
+	if l.isFirstEvent {
+		// save cursor location
+		l.Writer.Write([]byte("\033[s"))
+	}
+
+	// move cursor to saved location
+	l.Writer.Write([]byte("\033[u"))
+
+	// clear from cursor to end of screen
+	l.Writer.Write([]byte("\033[0J"))
+
+	l.Writer.Write([]byte(event.String()))
+	l.Writer.Write([]byte("\n"))
+	l.isFirstEvent = false
+}
+
 func (l *Logger) Log(event Event) {
 	switch l.Mode {
 	case flags.ModeInplace:
-		if l.isFirstEvent {
-			// save cursor location
-			l.Writer.Write([]byte("\033[s"))
+		if event.IsInplaceSupported() {
+			l.writeInplace(event)
+		} else {
+			l.writeAppend(event)
 		}
 
-		// move cursor to saved location
-		l.Writer.Write([]byte("\033[u"))
-
-		// clear from cursor to end of screen
-		l.Writer.Write([]byte("\033[0J"))
-
-		l.Writer.Write([]byte(event.String()))
-		l.Writer.Write([]byte("\n"))
-
 	case flags.ModeJson:
-		b, err := json.MarshalIndent(event, "", "  ")
-		if err != nil {
-			// we panic because there we cannot catch this in jobs.RunNowAndWait
-			panic(err)
-		}
-		l.Writer.Write([]byte(b))
-		l.Writer.Write([]byte("\n"))
+		l.writeJson(event)
 
 	case flags.ModeAppend:
-		l.Writer.Write([]byte(event.String()))
-		l.Writer.Write([]byte("\n"))
+		l.writeAppend(event)
 
 	default:
 		// we panic because errors are not captured in some log sides like
 		// jobs.RunNowAndWait
 		panic("unknown progress logger mode: " + l.Mode.String())
 	}
-	l.isFirstEvent = false
+
 }