internal/osbuildexecutor: forward progress to composer

Uses status scanner to monitor osbuild to generate progress
updates. Progress updates are sent every 30 seconds.

Only one level of subprogress needs to be supported. Because osbuild
manifests consist of pipelines and stages, more than 2 levels of
progress are not expected.

Author: croissanne

cmd/osbuild-worker/jobimpl-osbuild.go

@@ -527,7 +527,7 @@ func (impl *OSBuildJobImpl) Run(job worker.Job) error {
 		Stderr:     os.Stderr,
 		JSONOutput: true,
 	}
-	osbuildJobResult.OSBuildOutput, err = executor.RunOSBuild(jobArgs.Manifest, opts)
+	osbuildJobResult.OSBuildOutput, err = executor.RunOSBuild(jobArgs.Manifest, logWithId, job, opts)
 	// First handle the case when "running" osbuild failed
 	if err != nil {
 		osbuildJobResult.JobError = clienterrors.New(clienterrors.ErrorBuildJob, "osbuild build failed", err.Error())

internal/osbuildexecutor/osbuild-executor.go

@@ -1,9 +1,18 @@
 package osbuildexecutor
 
 import (
+	"time"
+
+	"github.com/sirupsen/logrus"
+
 	"github.com/osbuild/images/pkg/osbuild"
+	"github.com/osbuild/osbuild-composer/internal/worker"
+)
+
+const (
+	MinTimeBetweenUpdates = time.Second * 30
 )
 
 type Executor interface {
-	RunOSBuild(manifest []byte, opts *osbuild.OSBuildOptions) (*osbuild.Result, error)
+	RunOSBuild(manifest []byte, logger logrus.FieldLogger, job worker.Job, opts *osbuild.OSBuildOptions) (*osbuild.Result, error)
 }

internal/osbuildexecutor/runner-impl-aws-ec2.go

@@ -11,14 +11,16 @@ import (
 	"os/exec"
 	"path/filepath"
 	"strings"
+	"sync"
 	"time"
 
+	"github.com/sirupsen/logrus"
 	"golang.org/x/exp/slices"
 
 	"github.com/osbuild/images/pkg/osbuild"
-	"github.com/sirupsen/logrus"
 
 	"github.com/osbuild/osbuild-composer/internal/cloud/awscloud"
+	"github.com/osbuild/osbuild-composer/internal/worker"
 )
 
 type awsEC2Executor struct {
@@ -28,9 +30,9 @@ type awsEC2Executor struct {
 	tmpDir     string
 }
 
-func prepareSources(manifest []byte, opts *osbuild.OSBuildOptions) error {
+func prepareSources(manifest []byte, logger logrus.FieldLogger, opts *osbuild.OSBuildOptions) error {
 	hostExecutor := NewHostExecutor()
-	_, err := hostExecutor.RunOSBuild(manifest, &osbuild.OSBuildOptions{
+	_, err := hostExecutor.RunOSBuild(manifest, logger, nil, &osbuild.OSBuildOptions{
 		StoreDir:   opts.StoreDir,
 		ExtraEnv:   opts.ExtraEnv,
 		Stderr:     opts.Stderr,
@@ -123,42 +125,89 @@ func writeInputArchive(cacheDir, store string, exports []string, manifestData []
 	return archive, nil
 }
 
-func handleBuild(inputArchive, host string) (*osbuild.Result, error) {
+func handleBuild(inputArchive, host string, logger logrus.FieldLogger, job worker.Job) (*osbuild.Result, error) {
 	client := http.Client{
 		Timeout: time.Minute * 60,
 	}
 	inputFile, err := os.Open(inputArchive)
 	if err != nil {
-		return nil, fmt.Errorf("Unable to open inputArchive (%s): %w", inputArchive, err)
+		return nil, fmt.Errorf("unable to open inputArchive (%s): %w", inputArchive, err)
 	}
 	defer inputFile.Close()
 
 	resp, err := client.Post(fmt.Sprintf("%s/api/v1/build", host), "application/x-tar", inputFile)
 	if err != nil {
-		return nil, fmt.Errorf("Unable to request build from executor instance: %w", err)
+		return nil, fmt.Errorf("unable to request build from executor instance: %w", err)
 	}
 	defer resp.Body.Close()
 	if resp.StatusCode != 201 {
 		body, err := io.ReadAll(resp.Body)
 		if err != nil {
-			return nil, fmt.Errorf("Unable to read body waiting for build to run: %w,  http status: %d", err, resp.StatusCode)
+			return nil, fmt.Errorf("unable to read body waiting for build to run: %w,  http status: %d", err, resp.StatusCode)
 		}
-		return nil, fmt.Errorf("Something went wrong during executor build: http status: %v, %d, %s", err, resp.StatusCode, body)
+		return nil, fmt.Errorf("something went wrong during executor build: http status: %v, %d, %s", err, resp.StatusCode, body)
 	}
 
-	var osbuildResult osbuild.Result
-
-	body, err := io.ReadAll(resp.Body)
-	if err != nil {
-		return nil, fmt.Errorf("Unable to read response body: %w", err)
-	}
+	osbuildStatus := osbuild.NewStatusScanner(resp.Body)
+	var lastUpdated time.Time
+	var wg sync.WaitGroup
+	for {
+		st, err := osbuildStatus.Status()
+		if err != nil {
+			return nil, fmt.Errorf(`error parsing osbuild status, please report a bug: %w`, err)
+		}
+		if st == nil {
+			break
+		}
 
-	err = json.Unmarshal(body, &osbuildResult)
-	if err != nil {
-		return nil, fmt.Errorf("Unable to decode response body %q into osbuild result: %w", body, err)
+		var progress string
+		if st.Progress != nil {
+			var subProgress string
+			if st.Progress.SubProgress != nil {
+				subProgress = fmt.Sprintf(": %d/%d", st.Progress.SubProgress.Done, st.Progress.SubProgress.Total)
+			}
+			progress = fmt.Sprintf(" (step %d/%d%s)", st.Progress.Done, st.Progress.Total, subProgress)
+		}
+		if st.Message != "" {
+			logger.Infof("OSBuild status: %s%s", st.Message, progress)
+			if job != nil && time.Since(lastUpdated) > MinTimeBetweenUpdates {
+				lastUpdated = time.Now()
+				wg.Add(1)
+				go func() {
+					defer wg.Done()
+					partial := worker.JobResult{
+						Progress: &worker.JobProgress{
+							Message: st.Message,
+						},
+					}
+					if st.Progress != nil {
+						partial.Progress.Done = st.Progress.Done
+						partial.Progress.Total = st.Progress.Total
+					}
+					// more than 1 level of subprogress is not expected, just
+					// pipelines and stages.
+					if st.Progress.SubProgress != nil {
+						partial.Progress.SubProgress = &worker.JobProgress{
+							Message: st.Progress.SubProgress.Message,
+							Done:    st.Progress.SubProgress.Done,
+							Total:   st.Progress.SubProgress.Total,
+						}
+					}
+					err := job.Update(partial)
+					if err != nil {
+						logger.Errorf("Unable to update job: %s", err.Error())
+					}
+				}()
+			}
+		}
+		if st.Trace != "" {
+			logger.Errorf("%s", st.Trace)
+		}
 	}
 
-	return &osbuildResult, nil
+	// avoid a partial job updates after this function has returned
+	wg.Wait()
+	return osbuildStatus.Result()
 }
 
 func fetchOutputArchive(cacheDir, host string) (string, error) {
@@ -248,8 +297,8 @@ func extractOutputArchive(outputDirectory, outputTar string) error {
 
 }
 
-func (ec2e *awsEC2Executor) RunOSBuild(manifest []byte, opts *osbuild.OSBuildOptions) (*osbuild.Result, error) {
-	err := prepareSources(manifest, opts)
+func (ec2e *awsEC2Executor) RunOSBuild(manifest []byte, logger logrus.FieldLogger, job worker.Job, opts *osbuild.OSBuildOptions) (*osbuild.Result, error) {
+	err := prepareSources(manifest, logger, opts)
 	if err != nil {
 		return nil, fmt.Errorf("Failed to prepare sources: %w", err)
 	}
@@ -289,9 +338,9 @@ func (ec2e *awsEC2Executor) RunOSBuild(manifest []byte, opts *osbuild.OSBuildOpt
 		return nil, err
 	}
 
-	osbuildResult, err := handleBuild(inputArchive, executorHost)
+	osbuildResult, err := handleBuild(inputArchive, executorHost, logger, job)
 	if err != nil {
-		logrus.Errorf("Something went wrong handling the executor's build: %v", err)
+		logrus.Errorf("something went wrong handling the executor's build: %v", err)
 		return nil, err
 	}
 	if !osbuildResult.Success {

internal/osbuildexecutor/runner-impl-aws-ec2_test.go

@@ -3,7 +3,6 @@ package osbuildexecutor_test
 import (
 	"archive/tar"
 	"context"
-	"encoding/json"
 	"io"
 	"net/http"
 	"net/http/httptest"
@@ -14,13 +13,21 @@ import (
 	"testing"
 	"time"
 
-	"github.com/osbuild/images/pkg/osbuild"
+	"github.com/google/uuid"
+	"github.com/sirupsen/logrus"
+	"github.com/sirupsen/logrus/hooks/test"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
 	"github.com/osbuild/osbuild-composer/internal/osbuildexecutor"
+	"github.com/osbuild/osbuild-composer/internal/worker"
 )
 
+func makeMockEntry() (*logrus.Entry, *test.Hook) {
+	logger, hook := test.NewNullLogger()
+	return logger.WithField("test", "test"), hook
+}
+
 func TestWaitForSI(t *testing.T) {
 	server := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		w.WriteHeader(http.StatusOK)
@@ -66,29 +73,113 @@ func TestWriteInputArchive(t *testing.T) {
 	}, strings.Split(string(out), "\n"))
 }
 
+type testJob struct {
+	PartialUpdates []interface{}
+}
+
+func (j *testJob) Id() uuid.UUID {
+	return uuid.Nil
+}
+
+func (j *testJob) Type() string {
+	return "test-job"
+}
+
+func (j *testJob) Args(args interface{}) error {
+	return nil
+}
+
+func (j *testJob) DynamicArgs(i int, args interface{}) error {
+	return nil
+}
+
+func (j *testJob) NDynamicArgs() int {
+	return 0
+}
+
+func (j *testJob) Update(result interface{}) error {
+	j.PartialUpdates = append(j.PartialUpdates, result)
+	return nil
+}
+
+func (j *testJob) Finish(result interface{}) error {
+	return nil
+}
+
+func (j *testJob) Canceled() (bool, error) {
+	return false, nil
+}
+
+func (j *testJob) UploadArtifact(name string, readSeeker io.ReadSeeker) error {
+	return nil
+}
+
 func TestHandleBuild(t *testing.T) {
 	buildServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		input, err := io.ReadAll(r.Body)
 		require.NoError(t, err)
 		require.Equal(t, []byte("test"), input)
 
 		w.WriteHeader(http.StatusCreated)
-		osbuildResult := osbuild.Result{
-			Success: true,
-		}
-		data, err := json.Marshal(osbuildResult)
-		require.NoError(t, err)
-		_, err = w.Write(data)
+		_, err = w.Write([]byte(`{"message": "starting pipeline", "context": {"origin": "osbuild.monitor", "pipeline": {"name": "source", "id": "pipeline-id", "stage": {}}, "id": "context-id"}, "progress": {"name": "pipelines/sources", "total": 1, "done": 0}}
+{"message": "stage in pipeline", "context": {"id": "context-id"}, "progress": {"name": "source", "total": 1, "done": 0, "progress": {"name": "source/stage", "total": 2, "done": 0}}}
+{"message": "finishing pipeline", "result": {"name": "source", "id": "pipeline-id", "success": true}, "context": {"id": "context-id"}, "progress": {"name": "source", "total": 1, "done": 1}}`))
 		require.NoError(t, err)
 	}))
 
 	cacheDir := t.TempDir()
 	inputArchive := filepath.Join(cacheDir, "test.tar")
 	require.NoError(t, os.WriteFile(inputArchive, []byte("test"), 0600))
 
-	osbuildResult, err := osbuildexecutor.HandleBuild(inputArchive, buildServer.URL)
+	entry, hook := makeMockEntry()
+	job := testJob{}
+	osbuildResult, err := osbuildexecutor.HandleBuild(inputArchive, buildServer.URL, entry, &job)
 	require.NoError(t, err)
 	require.True(t, osbuildResult.Success)
+	require.Len(t, hook.Entries, 3)
+	require.Equal(t, "OSBuild status: starting pipeline (step 0/1)", hook.Entries[0].Message)
+	require.Equal(t, "OSBuild status: stage in pipeline (step 0/1: 0/2)", hook.Entries[1].Message)
+	require.Equal(t, "OSBuild status: finishing pipeline (step 1/1)", hook.Entries[2].Message)
+
+	// only 1 update every 30 seconds, so in a testing scenario only the first status should have been received
+	require.Eventually(t, func() bool { return len(job.PartialUpdates) == 1 }, time.Second, time.Millisecond*10)
+	partial, ok := job.PartialUpdates[0].(worker.JobResult)
+	require.True(t, ok)
+	require.Equal(t, worker.JobResult{
+		Progress: &worker.JobProgress{
+			Done:    0,
+			Total:   1,
+			Message: "starting pipeline",
+		},
+	}, partial)
+	require.Len(t, job.PartialUpdates, 1)
+}
+
+func TestHandleBuildTrace(t *testing.T) {
+	buildServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		input, err := io.ReadAll(r.Body)
+		require.NoError(t, err)
+		require.Equal(t, []byte("test"), input)
+
+		w.WriteHeader(http.StatusCreated)
+		_, err = w.Write([]byte(`{"message": "starting pipeline", "context": {"origin": "osbuild.monitor", "pipeline": {"name": "source", "id": "pipeline-id", "stage": {}}, "id": "context-id"}, "progress": {"name": "pipelines/sources", "total": 1, "done": 0}}
+{"message": "no context, thus trace"}
+{"message": "failed pipeline", "result": {"name": "source", "id": "pipeline-id", "success": false}, "context": {"id": "context-id"}, "progress": {"name": "source", "total": 1, "done": 1}}`))
+		require.NoError(t, err)
+	}))
+
+	cacheDir := t.TempDir()
+	inputArchive := filepath.Join(cacheDir, "test.tar")
+	require.NoError(t, os.WriteFile(inputArchive, []byte("test"), 0600))
+
+	entry, hook := makeMockEntry()
+	osbuildResult, err := osbuildexecutor.HandleBuild(inputArchive, buildServer.URL, entry, nil)
+	require.NoError(t, err)
+	require.False(t, osbuildResult.Success)
+	require.Len(t, hook.Entries, 3)
+	require.Equal(t, "OSBuild status: starting pipeline (step 0/1)", hook.Entries[0].Message)
+	require.Equal(t, "no context, thus trace", hook.Entries[1].Message)
+	require.Equal(t, "OSBuild status: failed pipeline (step 1/1)", hook.Entries[2].Message)
 }
 
 func TestHandleBuildNoJSON(t *testing.T) {
@@ -105,8 +196,9 @@ func TestHandleBuildNoJSON(t *testing.T) {
 	inputArchive := filepath.Join(cacheDir, "test.tar")
 	require.NoError(t, os.WriteFile(inputArchive, []byte("test"), 0600))
 
-	_, err := osbuildexecutor.HandleBuild(inputArchive, buildServer.URL)
-	require.ErrorContains(t, err, `Unable to decode response body "bad non-json text" into osbuild result:`)
+	entry, _ := makeMockEntry()
+	_, err := osbuildexecutor.HandleBuild(inputArchive, buildServer.URL, entry, nil)
+	require.ErrorContains(t, err, `error parsing osbuild status, please report a bug: cannot scan line "bad non-json text": invalid character 'b' looking for beginning of value`)
 }
 
 func TestHandleOutputArchive(t *testing.T) {