worker: upload local target image using jobqueue api

Prior this commit local target copied the image from a worker to a composer
using cp(1) command. This prevented the local target to work on remote
workers.

This commit switches the local target implementation to using the jobqueue
API introduced in the previous commit. I had some concerns about speed
of this solution (imho nothing can beat pure cp(1) implementation) but
ad hoc sanity tests showed the copying of the image using the jobqueue API
when running the worker on the same machine as the composer is still
more or less instant.
This commit is contained in:
Ondřej Budai 2020-02-10 15:22:37 +01:00 committed by Tom Gundersen
parent 98aac91083
commit 6902f730cb
3 changed files with 26 additions and 5 deletions

View file

@ -6,6 +6,7 @@ import (
"encoding/json" "encoding/json"
"errors" "errors"
"fmt" "fmt"
"io"
"log" "log"
"net" "net"
"net/http" "net/http"
@ -77,6 +78,14 @@ func (c *ComposerClient) UpdateJob(job *jobqueue.Job, status common.ImageBuildSt
return nil return nil
} }
func (c *ComposerClient) UploadImage(job *jobqueue.Job, reader io.Reader) error {
// content type doesn't really matter
url := fmt.Sprintf("http://localhost/job-queue/v1/jobs/%s/builds/%d/image", job.ID.String(), job.ImageBuildID)
_, err := c.client.Post(url, "application/octet-stream", reader)
return err
}
func handleJob(client *ComposerClient) error { func handleJob(client *ComposerClient) error {
fmt.Println("Waiting for a new job...") fmt.Println("Waiting for a new job...")
job, err := client.AddJob() job, err := client.AddJob()
@ -90,7 +99,7 @@ func handleJob(client *ComposerClient) error {
} }
fmt.Printf("Running job %s\n", job.ID.String()) fmt.Printf("Running job %s\n", job.ID.String())
result, err := job.Run() result, err := job.Run(client)
if err != nil { if err != nil {
log.Printf(" Job failed: %v", err) log.Printf(" Job failed: %v", err)
return client.UpdateJob(job, common.IBFailed, result) return client.UpdateJob(job, common.IBFailed, result)

View file

@ -44,7 +44,7 @@ func (e *TargetsError) Error() string {
return errString return errString
} }
func (job *Job) Run() (*common.ComposeResult, error) { func (job *Job) Run(uploader LocalTargetUploader) (*common.ComposeResult, error) {
distros := distro.NewRegistry([]string{"/etc/osbuild-composer", "/usr/share/osbuild-composer"}) distros := distro.NewRegistry([]string{"/etc/osbuild-composer", "/usr/share/osbuild-composer"})
d := distros.GetDistro(job.Distro) d := distros.GetDistro(job.Distro)
if d == nil { if d == nil {
@ -120,14 +120,19 @@ func (job *Job) Run() (*common.ComposeResult, error) {
for _, t := range job.Targets { for _, t := range job.Targets {
switch options := t.Options.(type) { switch options := t.Options.(type) {
case *target.LocalTargetOptions: case *target.LocalTargetOptions:
filename, _, err := d.FilenameFromType(job.OutputType)
err = runCommand("cp", "-a", "-L", tmpStore+"/refs/"+result.OutputID+"/.", options.Location)
if err != nil { if err != nil {
r = append(r, err) r = append(r, err)
continue continue
} }
err = runCommand("chown", "-R", "_osbuild-composer:_osbuild-composer", options.Location) f, err := os.Open(tmpStore + "/refs/" + result.OutputID + "/" + filename)
if err != nil {
r = append(r, err)
continue
}
err = uploader.UploadImage(job, f)
if err != nil { if err != nil {
r = append(r, err) r = append(r, err)
continue continue

View file

@ -0,0 +1,7 @@
package jobqueue
import "io"
type LocalTargetUploader interface {
UploadImage(job *Job, reader io.Reader) error
}