feat: realtime output

BREAKING CHANGE: Parameters for the kernel command are changed

from

--distro= --ver= --kernel=

to

--distro-id= --distro-release= --kernel-regex=

.github/workflows/ubuntu.yml

@@ -203,7 +203,7 @@ jobs:
 
           cp ../examples/kernel-module/{module.c,Makefile,test.sh} .
 
-          ../out-of-tree --log-level=debug kernel list-remote --distro=${{ matrix.os.distro }} --ver=${{ matrix.os.release }}
+          ../out-of-tree --log-level=debug kernel list-remote --distro-id=${{ matrix.os.distro }} --distro-release=${{ matrix.os.release }}
           ../out-of-tree --log-level=debug kernel autogen --max=1 --shuffle
           ../out-of-tree --log-level=debug pew --qemu-timeout=20m --include-internal-errors
 

artifact/artifact.go

@@ -12,7 +12,6 @@ import (
 
 	"github.com/naoina/toml"
 	"github.com/rs/zerolog"
-	"github.com/rs/zerolog/log"
 
 	"code.dumpstack.io/tools/out-of-tree/config/dotfiles"
 	"code.dumpstack.io/tools/out-of-tree/distro"
@@ -241,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)) {
@@ -338,8 +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 {
-			log.Error().Err(err).Msg("build")
+			slog.Error().Err(err).Msgf("build failure\n%v\n", result.Build.Output)
 			return
+		} else {
+			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 {
@@ -397,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 {
@@ -313,8 +319,7 @@ func copyArtifactAndTest(slog zerolog.Logger, q *qemu.System, ka Artifact,
 
 		res.Test.Output, err = testKernelModule(q, ka, remoteTest)
 		if err != nil {
-			slog.Error().Err(err).Msg(res.Test.Output)
-			return
+			break
 		}
 		res.Test.Ok = true
 	case KernelExploit:
@@ -327,16 +332,14 @@ func copyArtifactAndTest(slog zerolog.Logger, q *qemu.System, ka Artifact,
 		res.Test.Output, err = testKernelExploit(q, ka, remoteTest,
 			remoteExploit)
 		if err != nil {
-			slog.Error().Err(err).Msg(res.Test.Output)
-			return
+			break
 		}
 		res.Run.Ok = true // does not really used
 		res.Test.Ok = true
 	case Script:
 		res.Test.Output, err = runScript(q, remoteTest)
 		if err != nil {
-			slog.Error().Err(err).Msg(res.Test.Output)
-			return
+			break
 		}
 		res.Run.Ok = true
 		res.Test.Ok = true
@@ -344,7 +347,16 @@ func copyArtifactAndTest(slog zerolog.Logger, q *qemu.System, ka Artifact,
 		slog.Fatal().Msg("Unsupported artifact type")
 	}
 
-	slog.Info().Msgf("\n%v\n", res.Test.Output)
+	if err != nil || !res.Test.Ok {
+		slog.Error().Err(err).Msgf("test error\n%v\n", res.Test.Output)
+		return
+	}
+
+	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/container.go

@@ -24,6 +24,15 @@ type ContainerCmd struct {
 	Update  ContainerUpdateCmd  `cmd:"" help:"update containers"`
 	Save    ContainerSaveCmd    `cmd:"" help:"save containers"`
 	Cleanup ContainerCleanupCmd `cmd:"" help:"cleanup containers"`
+
+	RealtimeOutput RealtimeContainerOutputFlag `help:"show realtime output"`
+}
+
+type RealtimeContainerOutputFlag bool
+
+func (f RealtimeContainerOutputFlag) AfterApply() (err error) {
+	container.Stdout = bool(f)
+	return
 }
 
 func (cmd ContainerCmd) Containers() (diis []container.Image, err error) {

cmd/kernel.go

@@ -39,6 +39,8 @@ type KernelCmd struct {
 
 	ContainerTimeout time.Duration `help:"container timeout"`
 
+	RealtimeOutput RealtimeContainerOutputFlag `help:"show realtime output"`
+
 	List        KernelListCmd        `cmd:"" help:"list kernels"`
 	ListRemote  KernelListRemoteCmd  `cmd:"" help:"list remote kernels"`
 	Autogen     KernelAutogenCmd     `cmd:"" help:"generate kernels based on the current config"`
@@ -265,8 +267,8 @@ func (cmd *KernelListCmd) Run(g *Globals) (err error) {
 }
 
 type KernelListRemoteCmd struct {
-	Distro string `required:"" help:"distribution"`
-	Ver    string `help:"distro version"`
+	DistroID      string `required:"" help:"distribution"`
+	DistroRelease string `help:"distro version"`
 }
 
 func (cmd *KernelListRemoteCmd) Run(kernelCmd *KernelCmd, g *Globals) (err error) {
@@ -279,13 +281,13 @@ func (cmd *KernelListRemoteCmd) Run(kernelCmd *KernelCmd, g *Globals) (err error
 
 	container.UsePrebuilt = kernelCmd.PrebuiltContainers
 
-	distroType, err := distro.NewID(cmd.Distro)
+	distroType, err := distro.NewID(cmd.DistroID)
 	if err != nil {
 		return
 	}
 
 	km := artifact.Target{
-		Distro: distro.Distro{ID: distroType, Release: cmd.Ver},
+		Distro: distro.Distro{ID: distroType, Release: cmd.DistroRelease},
 		Kernel: artifact.Kernel{Regex: ".*"},
 	}
 
@@ -336,12 +338,12 @@ func (cmd *KernelAutogenCmd) Run(kernelCmd *KernelCmd, g *Globals) (err error) {
 }
 
 type KernelGenallCmd struct {
-	Distro string `help:"distribution"`
-	Ver    string `help:"distro version"`
+	DistroID      string `help:"distribution"`
+	DistroRelease string `help:"distro version"`
 }
 
 func (cmd *KernelGenallCmd) Run(kernelCmd *KernelCmd, g *Globals) (err error) {
-	distroType, err := distro.NewID(cmd.Distro)
+	distroType, err := distro.NewID(cmd.DistroID)
 	if err != nil {
 		return
 	}
@@ -357,7 +359,7 @@ func (cmd *KernelGenallCmd) Run(kernelCmd *KernelCmd, g *Globals) (err error) {
 			continue
 		}
 
-		if cmd.Ver != "" && dist.Release != cmd.Ver {
+		if cmd.DistroRelease != "" && dist.Release != cmd.DistroRelease {
 			continue
 		}
 
@@ -376,13 +378,13 @@ func (cmd *KernelGenallCmd) Run(kernelCmd *KernelCmd, g *Globals) (err error) {
 }
 
 type KernelInstallCmd struct {
-	Distro string `required:"" help:"distribution"`
-	Ver    string `required:"" help:"distro version"`
-	Kernel string `required:"" help:"kernel release mask"`
+	DistroID      string `required:"" help:"distribution"`
+	DistroRelease string `required:"" help:"distro version"`
+	KernelRegex   string `required:"" help:"kernel release mask"`
 }
 
 func (cmd *KernelInstallCmd) Run(kernelCmd *KernelCmd, g *Globals) (err error) {
-	distroType, err := distro.NewID(cmd.Distro)
+	distroType, err := distro.NewID(cmd.DistroID)
 	if err != nil {
 		return
 	}
@@ -390,8 +392,8 @@ func (cmd *KernelInstallCmd) Run(kernelCmd *KernelCmd, g *Globals) (err error) {
 	kernel.SetSigintHandler(&kernelCmd.shutdown)
 
 	km := artifact.Target{
-		Distro: distro.Distro{ID: distroType, Release: cmd.Ver},
-		Kernel: artifact.Kernel{Regex: cmd.Kernel},
+		Distro: distro.Distro{ID: distroType, Release: cmd.DistroRelease},
+		Kernel: artifact.Kernel{Regex: cmd.KernelRegex},
 	}
 	err = kernelCmd.Generate(g, km)
 	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"
@@ -45,6 +46,8 @@ var UsePrebuilt = true
 
 var Prune = true
 
+var Stdout = false
+
 type Image struct {
 	Name   string
 	Distro distro.Distro
@@ -176,6 +179,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 +242,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 +453,10 @@ func (c Container) build(imagePath string) (output string, err error) {
 		scanner := bufio.NewScanner(stdout)
 		for scanner.Scan() {
 			m := scanner.Text()
+			if Stdout {
+				fmt.Println(m)
+			}
+			c.handleCommandsOutput(m)
 			output += m + "\n"
 			flog.Trace().Str("stdout", m).Msg("")
 		}
@@ -481,6 +530,10 @@ func (c Container) Run(workdir string, cmds []string) (out string, err error) {
 		scanner := bufio.NewScanner(stdout)
 		for scanner.Scan() {
 			m := scanner.Text()
+			if Stdout {
+				fmt.Println(m)
+			}
+			c.handleCommandsOutput(m)
 			out += m + "\n"
 			flog.Trace().Str("stdout", m).Msg("")
 		}
@@ -488,9 +541,6 @@ func (c Container) Run(workdir string, cmds []string) (out string, err error) {
 
 	err = cmd.Wait()
 	if err != nil {
-		e := fmt.Sprintf("error `%v` for cmd `%v` with output `%v`",
-			err, cmds, out)
-		err = errors.New(e)
 		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")