293 lines
9.9 KiB
Go
293 lines
9.9 KiB
Go
// Command mesh-delivery-builder is delivery's builder (novox/hq ADR 0453 clause 4, ADR 0478, issue 1607): it builds every
|
|
// new-form fingerprint the indexer writes, dark, into its own build keys (builds.<module>.<fingerprint>), through the build
|
|
// modules' fetch and build tools. Nothing reads its keys to deliver until release R4e. It serves no request subject and
|
|
// holds no timer: every reaction is a watch or a bus event (ADR 0437 §8).
|
|
package main
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"crypto/tls"
|
|
"crypto/x509"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
|
|
"io"
|
|
"log"
|
|
"os"
|
|
"os/signal"
|
|
"strings"
|
|
"syscall"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
|
"git.novox.be/novox/mesh-sdk/go/process"
|
|
|
|
"mesh-delivery-builder/internal/build"
|
|
)
|
|
|
|
const (
|
|
// defaultStepBound is how long the builder waits for one step of a build module (a fetch downloads the Go modules, a
|
|
// build compiles).
|
|
defaultStepBound = 30 * time.Minute
|
|
// nodesPerFingerprint is how many build nodes build each fingerprint until release R4e (ADR 0478 clause 4).
|
|
nodesPerFingerprint = 2
|
|
)
|
|
|
|
func main() {
|
|
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
|
|
defer stop()
|
|
if err := run(ctx, os.Getenv, os.Stderr); err != nil {
|
|
fmt.Fprintln(os.Stderr, "mesh-delivery-builder:", err)
|
|
os.Exit(1)
|
|
}
|
|
}
|
|
|
|
// credential is the broker file the controller issues to the module: JSON of the bus address, the certificate
|
|
// fingerprint to pin, the principal and its password (as the indexer reads it; issue 1242 moves this into the SDK).
|
|
type credential struct {
|
|
URL string `json:"url"`
|
|
Fingerprint string `json:"fingerprint"`
|
|
Node string `json:"node"`
|
|
Module string `json:"module"`
|
|
User string `json:"user"`
|
|
Password string `json:"password"`
|
|
}
|
|
|
|
func readCredential(path string) (credential, error) {
|
|
raw, err := os.ReadFile(path)
|
|
if err != nil {
|
|
return credential{}, err
|
|
}
|
|
var c credential
|
|
if err := json.Unmarshal(raw, &c); err != nil {
|
|
return credential{}, fmt.Errorf("the broker credential is not one: %w", err)
|
|
}
|
|
if c.URL == "" || c.User == "" {
|
|
return credential{}, errors.New("the broker credential names no bus address or user")
|
|
}
|
|
return c, nil
|
|
}
|
|
|
|
// pinned accepts exactly the certificate with this SHA-256 and no other.
|
|
func pinned(want string) *tls.Config {
|
|
return &tls.Config{
|
|
InsecureSkipVerify: true, //nolint:gosec // replaced by the pin below, which is stricter
|
|
MinVersion: tls.VersionTLS12,
|
|
VerifyPeerCertificate: func(rawCerts [][]byte, _ [][]*x509.Certificate) error {
|
|
if len(rawCerts) == 0 {
|
|
return errors.New("the bus presented no certificate")
|
|
}
|
|
sum := sha256.Sum256(rawCerts[0])
|
|
if got := "sha256:" + hex.EncodeToString(sum[:]); got != want {
|
|
return fmt.Errorf("the bus presented a certificate this mesh does not know (%.23s…)", got)
|
|
}
|
|
return nil
|
|
},
|
|
}
|
|
}
|
|
|
|
// run is the program: it returns when ctx ends or the bus cannot be used.
|
|
func run(ctx context.Context, getenv func(string) string, logOut io.Writer) error {
|
|
logger := log.New(logOut, "mesh-delivery-builder: ", log.LstdFlags)
|
|
path := strings.TrimSpace(getenv("MESH_BROKER_FILE"))
|
|
if path == "" {
|
|
return errors.New("MESH_BROKER_FILE is not set, so there is no credential to hear the bus with")
|
|
}
|
|
cred, err := readCredential(path)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
opts := []nats.Option{
|
|
nats.Name(cred.Node + "." + cred.Module),
|
|
nats.UserInfo(cred.User, cred.Password),
|
|
nats.CustomInboxPrefix("_INBOX." + cred.User),
|
|
nats.MaxReconnects(-1),
|
|
nats.RetryOnFailedConnect(true),
|
|
}
|
|
if strings.TrimSpace(cred.Fingerprint) != "" {
|
|
opts = append(opts, nats.Secure(pinned(cred.Fingerprint)))
|
|
}
|
|
conn, err := nats.Connect(cred.URL, opts...)
|
|
if err != nil {
|
|
return fmt.Errorf("connecting to the bus: %w", err)
|
|
}
|
|
defer conn.Close()
|
|
js, err := jetstream.New(conn)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
bucket := func(name string) (jetstream.KeyValue, error) {
|
|
kv, err := js.KeyValue(ctx, name)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("the bucket %s: %w", name, err)
|
|
}
|
|
return kv, nil
|
|
}
|
|
builds, err := bucket(delivery.BuilderBuildsBucket)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
modules, err := bucket(delivery.ModulesBucket)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
manifests, err := bucket(delivery.ManifestsBucket)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
candidates, err := bucket(delivery.CandidatesBucket)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
// Calls to the build modules go out as this module's own account, naming the machine (mesh-sdk process.CallOn); the
|
|
// buckets are read and written on the connection above.
|
|
account, err := process.ReadAccount(path)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
caller, err := process.Connect(account, nats.Name(cred.Node+"."+cred.Module+".calls"), nats.RetryOnFailedConnect(true))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer caller.Close()
|
|
|
|
bound := defaultStepBound
|
|
if v := strings.TrimSpace(getenv("MESH_BUILDER_STEP_BOUND")); v != "" {
|
|
if bound, err = time.ParseDuration(v); err != nil {
|
|
return fmt.Errorf("MESH_BUILDER_STEP_BOUND: %w", err)
|
|
}
|
|
}
|
|
capacity, err := capacityFrom(strings.TrimSpace(getenv("MESH_BUILDER_SETTINGS")), getenv("MESH_BUILDER_CAPACITY"))
|
|
if err != nil {
|
|
return fmt.Errorf("the build capacity setting: %w", err)
|
|
}
|
|
tools := &buildrun.ToolchainSet{}
|
|
presence := &buildrun.PresenceSet{}
|
|
have := &buildrun.HaveSet{}
|
|
b := &build.Builder{
|
|
Claimer: build.Claimer{Store: kvStore{builds}, By: cred.Node + "." + cred.Module},
|
|
Runner: buildrun.Runner{Caller: caller, Steps: stepsWaiter{js}, Bound: bound},
|
|
Toolchains: tools,
|
|
Presence: presence,
|
|
Health: have,
|
|
Places: &buildrun.Places{Capacity: capacity},
|
|
Manifests: manifestReader{manifests},
|
|
Nodes: nodesPerFingerprint,
|
|
Logf: logger.Printf,
|
|
}
|
|
// The toolchains first, so a build that starts from the first modules value finds the nodes.
|
|
watch := newToolchainWatch(ctx, js, tools, b.NodesChanged, logger)
|
|
if err := watch.start(); err != nil {
|
|
return err
|
|
}
|
|
// A build module says each time it wrote its toolchains key: its stream, and the builder's durable consumer on it, are
|
|
// made by the controller. Where the stream is not there yet the builder says so and is started again with the build
|
|
// module.
|
|
for _, module := range []string{buildrun.GoBuildModule, buildrun.NodeBuildModule} {
|
|
stream := toolchainStreams[module]
|
|
cons, err := js.Consumer(ctx, stream, cred.Node+"_"+cred.Module)
|
|
if err != nil {
|
|
logger.Printf("the consumer %s_%s on %s does not exist yet, so %s's reports are not heard until the builder starts again: %v", cred.Node, cred.Module, stream, module, err)
|
|
continue
|
|
}
|
|
cc, err := cons.Consume(func(m jetstream.Msg) { watch.Reported(m) },
|
|
jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logger.Printf("hearing the bus: %v", err) }))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer cc.Stop()
|
|
}
|
|
// The watchman's presence says which nodes are asleep. Where its bucket is not there yet no node is online, so nothing
|
|
// builds; the builder says so and is started again with the watchman.
|
|
presenceErrs := make(chan error, 1)
|
|
switch kv, err := js.KeyValue(ctx, delivery.PresenceBucket); {
|
|
case errors.Is(err, jetstream.ErrBucketNotFound):
|
|
logger.Printf("the bucket %s does not exist yet: no node counts as online until the builder starts again", delivery.PresenceBucket)
|
|
case err != nil:
|
|
return fmt.Errorf("the bucket %s: %w", delivery.PresenceBucket, err)
|
|
default:
|
|
pw, err := settle(ctx, kv, func(e jetstream.KeyValueEntry) { presence.Put(e.Key(), e.Value(), !opOf(e)) })
|
|
if err != nil {
|
|
return err
|
|
}
|
|
go func() {
|
|
presenceErrs <- follow(ctx, pw, func(e jetstream.KeyValueEntry) {
|
|
if presence.Put(e.Key(), e.Value(), !opOf(e)) {
|
|
b.NodesChanged(ctx) // a node came online or went: a build that waited for one may go on
|
|
}
|
|
})
|
|
}()
|
|
}
|
|
// The node-engine's have says whether a node's build module runs and is healthy: a node whose build module is not is not
|
|
// chosen (ADR 0478 clause 3). Where its bucket is not there yet no build module counts as healthy, so nothing builds.
|
|
haveErrs := make(chan error, 1)
|
|
switch kv, err := js.KeyValue(ctx, delivery.HaveBucket); {
|
|
case errors.Is(err, jetstream.ErrBucketNotFound):
|
|
logger.Printf("the bucket %s does not exist yet: no build module counts as healthy until the builder starts again", delivery.HaveBucket)
|
|
case err != nil:
|
|
return fmt.Errorf("the bucket %s: %w", delivery.HaveBucket, err)
|
|
default:
|
|
hw, err := settle(ctx, kv, func(e jetstream.KeyValueEntry) { have.Put(e.Key(), e.Value(), !opOf(e)) })
|
|
if err != nil {
|
|
return err
|
|
}
|
|
go func() {
|
|
haveErrs <- follow(ctx, hw, func(e jetstream.KeyValueEntry) {
|
|
if have.Put(e.Key(), e.Value(), !opOf(e)) {
|
|
b.NodesChanged(ctx) // a build module became healthy or stopped being: a build that waited may go on
|
|
}
|
|
})
|
|
}()
|
|
}
|
|
each := func(e jetstream.KeyValueEntry) {
|
|
if opOf(e) {
|
|
b.Module(ctx, e.Key(), e.Value())
|
|
} else {
|
|
b.Forget(e.Key())
|
|
}
|
|
}
|
|
w, err := settle(ctx, modules, each)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
// A key left building for a fingerprint its module no longer has is closed (issue 1660); the modules are all known now.
|
|
if err := b.Sweep(ctx); err != nil {
|
|
return err
|
|
}
|
|
// The fingerprints the indexer gives the candidates' modules are built too (issue 1684), once the trunk's modules are known.
|
|
eachCandidate := func(e jetstream.KeyValueEntry) {
|
|
if opOf(e) {
|
|
b.Candidate(ctx, e.Key(), e.Value())
|
|
} else {
|
|
b.ForgetCandidate(e.Key())
|
|
}
|
|
}
|
|
cw, err := settle(ctx, candidates, eachCandidate)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
logger.Printf("started on %s", cred.Node)
|
|
errs := make(chan error, 2)
|
|
go func() { errs <- follow(ctx, w, each) }()
|
|
go func() { errs <- follow(ctx, cw, eachCandidate) }()
|
|
select {
|
|
case <-ctx.Done():
|
|
b.Wait()
|
|
return nil
|
|
case err := <-errs:
|
|
return err
|
|
case err := <-watch.errs:
|
|
return err
|
|
case err := <-presenceErrs:
|
|
return err
|
|
case err := <-haveErrs:
|
|
return err
|
|
}
|
|
}
|