Repository: osbuild/osbuild-composer
Pull request #467 (2020-04-04 12:32:50)

queue: cleanups
These commits don't change behavior. They're mostly cleanups and refactors around the `jobqueue` package.

larskarlitski: Update to run `gofmt`. Sorry.
larskarlitski: Schutzbot Fedora32 fails, because cockpit-composer failed to build.

codecov/patch failed, because this patch doesn't add tests for all the lines it adds. I think this shouldn't apply here though, because I'm just moving code around. 

Patch:
diff --git a/cmd/osbuild-worker/main.go b/cmd/osbuild-worker/main.go
index aeef105dd..dd8ddf93f 100644
--- a/cmd/osbuild-worker/main.go
+++ b/cmd/osbuild-worker/main.go
@@ -6,12 +6,16 @@ import (
 	"errors"
 	"flag"
 	"fmt"
+	"io"
 	"io/ioutil"
 	"log"
 	"os"
+	"path"
 
 	"github.com/osbuild/osbuild-composer/internal/common"
 	"github.com/osbuild/osbuild-composer/internal/jobqueue"
+	"github.com/osbuild/osbuild-composer/internal/target"
+	"github.com/osbuild/osbuild-composer/internal/upload/awsupload"
 )
 
 type connectionConfig struct {
@@ -43,26 +47,85 @@ func createTLSConfig(config *connectionConfig) (*tls.Config, error) {
 	}, nil
 }
 
