cmd/osbuild-worker-executor: use JSONSeqMonitor

Author: croissanne

cmd/osbuild-worker-executor/export_test.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"fmt"
 	"os"
 	"path/filepath"
 	"testing"
@@ -15,16 +16,15 @@ var (
 func MockOsbuildBinary(t *testing.T, new string) (restore func()) {
 	t.Helper()
 
-	saved := osbuildBinary
-
 	tmpdir := t.TempDir()
-	osbuildBinary = filepath.Join(tmpdir, "fake-osbuild")
 	/* #nosec G306 */
-	if err := os.WriteFile(osbuildBinary, []byte(new), 0755); err != nil {
+	if err := os.WriteFile(filepath.Join(tmpdir, "osbuild"), []byte(new), 0755); err != nil {
 		t.Fatal(err)
 	}
+	path := os.Getenv("PATH")
+	os.Setenv("PATH", fmt.Sprintf("%s:%s", tmpdir, path))
 
 	return func() {
-		osbuildBinary = saved
+		os.Setenv("PATH", path)
 	}
 }

cmd/osbuild-worker-executor/handler_build.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"archive/tar"
+	"bufio"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -15,11 +16,12 @@ import (
 	"golang.org/x/exp/slices"
 
 	"github.com/sirupsen/logrus"
+
+	"github.com/osbuild/images/pkg/osbuild"
 )
 
 var (
 	supportedBuildContentTypes = []string{"application/x-tar"}
-	osbuildBinary              = "osbuild"
 )
 
 var (
@@ -39,63 +41,96 @@ func (wf *writeFlusher) Write(p []byte) (n int, err error) {
 	return n, err
 }
 
-func runOsbuild(logger *logrus.Logger, buildDir string, control *controlJSON, output io.Writer) (string, error) {
-	flusher, ok := output.(http.Flusher)
-	if !ok {
-		return "", fmt.Errorf("cannot stream the output")
-	}
-	// stream output over http
-	wf := writeFlusher{w: output, flusher: flusher}
+func runOSBuild(logger *logrus.Logger, buildDir string, control *controlJSON, output io.Writer) (string, error) {
 	// note that the "filepath.Clean()" is here for gosec:G304
 	logfPath := filepath.Clean(filepath.Join(buildDir, "build.log"))
-	// and also write to our internal log
 	logf, err := os.Create(logfPath)
 	if err != nil {
 		return "", fmt.Errorf("cannot create log file: %v", err)
 	}
 	defer logf.Close()
 
+	manifest, err := os.ReadFile(filepath.Join(buildDir, "manifest.json"))
+	if err != nil {
+		return "", fmt.Errorf("cannot read manifest file: %w", err)
+	}
+
 	// use multi writer to get same output for stream and log
-	mw := io.MultiWriter(&wf, logf)
 	outputDir := filepath.Join(buildDir, "output")
 	storeDir := filepath.Join(buildDir, "osbuild-store")
-	cmd := exec.Command(osbuildBinary)
-	cmd.Stdout = mw
-	cmd.Stderr = mw
-	for _, exp := range control.Exports {
-		cmd.Args = append(cmd.Args, []string{"--export", exp}...)
+
+	// MonitorFile needs an *os.File:
+	// MonitorFile -> pipe -> bufio reader -> writeFlusher
+	rPipe, wPipe, err := os.Pipe()
+	if err != nil {
+		return "", fmt.Errorf("cannot create pipe for monitor file: %w", err)
 	}
-	cmd.Env = append(cmd.Env, control.Environments...)
-	cmd.Args = append(cmd.Args, []string{"--output-dir", outputDir}...)
-	cmd.Args = append(cmd.Args, []string{"--store", storeDir}...)
-	cmd.Args = append(cmd.Args, "--json")
-	cmd.Args = append(cmd.Args, filepath.Join(buildDir, "manifest.json"))
+	defer rPipe.Close()
+	defer wPipe.Close()
+
+	cmd := osbuild.NewOSBuildCmd(manifest, &osbuild.OSBuildOptions{
+		StoreDir:    storeDir,
+		OutputDir:   outputDir,
+		Exports:     control.Exports,
+		ExtraEnv:    control.Environments,
+		BuildLog:    logf,
+		Stdout:      os.Stdout,
+		Monitor:     osbuild.MonitorJSONSeq,
+		MonitorFile: wPipe,
+	})
 	if err := cmd.Start(); err != nil {
 		return "", err
 	}
+	flusher, ok := output.(http.Flusher)
+	if !ok {
+		return "", fmt.Errorf("cannot stream the output, output needs to be http.Flusher")
+	}
+
+	wf := writeFlusher{w: output, flusher: flusher}
+	reader := bufio.NewReader(rPipe)
+	wPipe.Close()
+	for {
+		line, err := reader.ReadBytes('\n')
+		if err != nil && err != io.EOF {
+			return "", fmt.Errorf("cannot read bytes from monitor pipe: %w", err)
+		}
+
+		// handle the case where we have a (final) line without a \n, here we
+		// get a EOF but still have line content to process before we can exit
+		// Note that we have not seen this in practise, its for compat with the
+		// previous bufio.Scanner implementation that had this behavior.
+		if err == io.EOF && len(line) == 0 {
+			break
+		}
+
+		_, err = wf.Write(line)
+		if err != nil {
+			return "", fmt.Errorf("cannot write bytes to writeFlusher: %w", err)
+		}
+	}
 
 	if err := cmd.Wait(); err != nil {
 		// we cannot use "http.Error()" here because the http
 		// header was already set to "201" when we started streaming
-		_, _ = mw.Write([]byte(fmt.Sprintf("cannot run osbuild: %v", err)))
+		_, _ = wf.Write([]byte(fmt.Sprintf("cannot run osbuild: %v", err)))
 		return "", err
 	}
 
 	// the result is put into a tar because we get sparse file support for free this way
 	// #nosec G204
-	cmd = exec.Command(
+	tarCmd := exec.Command(
 		"tar",
 		"--exclude=output/output.tar",
 		"-Scf",
 		filepath.Join(outputDir, "output.tar"),
 		"output",
 	)
-	cmd.Dir = buildDir
-	out, err := cmd.CombinedOutput()
+	tarCmd.Dir = buildDir
+	out, err := tarCmd.CombinedOutput()
 	if err != nil {
 		err = fmt.Errorf("cannot tar output directory: %w, output:\n%s", err, out)
 		logger.Errorf("%v", err)
-		_, _ = mw.Write([]byte(err.Error()))
+		_, _ = wf.Write([]byte(err.Error()))
 		return "", err
 	}
 	if len(out) > 0 {
@@ -291,12 +326,12 @@ func handleBuild(logger *logrus.Logger, config *Config) http.Handler {
 
 			// run osbuild and stream the output to the client
 			buildResult := newBuildResult(config)
-			_, err = runOsbuild(logger, buildDir, control, w)
+			_, err = runOSBuild(logger, buildDir, control, w)
 			if werr := buildResult.Mark(err); werr != nil {
 				logger.Errorf("cannot write result file %v", werr)
 			}
 			if err != nil {
-				logger.Errorf("canot run osbuild: %v", err)
+				logger.Errorf("cannot run osbuild: %v", err)
 				return
 			}
 		},

cmd/osbuild-worker-executor/handler_build_test.go

@@ -84,12 +84,18 @@ func TestBuildIntegration(t *testing.T) {
 	baseURL, baseBuildDir, loggerHook := runTestServer(t)
 	endpoint := baseURL + "api/v1/build"
 
-	// osbuild is called with --export tree and then the manifest.json
 	restore := main.MockOsbuildBinary(t, fmt.Sprintf(`#!/bin/sh -e
+# make sure the monitor is setup correctly
+>&3 echo '^^{"message": "osbuild-stage-message 1"}'
+>&3 echo '^^{"message": "osbuild-stage-message 2"}'
+>&3 echo '^^{"message": "osbuild-stage-message 3"}'
+
 # echo our inputs for the test to validate
-echo fake-osbuild "$1" "$2" "$3" "$4" "$5" "$6" "$7"
+echo osbuild $@
 echo ---
-cat "$8"
+# stdin
+cat -
+echo
 
 test "$MY" = "env"
 
@@ -104,18 +110,23 @@ echo "fake-build-result" > %[1]s/build/output/image/disk.img
 	assert.NoError(t, err)
 	defer func() { _, _ = io.ReadAll(rsp.Body) }()
 	defer rsp.Body.Close()
-
 	assert.Equal(t, http.StatusCreated, rsp.StatusCode)
-	reader := bufio.NewReader(rsp.Body)
 
-	// check that we get the output of osbuild streamed to us
-	expectedContent := fmt.Sprintf(`fake-osbuild --export tree --output-dir %[1]s/build/output --store %[1]s/build/osbuild-store --json
----
-{"fake": "manifest"}`, baseBuildDir)
+	// check that we get the monitor output of osbuild streamed to us
+	expectedMonitorContent := `^^{"message": "osbuild-stage-message 1"}
+^^{"message": "osbuild-stage-message 2"}
+^^{"message": "osbuild-stage-message 3"}
+`
+	reader := bufio.NewReader(rsp.Body)
 	content, err := io.ReadAll(reader)
 	assert.NoError(t, err)
-	assert.Equal(t, expectedContent, string(content))
+	assert.Equal(t, expectedMonitorContent, string(content))
+
 	// check log too
+	expectedContent := fmt.Sprintf(`osbuild --store %[1]s/build/osbuild-store --output-directory %[1]s/build/output --cache-max-size=21474836480 - --export tree --monitor=JSONSeqMonitor --monitor-fd=3
+---
+{"fake": "manifest"}
+`, baseBuildDir)
 	logFileContent, err := os.ReadFile(filepath.Join(baseBuildDir, "build/build.log"))
 	assert.NoError(t, err)
 	assert.Equal(t, expectedContent, string(logFileContent))
@@ -245,10 +256,7 @@ exit 23
 	reader := bufio.NewReader(rsp.Body)
 	content, err := io.ReadAll(reader)
 	assert.NoError(t, err)
-	expectedContent := `err on stdout
-err on stderr
-cannot run osbuild: exit status 23`
-	assert.Equal(t, expectedContent, string(content))
+	assert.Equal(t, "cannot run osbuild: exit status 23", string(content))
 
 	// check that the result is an error and we get the log
 	endpoint = baseURL + "api/v1/result/image/disk.img"
@@ -259,7 +267,10 @@ cannot run osbuild: exit status 23`
 	reader = bufio.NewReader(rsp.Body)
 	content, err = io.ReadAll(reader)
 	assert.NoError(t, err)
-	assert.Equal(t, "build failed\n"+expectedContent, string(content))
+	assert.Equal(t, `build failed
+err on stdout
+err on stderr
+`, string(content))
 }
 
 func TestBuildStreamsOutput(t *testing.T) {
@@ -269,7 +280,7 @@ func TestBuildStreamsOutput(t *testing.T) {
 	restore := main.MockOsbuildBinary(t, fmt.Sprintf(`#!/bin/sh -e
 for i in $(seq 3); do
    # generate the exact timestamp of the output line
-   echo "line-$i: $(date  +'%%s.%%N')"
+   >&3 echo "line-$i: $(date  +'%%s.%%N')"
    sleep 0.2
 done