Repository: osbuild/osbuild-composer
Pull request #673 (2020-05-26 20:19:16)

worker: add JobArtifact() and DeleteJobArtifacts()
This allows removing the `artifactsDir` from `weldr.API`. It makes more
sense to deal with that directory in one place only.

This is a follow-up to @teg's [comment](https://github.com/osbuild/osbuild-composer/pull/666#discussion_r430248077) on the job artifact PR.


Patch:
diff --git a/cmd/osbuild-composer/main.go b/cmd/osbuild-composer/main.go
index 96022ffd5..f6dcde882 100644
--- a/cmd/osbuild-composer/main.go
+++ b/cmd/osbuild-composer/main.go
@@ -137,7 +137,7 @@ func main() {
 	compatOutputDir := path.Join(stateDir, "outputs")
 
 	workers := worker.NewServer(logger, jobs, artifactsDir)
-	weldrAPI := weldr.New(rpm, arch, distribution, repoMap[common.CurrentArch()], logger, store, workers, artifactsDir, compatOutputDir)
+	weldrAPI := weldr.New(rpm, arch, distribution, repoMap[common.CurrentArch()], logger, store, workers, compatOutputDir)
 
 	go func() {
 		err := workers.Serve(jobListener)
diff --git a/distribution/osbuild-remote-worker@.service b/distribution/osbuild-remote-worker@.service
index 2ffd084fc..1a2eba0c9 100644
--- a/distribution/osbuild-remote-worker@.service
+++ b/distribution/osbuild-remote-worker@.service
@@ -6,7 +6,6 @@ After=multi-user.target
 Type=simple
 PrivateTmp=true
 ExecStart=/usr/libexec/osbuild-composer/osbuild-worker %i
-CacheDirectory=osbuild-composer
 Restart=on-failure
 RestartSec=10s
 CPUSchedulingPolicy=batch
diff --git a/internal/weldr/api.go b/internal/weldr/api.go
index 475e54847..f41f37cc0 100644
--- a/internal/weldr/api.go
+++ b/internal/weldr/api.go
@@ -26,6 +26,7 @@ import (
 	"github.com/osbuild/osbuild-composer/internal/blueprint"
 	"github.com/osbuild/osbuild-composer/internal/common"
 	"github.com/osbuild/osbuild-composer/internal/distro"
+	"github.com/osbuild/osbuild-composer/internal/jobqueue"
 	"github.com/osbuild/osbuild-composer/internal/rpmmd"
 	"github.com/osbuild/osbuild-composer/internal/store"
 	"github.com/osbuild/osbuild-composer/internal/target"
@@ -44,13 +45,12 @@ type API struct {
 	logger *log.Logger
 	router *httprouter.Router
 
-	artifactsDir    string
 	compatOutputDir string
 }
 
 var ValidBlueprintName = regexp.MustCompile(`^[a-zA-Z0-9._-]+$`)
 
-func New(rpmmd rpmmd.RPMMD, arch distro.Arch, distro distro.Distro, repos []rpmmd.RepoConfig, logger *log.Logger, store *store.Store, workers *worker.Server, artifactsDir, compatOutputDir string) *API {
+func New(rpmmd rpmmd.RPMMD, arch distro.Arch, distro distro.Distro, repos []rpmmd.RepoConfig, logger *log.Logger, store *store.Store, workers *worker.Server, compatOutputDir string) *API {
 	api := &API{
 		store:           store,
 		workers:         workers,
@@ -59,7 +59,6 @@ func New(rpmmd rpmmd.RPMMD, arch distro.Arch, distro distro.Distro, repos []rpmm
 		distro:          distro,
 		repos:           repos,
 		logger:          logger,
-		artifactsDir:    artifactsDir,
 		compatOutputDir: compatOutputDir,
 	}
 
@@ -198,30 +197,34 @@ func (api *API) getComposeStatus(compose store.Compose) *composeStatus {
 	}
 }
 
-// Opens the image file for `compose`. This looks under `{artifacts}/{jobId}`
-// first, and then under `{outputs}/{composeId}/{imageBuildId}` for backwards
-// compatibility.
+// Opens the image file for `compose`. This asks the worker server for the
+// artifact first, and then falls back to looking in
+// `{outputs}/{composeId}/{imageBuildId}` for backwards compatibility.
 func (api *API) openImageFile(composeId uuid.UUID, compose store.Compose) (io.Reader, int64, error) {
-	p := path.Join(api.artifactsDir, compose.ImageBuild.JobID.String(), compose.ImageBuild.ImageType.Filename())
+	name := compose.ImageBuild.ImageType.Filename()
 
-	f, err := os.Open(p)
+	reader, size, err := api.workers.JobArtifact(compose.ImageBuild.JobID, name)
 	if err != nil {
-		if api.compatOutputDir == "" || !os.IsNotExist(err) {
+		if api.compatOutputDir == "" || err != jobqueue.ErrNotExist {
 			return nil, 0, err
 		}
-		p = path.Join(api.compatOutputDir, composeId.String(), "0")
-		f, err = os.Open(p)
+
+		p := path.Join(api.compatOutputDir, composeId.String(), "0", name)
+		f, err := os.Open(p)
 		if err != nil {
 			return nil, 0, err
 		}
-	}
 
-	info, err := f.Stat()
-	if err != nil {
-		return nil, 0, err
+		info, err := f.Stat()
+		if err != nil {
+			return nil, 0, err
+		}
+
+		reader = f
+		size = info.Size()
 	}
 
-	return f, info.Size(), nil
+	return reader, size, nil
 }
 
 func verifyRequestVersion(writer http.ResponseWriter, params httprouter.Params, minVersion uint) bool {
@@ -1754,11 +1757,12 @@ func (api *API) composeDeleteHandler(writer http.ResponseWriter, request *http.R
 			continue
 		}
 
-		// Delete artifacts from jobs and the compat output dir. Ignore
-		// errors, because there's no point of reporting them to the
-		// client after the compose itself has already been deleted.
-		_ = os.RemoveAll(path.Join(api.artifactsDir, compose.ImageBuild.JobID.String()))
-		if api.compatOutputDir != "" {
+		// Delete artifacts from the worker server or — if that doesn't
+		// have this job — the compat output dir. Ignore errors,
+		// because there's no point of reporting them to the client
+		// after the compose itself has already been deleted.
+		err = api.workers.DeleteArtifacts(compose.ImageBuild.JobID)
+		if err == jobqueue.ErrNotExist && api.compatOutputDir != "" {
 			_ = os.RemoveAll(path.Join(api.compatOutputDir, id.String()))
 		}
 
diff --git a/internal/worker/json.go b/internal/worker/json.go
index 7fdc8f8f9..bcf323f95 100644
--- a/internal/worker/json.go
+++ b/internal/worker/json.go
@@ -25,6 +25,10 @@ type OSBuildJobResult struct {
 // JSON-serializable types for the HTTP API
 //
 
+type statusResponse struct {
+	Status string `json:"status"`
+}
+
 type errorResponse struct {
 	Message string `json:"message"`
 }
diff --git a/internal/worker/server.go b/internal/worker/server.go
index 1c6d5bf49..9e5f1e8d2 100644
--- a/internal/worker/server.go
+++ b/internal/worker/server.go
@@ -49,6 +49,10 @@ func NewServer(logger *log.Logger, jobs jobqueue.JobQueue, artifactsDir string)
 	s.router.MethodNotAllowed = http.HandlerFunc(methodNotAllowedHandler)
 	s.router.NotFound = http.HandlerFunc(notFoundHandler)
 
+	// Add a basic status handler for checking if osbuild-composer is alive.
+	s.router.GET("/status", s.statusHandler)
+
+	// Add handlers for managing jobs.
 	s.router.POST("/job-queue/v1/jobs", s.addJobHandler)
 	s.router.PATCH("/job-queue/v1/jobs/:job_id", s.updateJobHandler)
 	s.router.POST("/job-queue/v1/jobs/:job_id/artifacts/:name", s.addJobImageHandler)
@@ -115,6 +119,46 @@ func (s *Server) JobStatus(id uuid.UUID) (*JobStatus, error) {
 	}, nil
 }
 
+// Provides access to artifacts of a job. Returns an io.Reader for the artifact
+// and the artifact's size.
+func (s *Server) JobArtifact(id uuid.UUID, name string) (io.Reader, int64, error) {
+	status, err := s.JobStatus(id)
+	if err != nil {
+		return nil, 0, err
+	}
+
+	if status.Finished.IsZero() {
+		return nil, 0, fmt.Errorf("Cannot access artifacts before job is finished: %s", id)
+	}
+
+	p := path.Join(s.artifactsDir, id.String(), name)
+	f, err := os.Open(p)
+	if err != nil {
+		return nil, 0, fmt.Errorf("Error accessing artifact %s for job %s: %v", name, id, err)
+	}
+
+	info, err := f.Stat()
+	if err != nil {
+		return nil, 0, fmt.Errorf("Error getting size of artifact %s for job %s: %v", name, id, err)
+	}
+
+	return f, info.Size(), nil
+}
+
+// Deletes all artifacts for job `id`.
+func (s *Server) DeleteArtifacts(id uuid.UUID) error {
+	status, err := s.JobStatus(id)
+	if err != nil {
+		return err
+	}
+
+	if status.Finished.IsZero() {
+		return fmt.Errorf("Cannot delete artifacts before job is finished: %s", id)
+	}
+
+	return os.RemoveAll(path.Join(s.artifactsDir, id.String()))
+}
+
 // jsonErrorf() is similar to http.Error(), but returns the message in a json
 // object with a "message" field.
 func jsonErrorf(writer http.ResponseWriter, code int, message string, args ...interface{}) {
@@ -134,6 +178,15 @@ func notFoundHandler(writer http.ResponseWriter, request *http.Request) {
 	jsonErrorf(writer, http.StatusNotFound, "not found")
 }
 
+func (s *Server) statusHandler(writer http.ResponseWriter, request *http.Request, _ httprouter.Params) {
+	writer.WriteHeader(http.StatusOK)
+
+	// Send back a status message.
+	_ = json.NewEncoder(writer).Encode(&statusResponse{
+		Status: "OK",
+	})
+}
+
 func (s *Server) addJobHandler(writer http.ResponseWriter, request *http.Request, _ httprouter.Params) {
 	contentType := request.Header["Content-Type"]
 	if len(contentType) != 1 || contentType[0] != "application/json" {


Test patch:
diff --git a/internal/client/unit_test.go b/internal/client/unit_test.go
index 75dc24966..255b2704e 100644
--- a/internal/client/unit_test.go
+++ b/internal/client/unit_test.go
@@ -11,7 +11,6 @@ import (
 	"net"
 	"net/http"
 	"os"
-	"path"
 	"testing"
 
 	"github.com/osbuild/osbuild-composer/internal/distro/fedoratest"
@@ -37,12 +36,6 @@ func executeTests(m *testing.M) int {
 		panic(err)
 	}
 
-	artifactsDir := path.Join(tmpdir, "artifacts")
-	err = os.Mkdir(artifactsDir, 0755)
-	if err != nil {
-		panic(err)
-	}
-
 	// Create a mock API server listening on the temporary socket
 	fixture := rpmmd_mock.BaseFixture()
 	rpm := rpmmd_mock.NewRPMMDMock(fixture)
@@ -53,7 +46,7 @@ func executeTests(m *testing.M) int {
 	}
 	repos := []rpmmd.RepoConfig{{Id: "test-system-repo", BaseURL: "http://example.com/test/os/test_arch"}}
 	logger := log.New(os.Stdout, "", 0)
-	api := weldr.New(rpm, arch, distro, repos, logger, fixture.Store, fixture.Workers, artifactsDir, "")
+	api := weldr.New(rpm, arch, distro, repos, logger, fixture.Store, fixture.Workers, "")
 	server := http.Server{Handler: api}
 	defer server.Close()
 
diff --git a/internal/weldr/api_test.go b/internal/weldr/api_test.go
index c5ac0d474..e99f24df4 100644
--- a/internal/weldr/api_test.go
+++ b/internal/weldr/api_test.go
@@ -4,7 +4,6 @@ import (
 	"archive/tar"
 	"bytes"
 	"io"
-	"io/ioutil"
 	"math/rand"
 	"net/http"
 	"net/http/httptest"
@@ -38,13 +37,7 @@ func createWeldrAPI(fixtureGenerator rpmmd_mock.FixtureGenerator) (*API, *store.
 		panic(err)
 	}
 
-	artifactsDir, err := ioutil.TempDir("", "client_test-")
-	if err != nil {
-		panic(err)
-	}
-	defer os.RemoveAll(artifactsDir)
-
-	return New(rpm, arch, d, repos, nil, fixture.Store, fixture.Workers, artifactsDir, ""), fixture.Store
+	return New(rpm, arch, d, repos, nil, fixture.Store, fixture.Workers, ""), fixture.Store
 }
 
 func TestBasic(t *testing.T) {
diff --git a/internal/worker/server_test.go b/internal/worker/server_test.go
index 1ed3d254e..41dae2494 100644
--- a/internal/worker/server_test.go
+++ b/internal/worker/server_test.go
@@ -14,6 +14,12 @@ import (
 	"github.com/osbuild/osbuild-composer/internal/worker"
 )
 
+// Ensure that the status request returns OK.
+func TestStatus(t *testing.T) {
+	server := worker.NewServer(nil, testjobqueue.New(), "")
+	test.TestRoute(t, server, false, "GET", "/status", ``, http.StatusOK, `{"status":"OK"}`, "message")
+}
+
 func TestErrors(t *testing.T) {
 	var cases = []struct {
 		Method         string