The mesh builds: a machine takes the work, and the catalogue shows it
A build is work, not state. Everything else the control plane sends a node is a declaration — this is what you should be — reconciled forever. A build happens once and is finished. Putting it in a declaration would mean rebuilding on every reconcile, or a declaration carrying "and I already did this", which is state about an event rather than about a machine. So it travels on its own queue and the answer comes back correlated. One queue, so several build machines share the work and each request is done exactly once — which a per-machine routing key would not give. mesh-builder is the program a build machine runs. Not the control plane, which must not run commands on a machine; not the host, which would then need a container runtime and git everywhere to do something almost no machine will ever do. It holds its own broker credential and nothing else. Three properties that are decisions: - a request is acknowledged only once the answer is away, so a builder that dies mid-build leaves the work for another machine rather than losing it with nobody ever hearing why - one build at a time. Five at once against one runtime finishes all five slower than it would have finished the first, and the queue is what shares work between machines - a failure is a RESULT. A build that fails silently is indistinguishable from a builder that is not running, and those want different responses And `module list` is a catalogue: what exists, at which version, built from which commit or handed over by hand or shipped with the control plane, whether it is behind its source, and which machines run it. All of that was recorded from the first build and none of it was shown, so "is this current?" could only be answered by reading the database. Proven against a real broker, registry and store: the mesh asked, a builder consumed, built, published, answered; the manifest was recorded with its commit; the source moved and the catalogue said "behind"; rebuilding caught it up with a new digest because the content changed.
This commit is contained in:
@@ -0,0 +1,217 @@
|
||||
// mesh-builder — the thing a build machine runs.
|
||||
//
|
||||
// It takes work from the mesh, turns a repository into artifacts, publishes them, and says what
|
||||
// came out. It is **not** the control plane and it is **not** the host:
|
||||
//
|
||||
// - the control plane decides and never touches a machine. Building runs commands on one, and
|
||||
// what the control plane may send a machine is bounded by the declaration language
|
||||
// (novox/hq ADR 0005). "Run this build" is not in it, and widening the language so it could
|
||||
// be would make the control plane able to run anything anywhere.
|
||||
// - the host applies declarations and holds no opinion about what they contain. A host that
|
||||
// also built things would need a container runtime and git, on every machine, to do something
|
||||
// almost none of them will ever do.
|
||||
//
|
||||
// So it is a module: a program a machine runs because the mesh told it to, holding its own broker
|
||||
// credential and nothing else. Compromise of a build machine is compromise of a build machine.
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
"os/signal"
|
||||
"strings"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
amqp "github.com/rabbitmq/amqp091-go"
|
||||
|
||||
"github.com/novox/mesh-control/internal/builder"
|
||||
"github.com/novox/mesh-control/internal/link"
|
||||
)
|
||||
|
||||
// version is set at build time.
|
||||
var version = "development"
|
||||
|
||||
func main() {
|
||||
if err := run(); err != nil {
|
||||
fmt.Fprintf(os.Stderr, "mesh-builder: %v\n", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
}
|
||||
|
||||
const usage = `mesh-builder — builds modules for the mesh
|
||||
|
||||
It consumes build requests and answers with what it made. Nothing is listened on and nothing
|
||||
is dialled except the broker.
|
||||
|
||||
MESH_BROKER_AMQP where the broker is, with this builder's own credential
|
||||
MESH_REGISTRY host:port to publish artifacts to
|
||||
MESH_WORKSPACE where to clone and build (default: a temporary directory)
|
||||
`
|
||||
|
||||
func run() error {
|
||||
if len(os.Args) > 1 {
|
||||
switch os.Args[1] {
|
||||
case "version":
|
||||
fmt.Println(version)
|
||||
return nil
|
||||
default:
|
||||
fmt.Print(usage)
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
amqpURL := strings.TrimSpace(os.Getenv("MESH_BROKER_AMQP"))
|
||||
if amqpURL == "" {
|
||||
return fmt.Errorf("no MESH_BROKER_AMQP: a builder with no broker has nothing to build")
|
||||
}
|
||||
registry := strings.TrimSpace(os.Getenv("MESH_REGISTRY"))
|
||||
if registry == "" {
|
||||
return fmt.Errorf(
|
||||
"no MESH_REGISTRY: a built artifact nobody can fetch is not built")
|
||||
}
|
||||
workspace := os.Getenv("MESH_WORKSPACE")
|
||||
if workspace == "" {
|
||||
workspace = os.TempDir() + "/mesh-builder"
|
||||
}
|
||||
on, err := os.Hostname()
|
||||
if err != nil {
|
||||
on = "a build machine"
|
||||
}
|
||||
|
||||
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
|
||||
defer stop()
|
||||
|
||||
conn, err := amqp.Dial(amqpURL)
|
||||
if err != nil {
|
||||
// Not quoted back: the URL carries this builder's broker password.
|
||||
return fmt.Errorf("cannot reach the broker: %w", err)
|
||||
}
|
||||
defer conn.Close()
|
||||
channel, err := conn.Channel()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer channel.Close()
|
||||
|
||||
if _, err := channel.QueueDeclare(link.BuildQueue, true, false, false, false, nil); err != nil {
|
||||
return err
|
||||
}
|
||||
// One at a time. A build machine that took five requests at once would run five container
|
||||
// builds against one runtime and finish all of them slower than it would have finished the
|
||||
// first — and the queue is what shares work between machines, so nothing is lost by it.
|
||||
if err := channel.Qos(1, 0, false); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Not auto-acknowledged. A request acknowledged on arrival is a build that vanishes if this
|
||||
// process dies mid-way, with nobody waiting on it ever hearing why.
|
||||
requests, err := channel.ConsumeWithContext(ctx, link.BuildQueue, "mesh-builder",
|
||||
false, false, false, false, nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
fmt.Printf("building for the mesh, publishing to %s\n", registry)
|
||||
publisher := builder.Registry{Address: registry, Run: builder.Command}
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
fmt.Println("stopping")
|
||||
return nil
|
||||
case delivery, ok := <-requests:
|
||||
if !ok {
|
||||
return fmt.Errorf("the broker closed the connection")
|
||||
}
|
||||
answer(ctx, channel, publisher, on, workspace, delivery)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// answer does one build and says what happened, whichever way it went.
|
||||
func answer(ctx context.Context, channel *amqp.Channel, publisher builder.Publisher,
|
||||
on, workspace string, delivery amqp.Delivery) {
|
||||
|
||||
var request link.BuildRequest
|
||||
if err := json.Unmarshal(delivery.Body, &request); err != nil {
|
||||
// Unreadable. Acknowledged and dropped rather than requeued: a message this builder
|
||||
// cannot parse will not become parseable by being delivered again, and requeueing it
|
||||
// would put it in front of every real request for ever.
|
||||
fmt.Fprintf(os.Stderr, "a request could not be read and was dropped: %v\n", err)
|
||||
_ = delivery.Ack(false)
|
||||
return
|
||||
}
|
||||
|
||||
result := link.BuildResult{
|
||||
ID: request.ID, Repository: request.Repository, Ref: request.Ref, On: on,
|
||||
}
|
||||
fmt.Printf("building %s", request.Repository)
|
||||
if request.Ref != "" {
|
||||
fmt.Printf(" at %s", request.Ref)
|
||||
}
|
||||
fmt.Println()
|
||||
|
||||
built, err := builder.Build(ctx, builder.Command, publisher,
|
||||
request.Repository, request.Ref, workspace)
|
||||
if err != nil {
|
||||
// A failure is a result. A build that fails and says nothing is indistinguishable from a
|
||||
// builder that is not running, and those want completely different responses.
|
||||
result.Failed = err.Error()
|
||||
fmt.Fprintf(os.Stderr, " failed: %v\n", err)
|
||||
} else {
|
||||
manifest, marshalErr := json.Marshal(built.Manifest)
|
||||
if marshalErr != nil {
|
||||
result.Failed = marshalErr.Error()
|
||||
} else {
|
||||
result.Commit = built.Commit
|
||||
result.Manifest = manifest
|
||||
for _, made := range built.Built {
|
||||
result.Made = append(result.Made, link.MadeArtifact{
|
||||
Name: made.Name, Kind: made.Kind, Reference: made.Reference,
|
||||
})
|
||||
}
|
||||
fmt.Printf(" built %s from %s\n", built.Manifest.Module, short(built.Commit))
|
||||
}
|
||||
}
|
||||
|
||||
body, err := json.Marshal(result)
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "cannot report a build: %v\n", err)
|
||||
_ = delivery.Ack(false)
|
||||
return
|
||||
}
|
||||
|
||||
replyTo := delivery.ReplyTo
|
||||
if replyTo == "" {
|
||||
// Nobody is waiting. Still reported, to the exchange, so a control plane that records
|
||||
// builds hears about it — a build whose outcome exists nowhere is one nobody can audit.
|
||||
if err := channel.PublishWithContext(ctx, link.Exchange, link.KeyBuilt, false, false,
|
||||
amqp.Publishing{ContentType: "application/json", Body: body}); err != nil {
|
||||
fmt.Fprintf(os.Stderr, "cannot publish a build result: %v\n", err)
|
||||
}
|
||||
_ = delivery.Ack(false)
|
||||
return
|
||||
}
|
||||
publishCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
|
||||
defer cancel()
|
||||
if err := channel.PublishWithContext(publishCtx, "", replyTo, false, false, amqp.Publishing{
|
||||
ContentType: "application/json",
|
||||
CorrelationId: result.ID,
|
||||
Body: body,
|
||||
}); err != nil {
|
||||
fmt.Fprintf(os.Stderr, "cannot answer a build request: %v\n", err)
|
||||
}
|
||||
// Acknowledged only once the answer is away, so a builder that dies before answering leaves
|
||||
// the request for another machine rather than losing it.
|
||||
_ = delivery.Ack(false)
|
||||
}
|
||||
|
||||
func short(commit string) string {
|
||||
if len(commit) > 8 {
|
||||
return commit[:8]
|
||||
}
|
||||
return commit
|
||||
}
|
||||
Reference in New Issue
Block a user