b5769add2c
The store is responsible for two things: user state and the compose queue. This is problematic, because the rcm API has slightly different semantics from weldr and only used the queue part of the store. Also, the store is simply too complex. This commit splits the queue part out, using the new jobqueue package in both the weldr and the rcm package. The queue is saved to a new directory `queue/`. The weldr package now also has access to a worker server to enqueue and list jobs. Its store continues to track composes, but the `QueueStatus` for each compose (and image build) is deprecated. The field in `ImageBuild` is kept for backwards compatibility for composes which finished before this change, but a lot of code dealing with it in package compose is dropped. store.PushCompose() is degraded to storing a new compose. It should probably be renamed in the future. store.PopJob() is removed. Job ids are now independent of compose ids. Because of that, the local target gains ComposeId and ImageBuildId fields, because a worker cannot infer those from a job anymore. This also necessitates a change in the worker API: the job routes are changed to expect that instead of a (compose id, image build id) pair. The route that accepts built images keeps that pair, because it reports the image back to weldr. worker.Server() now interacts with a job queue instead of the store. It gains public functions that allow enqueuing an osbuild job and getting its status, because only it knows about the specific argument and result types in the job queue (OSBuildJob and OSBuildJobResult). One oddity remains: it needs to report an uploaded image to weldr. Do this with a function that's passed in for now, so that the dependency to the store can be dropped completely. The rcm API drops its dependencies to package blueprint and store, because it too interacts only with the worker server now. Fixes #342
131 lines
2.8 KiB
Go
131 lines
2.8 KiB
Go
package worker
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"crypto/tls"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"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 {
|
|
client *http.Client
|
|
scheme string
|
|
hostname string
|
|
}
|
|
|
|
type Job struct {
|
|
Id uuid.UUID
|
|
Manifest *osbuild.Manifest
|
|
Targets []*target.Target
|
|
}
|
|
|
|
func NewClient(address string, conf *tls.Config) *Client {
|
|
client := &http.Client{
|
|
Transport: &http.Transport{
|
|
TLSClientConfig: conf,
|
|
},
|
|
}
|
|
|
|
var scheme string
|
|
// Use https if the TLS configuration is present, otherwise use http.
|
|
if conf != nil {
|
|
scheme = "https"
|
|
} else {
|
|
scheme = "http"
|
|
}
|
|
|
|
return &Client{client, scheme, address}
|
|
}
|
|
|
|
func NewClientUnix(path string) *Client {
|
|
client := &http.Client{
|
|
Transport: &http.Transport{
|
|
DialContext: func(context context.Context, network, addr string) (net.Conn, error) {
|
|
return net.Dial("unix", path)
|
|
},
|
|
},
|
|
}
|
|
|
|
return &Client{client, "http", "localhost"}
|
|
}
|
|
|
|
func (c *Client) AddJob() (*Job, error) {
|
|
var b bytes.Buffer
|
|
err := json.NewEncoder(&b).Encode(addJobRequest{})
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
response, err := c.client.Post(c.createURL("/job-queue/v1/jobs"), "application/json", &b)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer response.Body.Close()
|
|
|
|
if response.StatusCode != http.StatusCreated {
|
|
var er errorResponse
|
|
_ = json.NewDecoder(response.Body).Decode(&er)
|
|
return nil, fmt.Errorf("couldn't create job, got %d: %s", response.StatusCode, er.Message)
|
|
}
|
|
|
|
var jr addJobResponse
|
|
err = json.NewDecoder(response.Body).Decode(&jr)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return &Job{
|
|
jr.Id,
|
|
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(&updateJobRequest{status, result})
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
urlPath := fmt.Sprintf("/job-queue/v1/jobs/%s", job.Id)
|
|
url := c.createURL(urlPath)
|
|
req, err := http.NewRequest("PATCH", url, &b)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
req.Header.Set("Content-Type", "application/json")
|
|
response, err := c.client.Do(req)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer response.Body.Close()
|
|
|
|
if response.StatusCode != http.StatusOK {
|
|
return errors.New("error setting job status")
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (c *Client) UploadImage(composeId uuid.UUID, imageBuildId int, reader io.Reader) error {
|
|
// content type doesn't really matter
|
|
url := c.createURL(fmt.Sprintf("/job-queue/v1/jobs/%s/builds/%d/image", composeId, imageBuildId))
|
|
_, err := c.client.Post(url, "application/octet-stream", reader)
|
|
|
|
return err
|
|
}
|
|
|
|
func (c *Client) createURL(path string) string {
|
|
return c.scheme + "://" + c.hostname + path
|
|
}
|