Repository: osbuild/osbuild-composer
Pull request #998 (2020-09-20 18:22:09)

worker API: introduce job types and architecture
This is an effort to future-proof the worker API. It removes the assumption that all jobs are osbuild jobs and requires a worker to pass both the jobs it can handle and the architecture it runs on. 

larskarlitski: Update to resolve the conflict and set a correct list of accepted job types.
larskarlitski: > Thanks! (CI failures are, again, unrelated and caused by the ostree tests).

There was a failure in `aws.sh` as well, which might have been a flake. I restarted the tests to make sure.
larskarlitski: > There was a failure in `aws.sh` as well, which might have been a flake. I restarted the tests to make sure.

This was an actual bug (yay testing and thanks for tracking it down @ondrejbudai). I've copied the logic to assemble job types from `osbuild-composer` to `osbuild-composer-cloud`. I didn't bother sharing the code because we're about to merge the two.
larskarlitski: That last fix did it indeed. The only remaining failure is due to https://github.com/osbuild/osbuild-composer/issues/997.

Patch:
diff --git a/cmd/osbuild-composer-cloud/main.go b/cmd/osbuild-composer-cloud/main.go
index ad0beb2a7..6becccf82 100644
--- a/cmd/osbuild-composer-cloud/main.go
+++ b/cmd/osbuild-composer-cloud/main.go
@@ -106,18 +106,32 @@ func main() {
 		log.Fatalf("cannot create queue directory: %v", err)
 	}
 
-	jobs, err := fsjobqueue.New(queueDir, []string{"osbuild"})
+	distros, err := distro.NewRegistry(fedora31.New(), fedora32.New(), rhel8.New())
 	if err != nil {
-		log.Fatalf("cannot create jobqueue: %v", err)
+		log.Fatalf("Error loading distros: %v", err)
 	}
 
-	rpm := rpmmd.NewRPMMD(path.Join(cacheDirectory, "rpmmd"), "/usr/libexec/osbuild-composer/dnf-json")
+	// construct job types of the form osbuild:{arch} for all arches
+	jobTypes := []string{"osbuild"}
+	jobTypesMap := map[string]bool{}
+	for _, name := range distros.List() {
+		d := distros.GetDistro(name)
+		for _, arch := range d.ListArches() {
+			jt := "osbuild:" + arch
+			if !jobTypesMap[jt] {
+				jobTypesMap[jt] = true
+				jobTypes = append(jobTypes, jt)
+			}
+		}
+	}
 
-	distros, err := distro.NewRegistry(fedora31.New(), fedora32.New(), rhel8.New())
+	jobs, err := fsjobqueue.New(queueDir, jobTypes)
 	if err != nil {
-		log.Fatalf("Error loading distros: %v", err)
+		log.Fatalf("cannot create jobqueue: %v", err)
 	}
 
+	rpm := rpmmd.NewRPMMD(path.Join(cacheDirectory, "rpmmd"), "/usr/libexec/osbuild-composer/dnf-json")
+
 	workerServer := worker.NewServer(logger, jobs, "")
 	cloudServer := cloudapi.NewServer(workerServer, rpm, distros)
 
diff --git a/cmd/osbuild-composer/main.go b/cmd/osbuild-composer/main.go
index 20455bb2b..593d6daa5 100644
--- a/cmd/osbuild-composer/main.go
+++ b/cmd/osbuild-composer/main.go
@@ -180,7 +180,21 @@ func main() {
 		log.Fatalf("cannot create queue directory: %v", err)
 	}
 
-	jobs, err := fsjobqueue.New(queueDir, []string{"osbuild"})
+	// construct job types of the form osbuild:{arch} for all arches
+	jobTypes := []string{"osbuild"}
+	jobTypesMap := map[string]bool{}
+	for _, name := range distros.List() {
+		d := distros.GetDistro(name)
+		for _, arch := range d.ListArches() {
+			jt := "osbuild:" + arch
+			if !jobTypesMap[jt] {
+				jobTypesMap[jt] = true
+				jobTypes = append(jobTypes, jt)
+			}
+		}
+	}
+
+	jobs, err := fsjobqueue.New(queueDir, jobTypes)
 	if err != nil {
 		log.Fatalf("cannot create jobqueue: %v", err)
 	}
