Repository: osbuild/osbuild-composer
Pull request #7 (2019-09-26 21:26:41)

job-queue: pass a well-formed job object to the worker
For now we will hardcode the org.osbuild.local target, so we might
as well fix this up front.

We do not yet support any target types, but for testing purposes we
claim to support 'tar', and we pass a noop tar pipeline to the worker.

This makes introspecting the job-queu api using curl a bit more
pleasant.

Signed-off-by: Tom Gundersen <teg@jklm.no>


Patch:
diff --git a/internal/job/job.go b/internal/job/job.go
index 8fd68256d..62dc51093 100644
--- a/internal/job/job.go
+++ b/internal/job/job.go
@@ -8,5 +8,5 @@ import (
 type Job struct {
 	ComposeID string
 	Pipeline  pipeline.Pipeline
-	Target    target.Target
+	Targets   []target.Target
 }
diff --git a/internal/job/store.go b/internal/job/store.go
index 0f21f794f..ac132f15a 100644
--- a/internal/job/store.go
+++ b/internal/job/store.go
@@ -38,7 +38,7 @@ func (s *Store) UpdateJob(id string, job Job) bool {
 	req, _ := s.jobs[id]
 	req.ComposeID = job.ComposeID
 	req.Pipeline = job.Pipeline
-	req.Target = job.Target
+	req.Targets = job.Targets
 
 	return true
 }
diff --git a/internal/jobqueue/api.go b/internal/jobqueue/api.go
index 8e6c66a35..026f0a66a 100644
--- a/internal/jobqueue/api.go
+++ b/internal/jobqueue/api.go
@@ -34,7 +34,7 @@ func New(logger *log.Logger, jobs <-chan job.Job) *API {
 	api.router.NotFound = http.HandlerFunc(notFoundHandler)
 
 	api.router.POST("/job-queue/v1/jobs", api.addJobHandler)
-	api.router.PATCH("/job-queue/v1/jobs/:job-id", api.updateJobHandler)
+	api.router.PATCH("/job-queue/v1/jobs/:id", api.updateJobHandler)
 
 	return api
 }
@@ -77,11 +77,11 @@ func statusResponseError(writer http.ResponseWriter, code int, errors ...string)
 
 func (api *API) addJobHandler(writer http.ResponseWriter, request *http.Request, _ httprouter.Params) {
 	type requestBody struct {
-		JobID string `json:"job-id"`
+		ID string `json:"id"`
 	}
 	type replyBody struct {
 		Pipeline pipeline.Pipeline `json:"pipeline"`
-		Target   target.Target     `json:"target"`
+		Targets  []target.Target   `json:"targets"`
 	}
 
 	contentType := request.Header["Content-Type"]
@@ -93,22 +93,22 @@ func (api *API) addJobHandler(writer http.ResponseWriter, request *http.Request,
 	var body requestBody
 	err := json.NewDecoder(request.Body).Decode(&body)
 	if err != nil {
-		statusResponseError(writer, http.StatusBadRequest, "invalid job-id: "+err.Error())
+		statusResponseError(writer, http.StatusBadRequest, "invalid id: "+err.Error())
 		return
 	}
 
-	id := body.JobID
-	var req job.Job
+	id := body.ID
+	var jobSlot job.Job
 
-	if !api.jobStore.AddJob(id, req) {
+	if !api.jobStore.AddJob(id, jobSlot) {
 		statusResponseError(writer, http.StatusBadRequest)
 		return
 	}
 
-	req = <-api.pendingJobs
-	api.jobStore.UpdateJob(id, req)
+	nextJob := <-api.pendingJobs
+	api.jobStore.UpdateJob(id, nextJob)
 
-	json.NewEncoder(writer).Encode(replyBody{req.Pipeline, req.Target})
+	json.NewEncoder(writer).Encode(replyBody{nextJob.Pipeline, nextJob.Targets})
 
 }
 
@@ -127,15 +127,12 @@ func (api *API) updateJobHandler(writer http.ResponseWriter, request *http.Reque
 	err := json.NewDecoder(request.Body).Decode(&body)
 	if err != nil {
 		statusResponseError(writer, http.StatusBadRequest, "invalid status: "+err.Error())
-		return
 	} else if body.Status == "running" {
 		statusResponseOK(writer)
-		return
-	} else if body.Status != "finished" {
+	} else if body.Status == "finished" {
+		api.jobStore.DeleteJob(params.ByName("id"))
+		statusResponseOK(writer)
+	} else {
 		statusResponseError(writer, http.StatusBadRequest, "invalid status: "+body.Status)
-		return
 	}
-
-	api.jobStore.DeleteJob(params.ByName("job-id"))
-	statusResponseOK(writer)
 }
diff --git a/internal/pipeline/pipeline.go b/internal/pipeline/pipeline.go
index 17ff93ec7..85aedc51c 100644
--- a/internal/pipeline/pipeline.go
+++ b/internal/pipeline/pipeline.go
@@ -1,3 +1,18 @@
 package pipeline
 
-type Pipeline string
+type Pipeline struct {
+	Stages    []Stage   `json:"stages,omitempty"`
+	Assembler Assembler `json:"assembler"`
+}
+
+type Stage struct {
+}
+
+type Assembler struct {
+	Name    string              `json:"name"`
+	Options AssemblerTarOptions `json:"options"`
+}
+
+type AssemblerTarOptions struct {
+	Filename string `json:"filename"`
+}
diff --git a/internal/target/target.go b/internal/target/target.go
index 1481a1fdf..de28142f7 100644
--- a/internal/target/target.go
+++ b/internal/target/target.go
@@ -1,3 +1,10 @@
 package target
 
-type Target string
+type Target struct {
+	Name    string       `json:"name"`
+	Options LocalOptions `json:"options"`
+}
+
+type LocalOptions struct {
+	Location string `json:"location"`
+}
diff --git a/internal/weldr/api.go b/internal/weldr/api.go
index 66772f28e..620b8d9c4 100644
--- a/internal/weldr/api.go
+++ b/internal/weldr/api.go
@@ -12,6 +12,7 @@ import (
 
 	"osbuild-composer/internal/job"
 	"osbuild-composer/internal/rpmmd"
+	"osbuild-composer/internal/target"
 )
 
 type API struct {
@@ -618,11 +619,18 @@ func (api *API) composeHandler(writer http.ResponseWriter, httpRequest *http.Req
 	changed := false
 	found := api.store.getBlueprint(cr.BlueprintName, &bp, &changed) // TODO: what to do with changed?
 
+	uuid := "ffffffff-ffff–ffff-ffff-ffffffffffff" // TODO: generate
+
 	if found {
 		api.pendingJobs <- job.Job{
-			ComposeID: "TODO",
+			ComposeID: uuid,
 			Pipeline:  bp.translateToPipeline(cr.ComposeType),
-			Target:    `{"output-path":"/var/cache/osbuild-composer"}`,
+			Targets: []target.Target{{
+				Name: "org.osbuild.local",
+				Options: target.LocalOptions{
+					Location: "/var/lib/osbuild-composer/" + uuid,
+				}},
+			},
 		}
 	} else {
 		statusResponseError(writer, http.StatusBadRequest, "blueprint does not exist")
diff --git a/internal/weldr/store.go b/internal/weldr/store.go
index ba08a2f64..9e485c73b 100644
--- a/internal/weldr/store.go
+++ b/internal/weldr/store.go
@@ -2,7 +2,6 @@ package weldr
 
 import (
 	"encoding/json"
-	"fmt"
 	"log"
 	"osbuild-composer/internal/pipeline"
 	"sort"
@@ -158,5 +157,12 @@ func (s *store) deleteBlueprintFromWorkspace(name string) {
 }
 
 func (b *blueprint) translateToPipeline(outputFormat string) pipeline.Pipeline {
-	return pipeline.Pipeline(fmt.Sprintf("{\"pipeline\":\"%s\"}", "string"))
+	return pipeline.Pipeline{
+		Assembler: pipeline.Assembler{
+			Name: "org.osbuild.tar",
+			Options: pipeline.AssemblerTarOptions{
+				Filename: "image.tar",
+			},
+		},
+	}
 }


Test patch:
diff --git a/internal/jobqueue/api_test.go b/internal/jobqueue/api_test.go
index 8160fc0f4..36a008ccb 100644
--- a/internal/jobqueue/api_test.go
+++ b/internal/jobqueue/api_test.go
@@ -64,6 +64,7 @@ func testRoute(t *testing.T, api *jobqueue.API, method, path, body string, expec
 }
 
 func TestBasic(t *testing.T) {
+	expected_job := `{"pipeline":{"assembler":{"name":"org.osbuild.tar","options":{"filename":"image.tar"}}},"targets":[{"name":"org.osbuild.local","options":{"location":"/var/lib/osbuild-composer/ffffffff-ffff–ffff-ffff-ffffffffffff"}}]}`
 	var cases = []struct {
 		Method         string
 		Path           string
@@ -76,8 +77,8 @@ func TestBasic(t *testing.T) {
 		{"PATH", "/job-queue/v1/foo", ``, http.StatusNotFound, ``},
 		{"DELETE", "/job-queue/v1/foo", ``, http.StatusNotFound, ``},
 
-		{"POST", "/job-queue/v1/jobs", `{"job-id":"ffffffff-ffff–ffff-ffff-ffffffffffff"}`, http.StatusOK, `{"pipeline":"pipeline","target":"target"}`},
-		{"POST", "/job-queue/v1/jobs", `{"job-id":"ffffffff-ffff–ffff-ffff-ffffffffffff"}`, http.StatusBadRequest, ``},
+		{"POST", "/job-queue/v1/jobs", `{"id":"ffffffff-ffff–ffff-ffff-ffffffffffff"}`, http.StatusOK, expected_job},
+		{"POST", "/job-queue/v1/jobs", `{"id":"ffffffff-ffff–ffff-ffff-ffffffffffff"}`, http.StatusBadRequest, ``},
 		//{"PATCH", "/job-queue/v1/jobs/aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa", `{"status":"finished"}`, http.StatusBadRequest, ``},
 		{"PATCH", "/job-queue/v1/jobs/ffffffff-ffff-ffff-ffff-ffffffffffff", `{"status":"running"}`, http.StatusOK, ``},
 		{"PATCH", "/job-queue/v1/jobs/ffffffff-ffff-ffff-ffff-ffffffffffff", `{"status":"running"}`, http.StatusOK, ``},
@@ -90,10 +91,23 @@ func TestBasic(t *testing.T) {
 	api := jobqueue.New(nil, jobChannel)
 	for _, c := range cases {
 		jobChannel <- job.Job{
-			ComposeID: "ID",
-			Pipeline:  pipeline.Pipeline("pipeline"),
-			Target:    target.Target("target"),
+			ComposeID: "ffffffff-ffff–ffff-ffff-ffffffffffff",
+			Pipeline: pipeline.Pipeline{
+				Assembler: pipeline.Assembler{
+					Name: "org.osbuild.tar",
+					Options: pipeline.AssemblerTarOptions{
+						Filename: "image.tar",
+					},
+				},
+			},
+			Targets: []target.Target{{
+				Name: "org.osbuild.local",
+				Options: target.LocalOptions{
+					Location: "/var/lib/osbuild-composer/ffffffff-ffff–ffff-ffff-ffffffffffff",
+				}},
+			},
 		}
+
 		testRoute(t, api, c.Method, c.Path, c.Body, c.ExpectedStatus, c.ExpectedJSON)
 	}
 }
diff --git a/internal/weldr/api_test.go b/internal/weldr/api_test.go
index 3f3eac30a..43ee6c2cf 100644
--- a/internal/weldr/api_test.go
+++ b/internal/weldr/api_test.go
@@ -11,9 +11,7 @@ import (
 	"testing"
 
 	"osbuild-composer/internal/job"
-	"osbuild-composer/internal/pipeline"
 	"osbuild-composer/internal/rpmmd"
-	"osbuild-composer/internal/target"
 	"osbuild-composer/internal/weldr"
 )
 
@@ -157,12 +155,10 @@ func TestCompose(t *testing.T) {
 		http.StatusOK, `{"status":true}`)
 
 	job := <-jobChannel
-	expected_pipeline := pipeline.Pipeline(`{"pipeline":"string"}`)
-	expected_target := target.Target(`{"output-path":"/var/cache/osbuild-composer"}`)
-	if expected_target != job.Target {
-		t.Errorf("Expected this manifest: %s; got this: %s", expected_target, job.Target)
+	if job.Pipeline.Assembler.Name != "org.osbuild.tar" {
+		t.Errorf("Expected tar assembler, got: %s", job.Pipeline.Assembler.Name)
 	}
-	if expected_pipeline != job.Pipeline {
-		t.Errorf("Expected this manifest: %s; got this: %s", expected_pipeline, job.Pipeline)
+	if job.Targets[0].Name != "org.osbuild.local" {
+		t.Errorf("Expected local target, got: %s", job.Targets[0].Name)
 	}
 }