mesh-controller/cmd/mesh-builder/main.go

// 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
	}

	// Always through the exchange, whether or not somebody is waiting.
	//
	// **Never the default exchange.** Permission there is granted per exchange rather than per
	// queue, so a builder allowed to use it could publish into any node's queue — the privilege a
	// build machine most obviously should not have. An asker binds its own reply queue to this
	// key and filters by correlation; a control plane that records builds is bound to it too, so
	// a result nobody asked for is still kept rather than reported into the void.
	publishCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
	defer cancel()
	if err := channel.PublishWithContext(publishCtx, link.Exchange, link.KeyBuilt, 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
}