-func handleJob(client *jobqueue.Client) error {
-	fmt.Println("Waiting for a new job...")
-	job, err := client.AddJob()
-	if err != nil {
-		return err
+type TargetsError struct {
+	Errors []error
+}
+
+func (e *TargetsError) Error() string {
+	errString := fmt.Sprintf("%d target(s) errored:\n", len(e.Errors))
+
+	for _, err := range e.Errors {
+		errString += err.Error() + "\n"
 	}
 
-	err = client.UpdateJob(job, common.IBRunning, nil)
+	return errString
+}
+
+func RunJob(job *jobqueue.Job, uploadFunc func(*jobqueue.Job, io.Reader) error) (*common.ComposeResult, error) {
+	tmpStore, err := ioutil.TempDir("/var/tmp", "osbuild-store")
 	if err != nil {
-		return err
+		return nil, fmt.Errorf("error setting up osbuild store: %v", err)
 	}
+	// FIXME: how to handle errors in defer?
+	defer os.RemoveAll(tmpStore)
 
-	fmt.Printf("Running job %s\n", job.ID.String())
-	result, err := job.Run(client)
+	result, err := RunOSBuild(job.Manifest, tmpStore, os.Stderr)
 	if err != nil {
-		log.Printf("  Job failed: %v", err)
-		return client.UpdateJob(job, common.IBFailed, result)
+		return nil, fmt.Errorf("osbuild error: %v", err)
 	}
 
-	return client.UpdateJob(job, common.IBFinished, result)
+	var r []error
+
+	for _, t := range job.Targets {
+		switch options := t.Options.(type) {
+		case *target.LocalTargetOptions:
+			f, err := os.Open(path.Join(tmpStore, "refs", result.OutputID, options.Filename))
+			if err != nil {
+				r = append(r, err)
+				continue
+			}
+
+			err = uploadFunc(job, f)
+			if err != nil {
+				r = append(r, err)
+				continue
+			}
+
+		case *target.AWSTargetOptions:
+
+			a, err := awsupload.New(options.Region, options.AccessKeyID, options.SecretAccessKey)
+			if err != nil {
+				r = append(r, err)
+				continue
+			}
+
+			if options.Key == "" {
+				options.Key = job.ComposeID.String()
+			}
+
+			_, err = a.Upload(path.Join(tmpStore, "refs", result.OutputID, options.Filename), options.Bucket, options.Key)
+			if err != nil {
+				r = append(r, err)
+				continue
+			}
+
+			/* TODO: communicate back the AMI */
+			_, err = a.Register(t.ImageName, options.Bucket, options.Key)
+			if err != nil {
+				r = append(r, err)
+				continue
+			}
+		case *target.AzureTargetOptions:
+		default:
+			r = append(r, fmt.Errorf("invalid target type"))
+		}
+	}
+
+	if len(r) > 0 {
+		return result, &TargetsError{r}
+	}
+
+	return result, nil
 }
 
 func main() {
@@ -99,8 +162,26 @@ func main() {
 	}
 
 	for {
-		if err := handleJob(client); err != nil {
-			log.Fatalf("Failed to handle job: " + err.Error())
+		fmt.Println("Waiting for a new job...")
+		job, err := client.AddJob()
+		if err != nil {
+			log.Fatal(err)
+		}
+
+		fmt.Printf("Running job %s\n", job.ComposeID.String())
+
+		var status common.ImageBuildState
+		result, err := RunJob(job, client.UploadImage)
+		if err != nil {
+			log.Printf("  Job failed: %v", err)
+			status = common.IBFailed
+		} else {
+			status = common.IBFinished
+		}
+
+		err = client.UpdateJob(job, status, result)
+		if err != nil {
+			log.Fatalf("Error reporting job result: %v", err)
 		}
 	}
 }
diff --git a/cmd/osbuild-worker/osbuild.go b/cmd/osbuild-worker/osbuild.go
new file mode 100644
index 000000000..7cb032951
--- /dev/null
+++ b/cmd/osbuild-worker/osbuild.go
@@ -0,0 +1,55 @@
+package main
+
+import (
+	"encoding/json"
+	"fmt"
+	"io"
+	"os/exec"
+
+	"github.com/osbuild/osbuild-composer/internal/common"
+	"github.com/osbuild/osbuild-composer/internal/osbuild"
+)
+
+func RunOSBuild(manifest *osbuild.Manifest, store string, errorWriter io.Writer) (*common.ComposeResult, error) {
+	cmd := exec.Command(
+		"osbuild",
+		"--store", store,
+		"--json", "-",
+	)
+	cmd.Stderr = errorWriter
+
+	stdin, err := cmd.StdinPipe()
+	if err != nil {
+		return nil, fmt.Errorf("error setting up stdin for osbuild: %v", err)
+	}
+
+	stdout, err := cmd.StdoutPipe()
+	if err != nil {
+		return nil, fmt.Errorf("error setting up stdout for osbuild: %v", err)
+	}
+
+	err = cmd.Start()
+	if err != nil {
+		return nil, fmt.Errorf("error starting osbuild: %v", err)
+	}
+
+	err = json.NewEncoder(stdin).Encode(manifest)
+	if err != nil {
+		return nil, fmt.Errorf("error encoding osbuild pipeline: %v", err)
+	}
+	// FIXME: handle or comment this possible error
+	_ = stdin.Close()
+
+	var result common.ComposeResult
+	err = json.NewDecoder(stdout).Decode(&result)
+	if err != nil {
+		return nil, fmt.Errorf("error decoding osbuild output: %#v", err)
+	}
+
+	err = cmd.Wait()
+	if err != nil {
+		return &result, err
+	}
+
+	return &result, nil
+}
diff --git a/internal/jobqueue/api.go b/internal/jobqueue/api.go
index bd017c3e4..4196c0f0b 100644
--- a/internal/jobqueue/api.go
+++ b/internal/jobqueue/api.go
@@ -7,10 +7,10 @@ import (
 	"net/http"
 	"strconv"
 
-	"github.com/osbuild/osbuild-composer/internal/store"
-
 	"github.com/google/uuid"
 	"github.com/julienschmidt/httprouter"
+
+	"github.com/osbuild/osbuild-composer/internal/store"
 )
 
 type API struct {
@@ -102,8 +102,8 @@ func (api *API) addJobHandler(writer http.ResponseWriter, request *http.Request,
 
 	writer.WriteHeader(http.StatusCreated)
 	// FIXME: handle or comment this possible error
-	_ = json.NewEncoder(writer).Encode(Job{
-		ID:           nextJob.ComposeID,
+	_ = json.NewEncoder(writer).Encode(addJobResponse{
+		ComposeID:    nextJob.ComposeID,
 		ImageBuildID: nextJob.ImageBuildID,
 		Manifest:     nextJob.Manifest,
 		Targets:      nextJob.Targets,
@@ -130,7 +130,7 @@ func (api *API) updateJobHandler(writer http.ResponseWriter, request *http.Reque
 		return
 	}
 
-	var body JobStatus
+	var body updateJobRequest
 	err = json.NewDecoder(request.Body).Decode(&body)
 	if err != nil {
 		statusResponseError(writer, http.StatusBadRequest, "invalid status: "+err.Error())
diff --git a/internal/jobqueue/client.go b/internal/jobqueue/client.go
index 6bd49b295..b9f1f4f1f 100644
--- a/internal/jobqueue/client.go
+++ b/internal/jobqueue/client.go
@@ -12,7 +12,11 @@ import (
 	"net"
 	"net/http"
 
+	"github.com/google/uuid"
+
 	"github.com/osbuild/osbuild-composer/internal/common"
+	"github.com/osbuild/osbuild-composer/internal/osbuild"
+	"github.com/osbuild/osbuild-composer/internal/target"
 )
 
 type Client struct {
@@ -21,6 +25,13 @@ type Client struct {
 	hostname string
 }
 
+type Job struct {
+	ComposeID    uuid.UUID
+	ImageBuildID int
+	Manifest     *osbuild.Manifest
+	Targets      []*target.Target
+}
+
 func NewClient(address string, conf *tls.Config) *Client {
 	client := &http.Client{
 		Transport: &http.Transport{
@@ -71,22 +82,27 @@ func (c *Client) AddJob() (*Job, error) {
 		return nil, fmt.Errorf("couldn't create job, got %d: %s", response.StatusCode, r)
 	}
 
-	job := &Job{}
-	err = json.NewDecoder(response.Body).Decode(job)
+	var jr addJobResponse
+	err = json.NewDecoder(response.Body).Decode(&jr)
 	if err != nil {
 		return nil, err
 	}
 
-	return job, nil
+	return &Job{
+		jr.ComposeID,
+		jr.ImageBuildID,
+		jr.Manifest,
+		jr.Targets,
+	}, nil
 }
 
 func (c *Client) UpdateJob(job *Job, status common.ImageBuildState, result *common.ComposeResult) error {
 	var b bytes.Buffer
-	err := json.NewEncoder(&b).Encode(&JobStatus{status, result})
+	err := json.NewEncoder(&b).Encode(&updateJobRequest{status, result})
 	if err != nil {
 		panic(err)
 	}
-	urlPath := fmt.Sprintf("/job-queue/v1/jobs/%s/builds/%d", job.ID.String(), job.ImageBuildID)
+	urlPath := fmt.Sprintf("/job-queue/v1/jobs/%s/builds/%d", job.ComposeID.String(), job.ImageBuildID)
 	url := c.createURL(urlPath)
 	req, err := http.NewRequest("PATCH", url, &b)
 	if err != nil {
@@ -109,7 +125,7 @@ func (c *Client) UpdateJob(job *Job, status common.ImageBuildState, result *comm
 
 func (c *Client) UploadImage(job *Job, reader io.Reader) error {
 	// content type doesn't really matter
-	url := c.createURL(fmt.Sprintf("/job-queue/v1/jobs/%s/builds/%d/image", job.ID.String(), job.ImageBuildID))
+	url := c.createURL(fmt.Sprintf("/job-queue/v1/jobs/%s/builds/%d/image", job.ComposeID.String(), job.ImageBuildID))
 	_, err := c.client.Post(url, "application/octet-stream", reader)
 
 	return err
diff --git a/internal/jobqueue/job.go b/internal/jobqueue/job.go
deleted file mode 100644
index 9de7592ef..000000000
--- a/internal/jobqueue/job.go
+++ /dev/null
@@ -1,146 +0,0 @@
-package jobqueue
-
-import (
-	"encoding/json"
-	"fmt"
-	"io/ioutil"
-	"os"
-	"os/exec"
-	"path"
-
-	"github.com/osbuild/osbuild-composer/internal/osbuild"
-
-	"github.com/google/uuid"
-
-	"github.com/osbuild/osbuild-composer/internal/common"
-	"github.com/osbuild/osbuild-composer/internal/target"
-	"github.com/osbuild/osbuild-composer/internal/upload/awsupload"
-)
-
-type Job struct {
-	ID           uuid.UUID         `json:"id"`
-	ImageBuildID int               `json:"image_build_id"`
-	Manifest     *osbuild.Manifest `json:"manifest"`
-	Targets      []*target.Target  `json:"targets"`
-}
-
-type JobStatus struct {
-	Status common.ImageBuildState `json:"status"`
-	Result *common.ComposeResult  `json:"result"`
-}
-
-type TargetsError struct {
-	Errors []error
-}
-
-func (e *TargetsError) Error() string {
-	errString := fmt.Sprintf("%d target(s) errored:\n", len(e.Errors))
-
-	for _, err := range e.Errors {
-		errString += err.Error() + "\n"
-	}
-
-	return errString
-}
-
-func (job *Job) Run(uploader LocalTargetUploader) (*common.ComposeResult, error) {
-	tmpStore, err := ioutil.TempDir("/var/tmp", "osbuild-store")
-	if err != nil {
-		return nil, fmt.Errorf("error setting up osbuild store: %v", err)
-	}
-	// FIXME: how to handle errors in defer?
-	defer os.RemoveAll(tmpStore)
-
-	cmd := exec.Command(
-		"osbuild",
-		"--store", tmpStore,
-		"--json", "-",
-	)
-	cmd.Stderr = os.Stderr
-
-	stdin, err := cmd.StdinPipe()
-	if err != nil {
-		return nil, fmt.Errorf("error setting up stdin for osbuild: %v", err)
-	}
-
-	stdout, err := cmd.StdoutPipe()
-	if err != nil {
-		return nil, fmt.Errorf("error setting up stdout for osbuild: %v", err)
-	}
-
-	err = cmd.Start()
-	if err != nil {
-		return nil, fmt.Errorf("error starting osbuild: %v", err)
-	}
-
-	err = json.NewEncoder(stdin).Encode(job.Manifest)
-	if err != nil {
-		return nil, fmt.Errorf("error encoding osbuild pipeline: %v", err)
-	}
-	// FIXME: handle or comment this possible error
-	_ = stdin.Close()
-
-	var result common.ComposeResult
-	err = json.NewDecoder(stdout).Decode(&result)
-	if err != nil {
-		return nil, fmt.Errorf("error decoding osbuild output: %#v", err)
-	}
-
-	err = cmd.Wait()
-	if err != nil {
-		return &result, err
-	}
-
-	var r []error
-
-	for _, t := range job.Targets {
-		switch options := t.Options.(type) {
-		case *target.LocalTargetOptions:
-			f, err := os.Open(path.Join(tmpStore, "refs", result.OutputID, options.Filename))
-			if err != nil {
-				r = append(r, err)
-				continue
-			}
-
-			err = uploader.UploadImage(job, f)
-			if err != nil {
-				r = append(r, err)
-				continue
-			}
-
-		case *target.AWSTargetOptions:
-
-			a, err := awsupload.New(options.Region, options.AccessKeyID, options.SecretAccessKey)
-			if err != nil {
-				r = append(r, err)
-				continue
-			}
-
-			if options.Key == "" {
-				options.Key = job.ID.String()
-			}
-
-			_, err = a.Upload(path.Join(tmpStore, "refs", result.OutputID, options.Filename), options.Bucket, options.Key)
-			if err != nil {
-				r = append(r, err)
-				continue
-			}
-
-			/* TODO: communicate back the AMI */
-			_, err = a.Register(t.ImageName, options.Bucket, options.Key)
-			if err != nil {
-				r = append(r, err)
-				continue
-			}
-		case *target.AzureTargetOptions:
-		default:
-			r = append(r, fmt.Errorf("invalid target type"))
-		}
-	}
-
-	if len(r) > 0 {
-		return &result, &TargetsError{r}
-	}
-
-	return &result, nil
-}
diff --git a/internal/jobqueue/json.go b/internal/jobqueue/json.go
new file mode 100644
index 000000000..f324479cd
--- /dev/null
+++ b/internal/jobqueue/json.go
@@ -0,0 +1,21 @@
+package jobqueue
+
+import (
+	"github.com/google/uuid"
+
+	"github.com/osbuild/osbuild-composer/internal/common"
+	"github.com/osbuild/osbuild-composer/internal/osbuild"
+	"github.com/osbuild/osbuild-composer/internal/target"
+)
+
+type addJobResponse struct {
+	ComposeID    uuid.UUID         `json:"compose_id"`
+	ImageBuildID int               `json:"image_build_id"`
+	Manifest     *osbuild.Manifest `json:"manifest"`
+	Targets      []*target.Target  `json:"targets"`
+}
+
+type updateJobRequest struct {
+	Status common.ImageBuildState `json:"status"`
+	Result *common.ComposeResult  `json:"result"`
+}
diff --git a/internal/jobqueue/local_target_uploader.go b/internal/jobqueue/local_target_uploader.go
deleted file mode 100644
index 292b34733..000000000
--- a/internal/jobqueue/local_target_uploader.go
+++ /dev/null
@@ -1,7 +0,0 @@
-package jobqueue
-
-import "io"
-
-type LocalTargetUploader interface {
-	UploadImage(job *Job, reader io.Reader) error
-}


Test patch:
diff --git a/internal/jobqueue/api_test.go b/internal/jobqueue/api_test.go
index 382dccbe0..c1a28d413 100644
--- a/internal/jobqueue/api_test.go
+++ b/internal/jobqueue/api_test.go
@@ -4,13 +4,13 @@ import (
 	"net/http"
 	"testing"
 
+	"github.com/google/uuid"
+
 	"github.com/osbuild/osbuild-composer/internal/blueprint"
 	"github.com/osbuild/osbuild-composer/internal/distro/fedoratest"
 	"github.com/osbuild/osbuild-composer/internal/jobqueue"
 	"github.com/osbuild/osbuild-composer/internal/store"
 	"github.com/osbuild/osbuild-composer/internal/test"
-
-	"github.com/google/uuid"
 )
 
 func TestBasic(t *testing.T) {
@@ -57,7 +57,7 @@ func TestCreate(t *testing.T) {
 	}
 
 	test.TestRoute(t, api, false, "POST", "/job-queue/v1/jobs", `{}`, http.StatusCreated,
-		`{"id":"`+id.String()+`","image_build_id":0,"manifest":{"sources":{},"pipeline":{}},"targets":[]}`, "created")
+		`{"compose_id":"`+id.String()+`","image_build_id":0,"manifest":{"sources":{},"pipeline":{}},"targets":[]}`, "created")
 }
 
 func testUpdateTransition(t *testing.T, from, to string, expectedStatus int, expectedResponse string) {