diff --git a/internal/cloudapi/server.go b/internal/cloudapi/server.go
index 178a98e99..8bfd909e9 100644
--- a/internal/cloudapi/server.go
+++ b/internal/cloudapi/server.go
@@ -69,6 +69,7 @@ func (server *Server) Compose(w http.ResponseWriter, r *http.Request) {
 
 	type imageRequest struct {
 		manifest distro.Manifest
+		arch     string
 	}
 	imageRequests := make([]imageRequest, len(request.ImageRequests))
 	var targets []*target.Target
@@ -128,6 +129,7 @@ func (server *Server) Compose(w http.ResponseWriter, r *http.Request) {
 		}
 
 		imageRequests[i].manifest = manifest
+		imageRequests[i].arch = arch.Name()
 
 		if len(ir.UploadRequests) != 1 {
 			http.Error(w, "Only compose requests with a single upload target are currently supported", http.StatusBadRequest)
@@ -179,7 +181,7 @@ func (server *Server) Compose(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	id, err := server.workers.Enqueue(ir.manifest, targets)
+	id, err := server.workers.Enqueue(ir.arch, ir.manifest, targets)
 	if err != nil {
 		http.Error(w, "Failed to enqueue manifest", http.StatusInternalServerError)
 		return
diff --git a/internal/kojiapi/server.go b/internal/kojiapi/server.go
index d77fcf462..1b2337ac2 100644
--- a/internal/kojiapi/server.go
+++ b/internal/kojiapi/server.go
@@ -95,6 +95,7 @@ func (h *apiHandlers) PostCompose(ctx echo.Context) error {
 
 	type imageRequest struct {
 		manifest distro.Manifest
+		arch     string
 		filename string
 	}
 	imageRequests := make([]imageRequest, len(request.ImageRequests))
@@ -135,6 +136,7 @@ func (h *apiHandlers) PostCompose(ctx echo.Context) error {
 		}
 
 		imageRequests[i].manifest = manifest
+		imageRequests[i].arch = arch.Name()
 		imageRequests[i].filename = imageType.Filename()
 	}
 
@@ -170,7 +172,7 @@ func (h *apiHandlers) PostCompose(ctx echo.Context) error {
 		return echo.NewHTTPError(http.StatusBadRequest, fmt.Sprintf("Could not initialize build with koji: %v", err))
 	}
 
-	id, err := h.server.workers.Enqueue(ir.manifest, []*target.Target{
+	id, err := h.server.workers.Enqueue(ir.arch, ir.manifest, []*target.Target{
 		target.NewKojiTarget(&target.KojiTargetOptions{
 			BuildID:         uint64(buildInfo.BuildID),
 			TaskID:          uint64(request.Koji.TaskId),
diff --git a/internal/weldr/api.go b/internal/weldr/api.go
index a4be5ca3e..e1edcb606 100644
--- a/internal/weldr/api.go
+++ b/internal/weldr/api.go
@@ -1825,7 +1825,7 @@ func (api *API) composeHandler(writer http.ResponseWriter, request *http.Request
 	} else {
 		var jobId uuid.UUID
 
-		jobId, err = api.workers.Enqueue(manifest, targets)
+		jobId, err = api.workers.Enqueue(api.arch.Name(), manifest, targets)
 		if err == nil {
 			err = api.store.PushCompose(composeID, manifest, imageType, bp, size, targets, jobId)
 		}
diff --git a/internal/worker/api/api.gen.go b/internal/worker/api/api.gen.go
index b2bebfcd0..3d925e49e 100644
--- a/internal/worker/api/api.gen.go
+++ b/internal/worker/api/api.gen.go
@@ -16,7 +16,10 @@ type Error struct {
 }
 
 // RequestJobJSONBody defines parameters for RequestJob.
-type RequestJobJSONBody map[string]interface{}
+type RequestJobJSONBody struct {
+	Arch  string   `json:"arch"`
+	Types []string `json:"types"`
+}
 
 // UpdateJobJSONBody defines parameters for UpdateJob.
 type UpdateJobJSONBody struct {
diff --git a/internal/worker/api/openapi.yml b/internal/worker/api/openapi.yml
index e1b132c1f..146f83f93 100644
--- a/internal/worker/api/openapi.yml
+++ b/internal/worker/api/openapi.yml
@@ -49,19 +49,20 @@ paths:
                 type: object
                 additionalProperties: false
                 properties:
-                  manifest: {}
-                  targets:
-                    type: array
-                    items: {}
+                  id:
+                    type: string
+                    format: uuid
                   location:
                     type: string
                   artifact_location:
                     type: string
-                  id:
+                  type:
                     type: string
-                    format: uuid
+                    enum:
+                      - osbuild
+                  args: {}
                 required:
-                  - manifest
+                  - type
                   - location
                   - id
         4XX:
@@ -82,6 +83,18 @@ paths:
             schema:
               type: object
               additionalProperties: false
+              properties:
+                types:
+                  type: array
+                  items:
+                    type: string
+                    enum:
+                      - osbuild
+                arch:
+                  type: string
+              required:
+                - types
+                - arch
         description: ''
       description: Requests a job. This operation blocks until a job is available.
     parameters: []
diff --git a/internal/worker/client.go b/internal/worker/client.go
index 8b6db3aed..619862bd7 100644
--- a/internal/worker/client.go
+++ b/internal/worker/client.go
@@ -5,6 +5,7 @@ import (
 	"context"
 	"crypto/tls"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"net"
@@ -35,10 +36,10 @@ type Job interface {
 type job struct {
 	requester        *http.Client
 	id               uuid.UUID
-	manifest         distro.Manifest
-	targets          []*target.Target
 	location         string
 	artifactLocation string
+	jobType          string
+	args             json.RawMessage
 }
 
 func NewClient(baseURL string, conf *tls.Config) (*Client, error) {
@@ -81,7 +82,10 @@ func (c *Client) RequestJob() (Job, error) {
 	}
 
 	var buf bytes.Buffer
-	err = json.NewEncoder(&buf).Encode(api.RequestJobJSONRequestBody{})
+	err = json.NewEncoder(&buf).Encode(api.RequestJobJSONRequestBody{
+		Types: []string{"osbuild"},
+		Arch:  common.CurrentArch(),
+	})
 	if err != nil {
 		panic(err)
 	}
@@ -115,8 +119,8 @@ func (c *Client) RequestJob() (Job, error) {
 	return &job{
 		requester:        c.requester,
 		id:               jr.Id,
-		manifest:         jr.Manifest,
-		targets:          jr.Targets,
+		jobType:          jr.Type,
+		args:             jr.Args,
 		location:         location.String(),
 		artifactLocation: artifactLocation.String(),
 	}, nil
@@ -127,7 +131,17 @@ func (j *job) Id() uuid.UUID {
 }
 
 func (j *job) OSBuildArgs() (distro.Manifest, []*target.Target, error) {
-	return j.manifest, j.targets, nil
+	if j.jobType != "osbuild" {
+		return nil, nil, errors.New("not an osbuild job")
+	}
+
+	var args OSBuildJob
+	err := json.Unmarshal(j.args, &args)
+	if err != nil {
+		return nil, nil, fmt.Errorf("error parsing osbuild job arguments: %v", err)
+	}
+
+	return args.Manifest, args.Targets, nil
 }
 
 func (j *job) Update(status common.ImageBuildState, result *osbuild.Result) error {
diff --git a/internal/worker/json.go b/internal/worker/json.go
index 60f95fc5c..977adc6a4 100644
--- a/internal/worker/json.go
+++ b/internal/worker/json.go
@@ -1,6 +1,8 @@
 package worker
 
 import (
+	"encoding/json"
+
 	"github.com/google/uuid"
 	"github.com/osbuild/osbuild-composer/internal/common"
 	"github.com/osbuild/osbuild-composer/internal/distro"
@@ -30,11 +32,11 @@ type statusResponse struct {
 }
 
 type requestJobResponse struct {
-	Id               uuid.UUID        `json:"id"`
-	Manifest         distro.Manifest  `json:"manifest"`
-	Targets          []*target.Target `json:"targets,omitempty"`
-	Location         string           `json:"location"`
-	ArtifactLocation string           `json:"artifact_location"`
+	Id               uuid.UUID       `json:"id"`
+	Location         string          `json:"location"`
+	ArtifactLocation string          `json:"artifact_location"`
+	Type             string          `json:"type"`
+	Args             json.RawMessage `json:"args,omitempty"`
 }
 
 type getJobResponse struct {
diff --git a/internal/worker/server.go b/internal/worker/server.go
index 98b5012d0..4e4a596d9 100644
--- a/internal/worker/server.go
+++ b/internal/worker/server.go
@@ -88,13 +88,13 @@ func (s *Server) ServeHTTP(writer http.ResponseWriter, request *http.Request) {
 	s.server.Handler.ServeHTTP(writer, request)
 }
 
-func (s *Server) Enqueue(manifest distro.Manifest, targets []*target.Target) (uuid.UUID, error) {
+func (s *Server) Enqueue(arch string, manifest distro.Manifest, targets []*target.Target) (uuid.UUID, error) {
 	job := OSBuildJob{
 		Manifest: manifest,
 		Targets:  targets,
 	}
 
-	return s.jobs.Enqueue("osbuild", job, nil)
+	return s.jobs.Enqueue("osbuild:"+arch, job, nil)
 }
 
 func (s *Server) JobStatus(id uuid.UUID) (*JobStatus, error) {
@@ -172,11 +172,14 @@ func (s *Server) DeleteArtifacts(id uuid.UUID) error {
 	return os.RemoveAll(path.Join(s.artifactsDir, id.String()))
 }
 
-func (s *Server) RequestJob(ctx context.Context) (uuid.UUID, uuid.UUID, *OSBuildJob, error) {
+func (s *Server) RequestOSBuildJob(ctx context.Context, arch string) (uuid.UUID, uuid.UUID, *OSBuildJob, error) {
 	token := uuid.New()
 
+	// wait on "osbuild" jobs for backwards compatiblity
+	jobTypes := []string{"osbuild", "osbuild:" + arch}
+
 	var args OSBuildJob
-	jobId, err := s.jobs.Dequeue(ctx, []string{"osbuild"}, &args)
+	jobId, err := s.jobs.Dequeue(ctx, jobTypes, &args)
 	if err != nil {
 		return uuid.Nil, uuid.Nil, nil, err
 	}
@@ -252,23 +255,32 @@ func (h *apiHandlers) GetStatus(ctx echo.Context) error {
 }
 
 func (h *apiHandlers) RequestJob(ctx echo.Context) error {
-	var body struct{}
+	var body api.RequestJobJSONRequestBody
 	err := ctx.Bind(&body)
 	if err != nil {
 		return err
 	}
 
-	token, jobId, jobArgs, err := h.server.RequestJob(ctx.Request().Context())
+	if len(body.Types) != 1 || body.Types[0] != "osbuild" {
+		return echo.NewHTTPError(http.StatusBadRequest, "invalid job types")
+	}
+
+	token, jobId, jobArgs, err := h.server.RequestOSBuildJob(ctx.Request().Context(), body.Arch)
+	if err != nil {
+		return err
+	}
+
+	serializedArgs, err := json.Marshal(jobArgs)
 	if err != nil {
 		return err
 	}
 
 	return ctx.JSON(http.StatusCreated, requestJobResponse{
 		Id:               jobId,
-		Manifest:         jobArgs.Manifest,
-		Targets:          jobArgs.Targets,
 		Location:         fmt.Sprintf("/jobs/%v", token),
 		ArtifactLocation: fmt.Sprintf("/jobs/%v/artifacts/", token),
+		Type:             "osbuild",
+		Args:             serializedArgs,
 	})
 }
 


Test patch:
diff --git a/internal/worker/server_test.go b/internal/worker/server_test.go
index ee3200613..e91af891c 100644
--- a/internal/worker/server_test.go
+++ b/internal/worker/server_test.go
@@ -64,11 +64,11 @@ func TestCreate(t *testing.T) {
 	}
 	server := worker.NewServer(nil, testjobqueue.New(), "")
 
-	_, err = server.Enqueue(manifest, nil)
+	_, err = server.Enqueue(arch.Name(), manifest, nil)
 	require.NoError(t, err)
 
-	test.TestRoute(t, server, false, "POST", "/jobs", `{}`, http.StatusCreated,
-		`{"manifest":{"sources":{},"pipeline":{}}}`, "id", "location", "artifact_location")
+	test.TestRoute(t, server, false, "POST", "/jobs", `{"types":["osbuild"],"arch":"x86_64"}`, http.StatusCreated,
+		`{"type":"osbuild","args":{"manifest":{"pipeline":{},"sources":{}}}}`, "id", "location", "artifact_location")
 }
 
 func TestCancel(t *testing.T) {
@@ -87,10 +87,10 @@ func TestCancel(t *testing.T) {
 	}
 	server := worker.NewServer(nil, testjobqueue.New(), "")
 
-	jobId, err := server.Enqueue(manifest, nil)
+	jobId, err := server.Enqueue(arch.Name(), manifest, nil)
 	require.NoError(t, err)
 
-	token, j, _, err := server.RequestJob(context.Background())
+	token, j, _, err := server.RequestOSBuildJob(context.Background(), arch.Name())
 	require.NoError(t, err)
 	require.Equal(t, jobId, j)