feat: realtime output

artifact/artifact.go

@@ -240,8 +240,9 @@ func (ka Artifact) Supported(ki distro.KernelInfo) (supported bool, err error) {
 	return
 }
 
+// TODO too many parameters
 func (ka Artifact) Process(slog zerolog.Logger, ki distro.KernelInfo,
-	endless bool, cBinary,
+	outputOnSuccess, realtimeOutput, endless bool, cBinary,
 	cEndlessStress string, cEndlessTimeout time.Duration,
 	dump func(q *qemu.System, ka Artifact, ki distro.KernelInfo,
 		result *Result)) {
@@ -337,10 +338,14 @@ func (ka Artifact) Process(slog zerolog.Logger, ki distro.KernelInfo,
 		slog.Debug().Str("duration", time.Since(start).String()).
 			Msg("build done")
 		if err != nil {
-			slog.Error().Err(err).Msg("build")
+			slog.Error().Err(err).Msgf("build failure\n%v\n", result.Build.Output)
 			return
 		} else {
-			slog.Info().Err(err).Msg("build success")
+			if outputOnSuccess && !realtimeOutput {
+				slog.Info().Msgf("build success\n%v\n", result.Build.Output)
+			} else {
+				slog.Info().Msg("build success")
+			}
 		}
 		result.Build.Ok = true
 	} else {
@@ -398,11 +403,39 @@ func (ka Artifact) Process(slog zerolog.Logger, ki distro.KernelInfo,
 		return
 	}
 
+	var qemuTestOutput string
+	q.SetQemuOutputHandler(func(s string) {
+		if realtimeOutput {
+			fmt.Printf("kmsg: %s\n", s)
+		} else {
+			qemuTestOutput += s + "\n"
+		}
+	})
+
+	if realtimeOutput {
+		q.SetCommandsOutputHandler(func(s string) {
+			fmt.Printf("test: %s\n", s)
+		})
+	}
+
 	start := time.Now()
-	copyArtifactAndTest(slog, q, ka, &result, remoteTest)
+	copyArtifactAndTest(slog, q, ka, &result, remoteTest, outputOnSuccess, realtimeOutput)
 	slog.Debug().Str("duration", time.Since(start).String()).
 		Msgf("test completed (success: %v)", result.Test.Ok)
 
+	if result.Build.Ok {
+		if !result.Run.Ok || !result.Test.Ok {
+			slog.Error().Msgf("qemu output\n%v\n", qemuTestOutput)
+		} else if outputOnSuccess && !realtimeOutput {
+			slog.Info().Msgf("qemu output\n%v\n", qemuTestOutput)
+		}
+	}
+
+	if realtimeOutput {
+		q.CloseCommandsOutputHandler()
+	}
+	q.CloseQemuOutputHandler()
+
 	if !endless {
 		return
 	}

artifact/process.go

@@ -157,9 +157,15 @@ func Build(flog zerolog.Logger, tmp string, ka Artifact,
 
 		c.Args = append(c.Args, "--network", "none")
 
+		c.SetCommandsOutputHandler(func(s string) {
+			fmt.Printf("%s\n", s)
+		})
+
 		output, err = c.Run(outdir, []string{
 			buildCommand + " && chmod -R 777 /work",
 		})
+
+		c.CloseCommandsOutputHandler()
 	} else {
 		cmd := exec.Command("bash", "-c", "cd "+outdir+" && "+
 			buildCommand)
@@ -281,7 +287,7 @@ func CopyFile(sourcePath, destinationPath string) (err error) {
 }
 
 func copyArtifactAndTest(slog zerolog.Logger, q *qemu.System, ka Artifact,
-	res *Result, remoteTest string) (err error) {
+	res *Result, remoteTest string, outputOnSuccess, realtimeOutput bool) (err error) {
 
 	// Copy all test files to the remote machine
 	for _, f := range ka.TestFiles {
@@ -346,7 +352,11 @@ func copyArtifactAndTest(slog zerolog.Logger, q *qemu.System, ka Artifact,
 		return
 	}
 
-	slog.Info().Msgf("test success\n%v\n", res.Test.Output)
+	if outputOnSuccess && !realtimeOutput {
+		slog.Info().Msgf("test success\n%v\n", res.Test.Output)
+	} else {
+		slog.Info().Msg("test success")
+	}
 
 	_, err = q.Command("root", "echo")
 	if err != nil {

cmd/pew.go

@@ -87,6 +87,9 @@ type PewCmd struct {
 	Threshold             float64 `help:"reliablity threshold for exit code" default:"1.00"`
 	IncludeInternalErrors bool    `help:"count internal errors as part of the success rate"`
 
+	OutputOnSuccess bool `help:"show output on success"`
+	RealtimeOutput  bool `help:"show realtime output"`
+
 	Endless        bool          `help:"endless tests"`
 	EndlessTimeout time.Duration `help:"timeout between tests" default:"1m"`
 	EndlessStress  string        `help:"endless stress script" type:"existingfile"`
@@ -445,7 +448,7 @@ func (cmd PewCmd) testArtifact(swg *sizedwaitgroup.SizedWaitGroup,
 		Str("kernel", ki.KernelRelease).
 		Logger()
 
-	ka.Process(slog, ki,
+	ka.Process(slog, ki, cmd.OutputOnSuccess, cmd.RealtimeOutput,
 		cmd.Endless, cmd.Binary, cmd.EndlessStress, cmd.EndlessTimeout,
 		func(q *qemu.System, ka artifact.Artifact, ki distro.KernelInfo, result *artifact.Result) {
 			dumpResult(q, ka, ki, result, cmd.Dist, cmd.Tag, cmd.Binary, cmd.DB)

container/container.go

@@ -14,6 +14,7 @@ import (
 	"path/filepath"
 	"regexp"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/cavaliergopher/grab/v3"
@@ -176,6 +177,11 @@ type Container struct {
 	Args []string
 
 	Log zerolog.Logger
+
+	commandsOutput struct {
+		listener chan string
+		mu       sync.Mutex
+	}
 }
 
 func New(dist distro.Distro) (c Container, err error) {
@@ -234,6 +240,43 @@ func NewFromKernelInfo(ki distro.KernelInfo) (
 	return
 }
 
+// c.SetCommandsOutputHandler(func(s string) { fmt.Println(s) })
+// defer c.CloseCommandsOutputHandler()
+func (c *Container) SetCommandsOutputHandler(handler func(s string)) {
+	c.commandsOutput.mu.Lock()
+	defer c.commandsOutput.mu.Unlock()
+
+	c.commandsOutput.listener = make(chan string)
+
+	go func(l chan string) {
+		for m := range l {
+			if m != "" {
+				handler(m)
+			}
+		}
+	}(c.commandsOutput.listener)
+}
+
+func (c *Container) CloseCommandsOutputHandler() {
+	c.commandsOutput.mu.Lock()
+	defer c.commandsOutput.mu.Unlock()
+
+	close(c.commandsOutput.listener)
+	c.commandsOutput.listener = nil
+}
+
+func (c *Container) handleCommandsOutput(m string) {
+	if c.commandsOutput.listener == nil {
+		return
+	}
+	c.commandsOutput.mu.Lock()
+	defer c.commandsOutput.mu.Unlock()
+
+	if c.commandsOutput.listener != nil {
+		c.commandsOutput.listener <- m
+	}
+}
+
 func (c Container) Name() string {
 	return c.name
 }
@@ -408,6 +451,7 @@ func (c Container) build(imagePath string) (output string, err error) {
 		scanner := bufio.NewScanner(stdout)
 		for scanner.Scan() {
 			m := scanner.Text()
+			c.handleCommandsOutput(m)
 			output += m + "\n"
 			flog.Trace().Str("stdout", m).Msg("")
 		}
@@ -481,6 +525,7 @@ func (c Container) Run(workdir string, cmds []string) (out string, err error) {
 		scanner := bufio.NewScanner(stdout)
 		for scanner.Scan() {
 			m := scanner.Text()
+			c.handleCommandsOutput(m)
 			out += m + "\n"
 			flog.Trace().Str("stdout", m).Msg("")
 		}
@@ -488,7 +533,6 @@ func (c Container) Run(workdir string, cmds []string) (out string, err error) {
 
 	err = cmd.Wait()
 	if err != nil {
-		flog.Error().Msg(out)
 		return
 	}
 

daemon/process.go

@@ -131,7 +131,7 @@ func (pj *jobProcessor) Process(res *Resources) (err error) {
 	var result *artifact.Result
 	var dq *qemu.System
 
-	pj.job.Artifact.Process(pj.log, pj.job.Target, false, "", "", 0,
+	pj.job.Artifact.Process(pj.log, pj.job.Target, false, false, false, "", "", 0,
 		func(q *qemu.System, ka artifact.Artifact, ki distro.KernelInfo,
 			res *artifact.Result) {
 

qemu/qemu-kernel.go

@@ -101,6 +101,16 @@ type System struct {
 
 	Stdout, Stderr string
 
+	qemuOutput struct {
+		listener chan string
+		mu       sync.Mutex
+	}
+
+	commandsOutput struct {
+		listener chan string
+		mu       sync.Mutex
+	}
+
 	// accessible after qemu is closed
 	exitErr error
 
@@ -138,6 +148,80 @@ func NewSystem(arch arch, kernel Kernel, drivePath string) (q *System, err error
 	return
 }
 
+// q.SetQemuOutputHandler(func(s string) { fmt.Println(s) })
+// defer q.CloseQemuOutputHandler()
+func (q *System) SetQemuOutputHandler(handler func(s string)) {
+	q.qemuOutput.mu.Lock()
+	defer q.qemuOutput.mu.Unlock()
+
+	q.qemuOutput.listener = make(chan string)
+
+	go func(l chan string) {
+		for m := range l {
+			if m != "" {
+				handler(m)
+			}
+		}
+	}(q.qemuOutput.listener)
+}
+
+func (q *System) CloseQemuOutputHandler() {
+	q.qemuOutput.mu.Lock()
+	defer q.qemuOutput.mu.Unlock()
+
+	close(q.qemuOutput.listener)
+	q.qemuOutput.listener = nil
+}
+
+func (q *System) handleQemuOutput(m string) {
+	if q.qemuOutput.listener == nil {
+		return
+	}
+	q.qemuOutput.mu.Lock()
+	defer q.qemuOutput.mu.Unlock()
+
+	if q.qemuOutput.listener != nil {
+		q.qemuOutput.listener <- m
+	}
+}
+
+// q.SetCommandsOutputHandler(func(s string) { fmt.Println(s) })
+// defer q.CloseCommandsOutputHandler()
+func (q *System) SetCommandsOutputHandler(handler func(s string)) {
+	q.commandsOutput.mu.Lock()
+	defer q.commandsOutput.mu.Unlock()
+
+	q.commandsOutput.listener = make(chan string)
+
+	go func(l chan string) {
+		for m := range l {
+			if m != "" {
+				handler(m)
+			}
+		}
+	}(q.commandsOutput.listener)
+}
+
+func (q *System) CloseCommandsOutputHandler() {
+	q.commandsOutput.mu.Lock()
+	defer q.commandsOutput.mu.Unlock()
+
+	close(q.commandsOutput.listener)
+	q.commandsOutput.listener = nil
+}
+
+func (q *System) handleCommandsOutput(m string) {
+	if q.commandsOutput.listener == nil {
+		return
+	}
+	q.commandsOutput.mu.Lock()
+	defer q.commandsOutput.mu.Unlock()
+
+	if q.commandsOutput.listener != nil {
+		q.commandsOutput.listener <- m
+	}
+}
+
 func (q *System) SetSSHAddrPort(addr string, port int) (err error) {
 	// TODO validate
 	q.SSH.AddrPort = fmt.Sprintf("%s:%d", addr, port)
@@ -313,6 +397,7 @@ func (q *System) Start() (err error) {
 		scanner := bufio.NewScanner(q.pipe.stdout)
 		for scanner.Scan() {
 			m := scanner.Text()
+			q.handleQemuOutput(m)
 			q.Stdout += m + "\n"
 			q.Log.Trace().Str("stdout", m).Msg("qemu")
 			go q.checkOopsPanic(m)
@@ -323,6 +408,7 @@ func (q *System) Start() (err error) {
 		scanner := bufio.NewScanner(q.pipe.stderr)
 		for scanner.Scan() {
 			m := scanner.Text()
+			q.handleQemuOutput(m)
 			q.Stderr += m + "\n"
 			q.Log.Trace().Str("stderr", m).Msg("qemu")
 		}
@@ -475,6 +561,7 @@ func (q System) Command(user, cmd string) (output string, err error) {
 		scanner := bufio.NewScanner(stdout)
 		for scanner.Scan() {
 			m := scanner.Text()
+			q.handleCommandsOutput(m)
 			output += m + "\n"
 			flog.Trace().Str("stdout", m).Msg("qemu command")
 		}
@@ -488,6 +575,7 @@ func (q System) Command(user, cmd string) (output string, err error) {
 		scanner := bufio.NewScanner(stderr)
 		for scanner.Scan() {
 			m := scanner.Text()
+			q.handleCommandsOutput(m)
 			output += m + "\n"
 			// Note: it prints stderr as stdout
 			flog.Trace().Str("stdout", m).Msg("qemu command")