Merge branch 'feat/47-G-kepton7' into feat/47-G-join8
This commit is contained in:
@@ -0,0 +1,124 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"sync"
|
||||
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
|
||||
"mesh-delivery-assigner/internal/assign"
|
||||
)
|
||||
|
||||
// What a want names of a build is heard, never read from the builder's state (design 47 G §379, §526, A-94, A-96; ADR 0596 §2): the
|
||||
// builder's event `build.ended` (a done build's parts by digest, its keeping node and its branch) and the catalog's
|
||||
// `version-released` (a released version and each part's digest, with the source revision it was built from), each kept as heard.
|
||||
|
||||
// buildEndedSubjects is the builder's event build.ended on its own stream, as the manifest's consumes names it
|
||||
// (`mesh-delivery-builder.build.ended`), each module and fingerprint after it.
|
||||
var buildEndedSubjects = process.ModuleStreamSubject(builds.BuilderModule, builds.BuildEndedEvent+".>")
|
||||
|
||||
// builtEnds are the done builds' ends heard, by module and fingerprint (hearBuildEnded).
|
||||
type builtEnds struct {
|
||||
mu sync.Mutex
|
||||
ends map[string]builds.BuildEnded
|
||||
}
|
||||
|
||||
func (b *builtEnds) key(module, fp string) string { return module + "@" + fp }
|
||||
|
||||
// put takes one build.ended in.
|
||||
func (b *builtEnds) put(data []byte) error {
|
||||
var e builds.BuildEnded
|
||||
if err := json.Unmarshal(data, &e); err != nil {
|
||||
return fmt.Errorf("a build.ended that is no end: %w", err)
|
||||
}
|
||||
if e.Module == "" || e.Fingerprint == "" {
|
||||
return fmt.Errorf("a build's end naming no module or fingerprint: %s", data)
|
||||
}
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
if b.ends == nil {
|
||||
b.ends = map[string]builds.BuildEnded{}
|
||||
}
|
||||
b.ends[b.key(e.Module, e.Fingerprint)] = e
|
||||
return nil
|
||||
}
|
||||
|
||||
// partsOf is the parts of module's build at fp as its end names them (each artifact's digest by `<module>:<artifact>`), with the
|
||||
// node that keeps it, for a kept build or a build of another watched branch; false for a build of the trunk not kept, whose
|
||||
// parts the version's release says (releasedVersions).
|
||||
func (b *builtEnds) partsOf(module, fp string) (assign.KeptBuild, bool) {
|
||||
b.mu.Lock()
|
||||
e, ok := b.ends[b.key(module, fp)]
|
||||
b.mu.Unlock()
|
||||
if !ok || (e.KeptOn == "" && e.Branch == "") {
|
||||
return assign.KeptBuild{}, false
|
||||
}
|
||||
parts := map[string]string{}
|
||||
for name, digest := range e.Made {
|
||||
parts[module+":"+name] = digest
|
||||
}
|
||||
return assign.KeptBuild{Parts: parts, KeptOn: e.KeptOn}, true
|
||||
}
|
||||
|
||||
// releasedVersions are the catalog's version-released heard, by module and source revision.
|
||||
type releasedVersions struct {
|
||||
mu sync.Mutex
|
||||
seen map[string]assign.KeptBuild
|
||||
}
|
||||
|
||||
func (r *releasedVersions) key(module string, source int64) string {
|
||||
return fmt.Sprintf("%s@%d", module, source)
|
||||
}
|
||||
|
||||
// put takes one version-released in.
|
||||
func (r *releasedVersions) put(data []byte) error {
|
||||
var e meshcatalog.VersionReleasedEvent
|
||||
if err := json.Unmarshal(data, &e); err != nil {
|
||||
return fmt.Errorf("a version-released that is none: %w", err)
|
||||
}
|
||||
if e.Module == "" || e.Version == "" {
|
||||
return fmt.Errorf("a version-released naming no module or version: %s", data)
|
||||
}
|
||||
parts := map[string]string{}
|
||||
for _, a := range e.Artifacts {
|
||||
parts[e.Module+":"+a.Artifact] = a.Digest
|
||||
}
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
if r.seen == nil {
|
||||
r.seen = map[string]assign.KeptBuild{}
|
||||
}
|
||||
r.seen[r.key(e.Module, e.Source)] = assign.KeptBuild{Version: e.Version, Parts: parts}
|
||||
return nil
|
||||
}
|
||||
|
||||
// of is the version of module released from the source revision, with its parts' digests; false before it is heard.
|
||||
func (r *releasedVersions) of(module string, source int64) (assign.KeptBuild, bool) {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
k, ok := r.seen[r.key(module, source)]
|
||||
return k, ok
|
||||
}
|
||||
|
||||
// hearBuildsAndReleases registers the hub's takers of build.ended and version-released, each waking the pass; before the hub starts.
|
||||
func hearBuildsAndReleases(hub *process.EventHub, ends *builtEnds, released *releasedVersions, wake func(), logf func(string, ...any)) error {
|
||||
taker := func(put func([]byte) error) func(jetstream.Msg) {
|
||||
return func(m jetstream.Msg) {
|
||||
if err := put(m.Data()); err != nil {
|
||||
logf("%s is not kept: %v", m.Subject(), err)
|
||||
}
|
||||
if err := m.Ack(); err != nil {
|
||||
logf("acknowledging %s: %v", m.Subject(), err)
|
||||
}
|
||||
wake()
|
||||
}
|
||||
}
|
||||
if err := hub.Handle(buildEndedSubjects, taker(ends.put)); err != nil {
|
||||
return err
|
||||
}
|
||||
return hub.Handle(meshcatalog.SubjectVersionReleased, taker(released.put))
|
||||
}
|
||||
@@ -0,0 +1,88 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"reflect"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
|
||||
"mesh-delivery-assigner/internal/assign"
|
||||
)
|
||||
|
||||
// Design 47 G §379, §526, A-96: a build.ended heard names a kept build's or another branch's parts by digest (`<module>:<artifact>`) and its
|
||||
// keeping node; a build of the trunk not kept names none, its parts being the release's.
|
||||
func TestABuildEndedHeardNamesAKeptBuildsPartsAndItsNode(t *testing.T) {
|
||||
put := func(b *builtEnds, e builds.BuildEnded) {
|
||||
raw, _ := json.Marshal(e)
|
||||
if err := b.put(raw); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
made := map[string]string{"program": "sha256:1"}
|
||||
var b builtEnds
|
||||
put(&b, builds.BuildEnded{Module: "node-engine", Fingerprint: "a", Made: made})
|
||||
if _, ok := b.partsOf("node-engine", "a"); ok {
|
||||
t.Error("a trunk build not kept names parts")
|
||||
}
|
||||
put(&b, builds.BuildEnded{Module: "node-engine", Fingerprint: "b", Made: made, KeptOn: "ace"})
|
||||
if k, ok := b.partsOf("node-engine", "b"); !ok || k.KeptOn != "ace" || !reflect.DeepEqual(k.Parts, map[string]string{"node-engine:program": "sha256:1"}) {
|
||||
t.Errorf("kept build = %+v, %v", k, ok)
|
||||
}
|
||||
put(&b, builds.BuildEnded{Module: "node-engine", Fingerprint: "c", Made: made, Branch: "cand"})
|
||||
if k, ok := b.partsOf("node-engine", "c"); !ok || k.KeptOn != "" || len(k.Parts) != 1 {
|
||||
t.Errorf("branch build = %+v, %v", k, ok)
|
||||
}
|
||||
if err := b.put([]byte(`{"module":"x"}`)); err == nil {
|
||||
t.Error("an end naming no fingerprint is kept")
|
||||
}
|
||||
}
|
||||
|
||||
// Design 47 G §379, A-96: a version-released heard names the version and the parts' digests of the module's source revision.
|
||||
func TestAVersionReleasedHeardNamesItsVersionAndPartsBySource(t *testing.T) {
|
||||
var r releasedVersions
|
||||
raw, _ := json.Marshal(meshcatalog.VersionReleasedEvent{Module: "mail", Source: 7, Version: "1.2.3", Artifacts: []meshcatalog.VersionArtifact{{Artifact: "image", Digest: "sha256:9"}}})
|
||||
if err := r.put(raw); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
want := assign.KeptBuild{Version: "1.2.3", Parts: map[string]string{"mail:image": "sha256:9"}}
|
||||
if k, ok := r.of("mail", 7); !ok || !reflect.DeepEqual(k, want) {
|
||||
t.Errorf("released = %+v, %v", k, ok)
|
||||
}
|
||||
if _, ok := r.of("mail", 8); ok {
|
||||
t.Error("another source revision names a version")
|
||||
}
|
||||
}
|
||||
|
||||
// Design 47 G §379, A-96: the trunk's entry takes its version and parts from the version-released of the kept assignment graph
|
||||
// revision's source; a kept build from its end; no revision kept, or another fingerprint, names none.
|
||||
func TestATrunkEntryTakesItsVersionFromTheReleaseOfTheKeptRevisionsSource(t *testing.T) {
|
||||
w := newWorld()
|
||||
raw, _ := json.Marshal(meshcatalog.VersionReleasedEvent{Module: "mail", Source: 7, Version: "1.2.3", Artifacts: []meshcatalog.VersionArtifact{{Artifact: "image", Digest: "sha256:9"}}})
|
||||
if err := w.released.put(raw); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
endRaw, _ := json.Marshal(builds.BuildEnded{Module: "mail", Fingerprint: "k", Made: map[string]string{"image": "sha256:2"}, KeptOn: "g14"})
|
||||
if err := w.ends.put(endRaw); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
p := &passer{w: w, graph: &graphKeeper{ok: true, source: 7}}
|
||||
named := p.namedBuild("mail", "t")
|
||||
if k, ok := named("t"); !ok || k.Version != "1.2.3" || k.Parts["mail:image"] != "sha256:9" || k.KeptOn != "" {
|
||||
t.Errorf("trunk entry = %+v, %v", k, ok)
|
||||
}
|
||||
if k, ok := named("k"); !ok || k.KeptOn != "g14" || k.Version != "" {
|
||||
t.Errorf("kept build = %+v, %v", k, ok)
|
||||
}
|
||||
if _, ok := named("other"); ok {
|
||||
t.Error("another fingerprint names a release")
|
||||
}
|
||||
p.graph.source = 8
|
||||
if _, ok := named("t"); ok {
|
||||
t.Error("a source with no release names a version")
|
||||
}
|
||||
p.graph.ok = false
|
||||
if _, ok := p.namedBuild("mail", "t")("t"); ok {
|
||||
t.Error("no revision kept names a release")
|
||||
}
|
||||
}
|
||||
@@ -14,6 +14,7 @@ import (
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants"
|
||||
"git.novox.be/novox/mesh-sdk/go/keys"
|
||||
|
||||
"mesh-delivery-assigner/internal/assign"
|
||||
"mesh-delivery-assigner/internal/assigngraph"
|
||||
)
|
||||
|
||||
@@ -118,3 +119,18 @@ func (p *passer) sayWanted(node, module, fp string, removed bool) {
|
||||
p.logf("wanted of %s on %s is not said: %v", module, node, err)
|
||||
}
|
||||
}
|
||||
|
||||
// namedBuild answers what a want of module at fp names of its build (design 47 G §379, §526, A-96): a kept build's or another watched
|
||||
// branch's parts and keeping node from the builder's `build.ended`; for the trunk's entry (fp is the module's, trunkFP), the released
|
||||
// version and the parts of the `version-released` whose source is the one the kept assignment graph revision was computed from (A-67).
|
||||
func (p *passer) namedBuild(module, trunkFP string) func(fp string) (assign.KeptBuild, bool) {
|
||||
return func(fp string) (assign.KeptBuild, bool) {
|
||||
if k, ok := p.w.ends.partsOf(module, fp); ok {
|
||||
return k, true
|
||||
}
|
||||
if fp != trunkFP || p.graph == nil || !p.graph.ok {
|
||||
return assign.KeptBuild{}, false
|
||||
}
|
||||
return p.w.released.of(module, p.graph.source)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -576,6 +576,9 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
|
||||
logger.Printf("a pipeline's end or a registration is not heard, so the wants keep the trunk-only form until the assigner starts again: %v", err)
|
||||
}
|
||||
}
|
||||
if err := hearBuildsAndReleases(hub, &w.ends, &w.released, wake, logger.Printf); err != nil {
|
||||
logger.Printf("a build's end or a version's release is not heard, so no want names a build's parts, its keeping node or its version until the assigner starts again: %v", err)
|
||||
}
|
||||
hearReadsChanged(ctx, js, hub, reads, findings, gate.EventsLive, logger.Printf)
|
||||
hearRunning(ctx, js, w, wake, findings, logger.Printf)
|
||||
}
|
||||
|
||||
@@ -325,6 +325,7 @@ func (p *passer) wants(ctx context.Context) {
|
||||
held, _ := mwc.HeldKey(name, fp)
|
||||
grant, _ := grantsstate.PartGrantKey(name, fp)
|
||||
hold, _ := asks.HoldAskKey(name, fp)
|
||||
m.KeptBuilds = p.namedBuild(name, m.FP)
|
||||
m.Released = p.answeredWith(hold, "release")
|
||||
m.Kept = p.answeredWith(hold, "keep")
|
||||
m.Built, m.Held, m.Granted = p.w.built(name, fp), p.w.has(buckets.Wants, held) && !m.Released, p.w.has(grants, grant)
|
||||
|
||||
@@ -41,8 +41,10 @@ type world struct {
|
||||
changes map[string]uint64
|
||||
// meta is, by bucket and key, the revision and the time of the value held now, where the watch said them (putEntry).
|
||||
meta map[string]entryMeta
|
||||
// trunkBuilt is, by module, its newest done trunk build in the builder's log (putBuilt).
|
||||
trunkBuilt map[string]builds.BuildLogEntry
|
||||
// ends are the done builds' ends the builder says (build.ended), by module and fingerprint (buildlog.go).
|
||||
ends builtEnds
|
||||
// released are the catalog's version-released heard, by module and source revision (buildended.go).
|
||||
released releasedVersions
|
||||
// assignments is the seat's state assignments assembled node by node from its keys in either form (mesh-sdk
|
||||
// mmac.Keys; research 072 row 37): the one reader of its keys in the assigner.
|
||||
assignmentKeys *mmac.Keys
|
||||
@@ -74,7 +76,7 @@ type parsed struct {
|
||||
|
||||
func newWorld() *world {
|
||||
return &world{values: map[string]map[string][]byte{}, manifests: map[string]parsed{}, healthy: map[string][]string{}, changes: map[string]uint64{},
|
||||
meta: map[string]entryMeta{}, trunkBuilt: map[string]builds.BuildLogEntry{}, assignmentKeys: mmac.NewKeys(),
|
||||
meta: map[string]entryMeta{}, assignmentKeys: mmac.NewKeys(),
|
||||
sourceGroups: map[uint64]string{}, running: map[string]nodeengine.Running{},
|
||||
modules: map[string]map[string]meshcatalog.ModulesByModuleValue{}, graphs: map[string]map[string]graph.Node{}}
|
||||
}
|
||||
|
||||
@@ -95,11 +95,41 @@ type Module struct {
|
||||
// source of a want's Self: every want the rules write or keep is derived from it (selfDerived), so a want written Self while
|
||||
// the node was placed does not outlive the assignments (issue 1126). Nil: no node does.
|
||||
Self func(node string) bool
|
||||
// KeptBuilds answers what a want at a fingerprint names of its build: its parts and its keeping node as the builder's `build.ended`
|
||||
// names them, or the released version and parts of the catalog's `version-released` (design 47 G §379, §526, A-96), for a node that states the capability WantParts. Nil: none is read.
|
||||
KeptBuilds func(fp string) (KeptBuild, bool)
|
||||
// MachineHeld says whether the node's own network condition is open, which holds the machine as a whole: no want moves to it
|
||||
// until that clears (design 47 G §706 *M's own network*). Nil: none is held.
|
||||
MachineHeld func(node string) bool
|
||||
}
|
||||
|
||||
// WantParts is the capability a node-engine of G states (ADR 0614 §4, design 47 G §525): it reads a want's `parts` and `kept-on`.
|
||||
const WantParts = "want-parts"
|
||||
|
||||
// KeptBuild is what a want names of a build: the digest of each of its parts by `<module>:<artifact>`, and, as the builder's
|
||||
// `build.ended` says it, the build node that keeps a kept build; for a trunk build the released version the catalog's
|
||||
// `version-released` says (design 47 G §379, §526, A-96).
|
||||
type KeptBuild struct {
|
||||
Parts map[string]string
|
||||
KeptOn string
|
||||
Version string
|
||||
}
|
||||
|
||||
// withKept is w naming the kept build at its fingerprint, the parts by digest and the keeping node, only while the node's
|
||||
// engine states WantParts (design 47 G §525, §526); a want of a build that is not kept, or of a node that does not state it,
|
||||
// is w.
|
||||
func (n Nodes) withKept(m Module, node string, w mwc.Want) mwc.Want {
|
||||
if m.KeptBuilds == nil {
|
||||
return w
|
||||
}
|
||||
k, ok := m.KeptBuilds(w.Fp)
|
||||
if !ok || !slices.Contains(n.Reads[node], WantParts) {
|
||||
return w
|
||||
}
|
||||
w.Parts, w.KeptOn, w.Version = k.Parts, k.KeptOn, k.Version
|
||||
return w
|
||||
}
|
||||
|
||||
func (m Module) selfOn(node string) bool { return m.Self != nil && m.Self(node) }
|
||||
|
||||
func (m Module) composition(node, fp string) (ComposedDigest, bool) {
|
||||
@@ -514,7 +544,7 @@ func Wants(m Module, n Nodes, wants map[string]mwc.Want, lastHealthy map[string]
|
||||
switch {
|
||||
case ok:
|
||||
d, _ := m.composition(node, target)
|
||||
next = mwc.Want{Fp: target, Composed: d.Digest}
|
||||
next = n.withKept(m, node, mwc.Want{Fp: target, Composed: d.Digest})
|
||||
at := m
|
||||
if target != m.FP {
|
||||
// the first want at the older fingerprint the node runs: held back from F by the rule, and judged as a want at it
|
||||
|
||||
@@ -1583,3 +1583,28 @@ func TestAProviderRunsHealthyByItsRunning(t *testing.T) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Design 47 G §525, §526: a want at a kept build names its parts and the node that keeps it, only for a node whose engine states
|
||||
// the capability want-parts.
|
||||
func TestAWantNamesAKeptBuildsPartsAndNodeForANodeThatReadsThem(t *testing.T) {
|
||||
kept := KeptBuild{Parts: map[string]string{"mail:image": "sha256:1"}, KeptOn: "g14", Version: "1.2.3"}
|
||||
for name, row := range map[string]struct {
|
||||
reads []string
|
||||
kept func(string) (KeptBuild, bool)
|
||||
wanted bool
|
||||
}{
|
||||
"states it": {[]string{WantParts}, func(fp string) (KeptBuild, bool) { return kept, fp == fNew }, true},
|
||||
"does not state": {[]string{"form:oci"}, func(fp string) (KeptBuild, bool) { return kept, fp == fNew }, false},
|
||||
"build not kept": {[]string{WantParts}, func(string) (KeptBuild, bool) { return KeptBuild{}, false }, false},
|
||||
"no kept builds": {[]string{WantParts}, nil, false},
|
||||
"another fingerp": {[]string{WantParts}, func(fp string) (KeptBuild, bool) { return kept, fp == fOld }, false},
|
||||
} {
|
||||
m, n := ready()
|
||||
n.Reads = map[string][]string{"ace": row.reads, "g14": row.reads}
|
||||
m.KeptBuilds = row.kept
|
||||
w := Wants(m, n, wantedOld(), nil).Wants["ace"]
|
||||
if w.Fp != fNew || (w.KeptOn == "g14") != row.wanted || (w.Version == "1.2.3") != row.wanted || (len(w.Parts) == 1) != row.wanted {
|
||||
t.Errorf("%s: ace's want = %+v", name, w)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"encoding/json"
|
||||
"git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-teller/tellerevents"
|
||||
"os"
|
||||
"reflect"
|
||||
@@ -129,7 +130,7 @@ func TestManifest(t *testing.T) {
|
||||
}
|
||||
// ADR 0618 §4: and it hears each node's engine say its reads changed; ADR 0636 §4 (issue 1966): and the seat list's two events;
|
||||
// issue 1966 note 151: and the teller's ask-ended, its subject the one the assigner hears.
|
||||
if !slices.Equal(m.Consumes, []string{"mesh-conditions.findings-asked", "mesh-conditions.condition-raised", "mesh-conditions.condition-changed", "mesh-conditions.condition-cleared", "mesh-catalog.catalog-revision", "mesh-catalog.registration-changed", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", buckets.TellerModule + "." + tellerevents.AskEnded, "mesh-pipeline-runner.pipeline-ended", "mesh-pipeline-runner.group-ended", "node-engine.manifest-reads-changed", "node-engine.running"}) {
|
||||
if !slices.Equal(m.Consumes, []string{"mesh-conditions.findings-asked", "mesh-conditions.condition-raised", "mesh-conditions.condition-changed", "mesh-conditions.condition-cleared", "mesh-catalog.catalog-revision", "mesh-catalog.registration-changed", "mesh-catalog.version-released", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", builds.BuilderModule + "." + builds.BuildEndedEvent, buckets.TellerModule + "." + tellerevents.AskEnded, "mesh-pipeline-runner.pipeline-ended", "mesh-pipeline-runner.group-ended", "node-engine.manifest-reads-changed", "node-engine.running"}) {
|
||||
t.Errorf("consumes = %v, want the catalog half's revisions (design 47 G §195), the teller's ask-ended, the engines' reads changed and what they run", m.Consumes)
|
||||
}
|
||||
if want := []string{"mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules",
|
||||
|
||||
@@ -50,8 +50,10 @@
|
||||
"mesh-conditions.condition-cleared",
|
||||
"mesh-catalog.catalog-revision",
|
||||
"mesh-catalog.registration-changed",
|
||||
"mesh-catalog.version-released",
|
||||
"mesh-catalog.seat-definition-changed",
|
||||
"mesh-catalog.seat-definition-removed",
|
||||
"mesh-delivery-builder.build.ended",
|
||||
"mesh-delivery-teller.ask-ended",
|
||||
"mesh-pipeline-runner.pipeline-ended",
|
||||
"mesh-pipeline-runner.group-ended",
|
||||
|
||||
@@ -0,0 +1,43 @@
|
||||
package builds
|
||||
|
||||
import (
|
||||
"git.novox.be/novox/mesh-sdk/go/keys"
|
||||
)
|
||||
|
||||
// The builder says each done build's end itself, as an event on its own stream (design 47 G §379, A-96): the stream
|
||||
// BuildsStream carries `build.ended.<module>.<fingerprint>` (each part escaped), published where the log entry of the end is
|
||||
// written. A part that wants a build's parts or its keeping node hears it (the assigner consumes `mesh-delivery-builder.build.ended`).
|
||||
const (
|
||||
// BuildsStream is the builder's stream of build ends, named by its manifest's `streams` entry.
|
||||
BuildsStream = "mesh-delivery-builder-builds"
|
||||
// BuildsStreamEvents is the pattern of the events that entry carries.
|
||||
BuildsStreamEvents = BuildEndedEvent + ".**"
|
||||
)
|
||||
|
||||
// BuildEnded is the body of the event build.ended: the module, the fingerprint, each artifact made with its digest, the build
|
||||
// node that keeps a kept build (none for one not kept) and the watched branch a build of a candidate is for (none for the trunk).
|
||||
type BuildEnded struct {
|
||||
Module string `json:"module"`
|
||||
Fingerprint string `json:"fingerprint"`
|
||||
Made map[string]string `json:"made,omitempty"`
|
||||
KeptOn string `json:"kept-on,omitempty"`
|
||||
Branch string `json:"branch,omitempty"`
|
||||
}
|
||||
|
||||
// BuildEndedEventName is the event's name for a module and fingerprint: `build.ended.<module>.<fingerprint>`, each escaped.
|
||||
func BuildEndedEventName(module, fp string) string {
|
||||
return BuildEndedEvent + "." + keys.EscapeName(module) + "." + keys.EscapeName(fp)
|
||||
}
|
||||
|
||||
// BuildEndedOf is the event of a log entry of a build's end: false for any entry but the end of a done build, which says no
|
||||
// digest of any part.
|
||||
func BuildEndedOf(e BuildLogEntry) (BuildEnded, bool) {
|
||||
if e.Event != BuildEndedEvent || e.Outcome != BuildDone || e.Module == "" || e.FP == "" {
|
||||
return BuildEnded{}, false
|
||||
}
|
||||
made := map[string]string{}
|
||||
for _, a := range e.Made {
|
||||
made[a.Name] = a.Digest
|
||||
}
|
||||
return BuildEnded{Module: e.Module, Fingerprint: e.FP, Made: made, KeptOn: e.KeptOn, Branch: e.Ref}, true
|
||||
}
|
||||
@@ -0,0 +1,32 @@
|
||||
package builds
|
||||
|
||||
import (
|
||||
"reflect"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// Design 47 G §379, A-96: a done build's end is the event build.ended, its subject escaped by module and fingerprint; any other
|
||||
// entry is none.
|
||||
func TestADoneBuildsEndIsTheEventBuildEnded(t *testing.T) {
|
||||
made := []BuiltArtifact{{Name: "program", Kind: "program", Digest: "sha256:1"}}
|
||||
e := BuildLogEntry{Event: BuildEndedEvent, Outcome: BuildDone, Module: "node-engine", FP: "src2:a", Ref: "cand", KeptOn: "ace", Made: made}
|
||||
got, ok := BuildEndedOf(e)
|
||||
want := BuildEnded{Module: "node-engine", Fingerprint: "src2:a", Made: map[string]string{"program": "sha256:1"}, KeptOn: "ace", Branch: "cand"}
|
||||
if !ok || !reflect.DeepEqual(got, want) {
|
||||
t.Errorf("event = %+v, %v", got, ok)
|
||||
}
|
||||
if n := BuildEndedEventName("node-engine", "src2:a"); n != "build.ended.node-engine.src2=3aa" {
|
||||
t.Errorf("event name = %q", n)
|
||||
}
|
||||
for name, mut := range map[string]func(*BuildLogEntry){
|
||||
"started": func(e *BuildLogEntry) { e.Event = BuildStartedEvent },
|
||||
"failed": func(e *BuildLogEntry) { e.Outcome = BuildFailed },
|
||||
"no fp": func(e *BuildLogEntry) { e.FP = "" },
|
||||
} {
|
||||
c := e
|
||||
mut(&c)
|
||||
if _, ok := BuildEndedOf(c); ok {
|
||||
t.Errorf("%s: an event", name)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -314,6 +314,10 @@ func startBedWith(t *testing.T, closed bool) *bed {
|
||||
"mesh.stream.mesh-delivery-builder.event.finding-cleared.*", "mesh.stream.mesh-delivery-builder.event.findings-stated.*"}}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// The builder's stream of build ends (design 47 G A-96), which it says build.ended into.
|
||||
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "BUILDER_BUILDS", Subjects: []string{"mesh.stream.mesh-delivery-builder.event.build.ended.>"}}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
b.fake = &fakeBuildModules{js: js, ctx: ctx, calls: map[string]int{}}
|
||||
b.fake.closed.Store(closed)
|
||||
b.fake.serve(t, "mesh-build-go", "n1", "n2")
|
||||
@@ -555,6 +559,18 @@ func TestBuilderMakesADoneKeyWithTheDigests(t *testing.T) { // 1607/AC7
|
||||
ended[1].Manifest != digest || len(ended[1].Made) != 2 || ended[1].Node != "n1" {
|
||||
t.Errorf("logged = %+v, want the start and the end of netcheck's build", ended)
|
||||
}
|
||||
// Design 47 G §379, A-96: the end is also said as the event build.ended on the builder's stream, its body the module, the
|
||||
// fingerprint and each artifact's digest.
|
||||
if st, err := b.js.Stream(b.ctx, "BUILDER_BUILDS"); err != nil {
|
||||
t.Fatal(err)
|
||||
} else if raw, err := st.GetLastMsgForSubject(b.ctx, "mesh.stream.mesh-delivery-builder.event."+builds.BuildEndedEventName("netcheck", fp)); err != nil {
|
||||
t.Errorf("no build.ended of netcheck's build: %v", err)
|
||||
} else {
|
||||
var ev builds.BuildEnded
|
||||
if err := json.Unmarshal(raw.Data, &ev); err != nil || ev.Module != "netcheck" || ev.Fingerprint != fp || len(ev.Made) != 2 {
|
||||
t.Errorf("build.ended = %s (%v), want netcheck at its fingerprint with two artifacts' digests", raw.Data, err)
|
||||
}
|
||||
}
|
||||
for tool, want := range map[string]int{"mesh-build-go.build_go_fetch": 1, "mesh-build-go.build_go_build": 1, "mesh-build-node.build_node_fetch": 1, "mesh-build-node.build_node_build": 1} {
|
||||
if got := b.fake.count(tool); got != want {
|
||||
t.Errorf("%s called %d times, want %d (on one node, ADR 0478 clause 4)", tool, got, want)
|
||||
|
||||
@@ -253,6 +253,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
if err := appendLog(context.WithoutCancel(ctx), js, cred, e); err != nil {
|
||||
logger.Printf("%v", err)
|
||||
}
|
||||
if err := emitBuildEnded(context.WithoutCancel(ctx), js, cred, e); err != nil {
|
||||
logger.Printf("%v", err)
|
||||
}
|
||||
// A failed trunk build is the builder's finding build-failed, cleared by a later done one (DLV-B7).
|
||||
found.buildEnded(context.WithoutCancel(ctx), e)
|
||||
}
|
||||
|
||||
@@ -161,6 +161,34 @@ func appendLog(ctx context.Context, js jetstream.JetStream, cred process.Account
|
||||
return nil
|
||||
}
|
||||
|
||||
// emitBuildEnded says the builder's event build.ended of a done build, on the builder's own stream (design 47 G §379, A-96):
|
||||
// mesh.stream.<module>.event.build.ended.<module>.<fingerprint>, published where the log entry of the end is written. An
|
||||
// entry that is not a done build's end says none. The id makes a repeated end one event.
|
||||
func emitBuildEnded(ctx context.Context, js jetstream.JetStream, cred process.Account, e builds.BuildLogEntry) error {
|
||||
body, ok := builds.BuildEndedOf(e)
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
data, err := json.Marshal(body)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
msg := nats.NewMsg(cred.StreamSubject(builds.BuildEndedEventName(e.Module, e.FP)))
|
||||
id := "build-ended." + e.Module + "." + e.FP + "." + e.At.Format(time.RFC3339Nano)
|
||||
msg.Header.Set("x-event-id", id)
|
||||
msg.Header.Set("x-source", cred.Module)
|
||||
msg.Header.Set("x-node", cred.Node)
|
||||
msg.Header.Set("x-time", time.Now().UTC().Format(time.RFC3339Nano))
|
||||
msg.Header.Set("content-type", "application/json")
|
||||
msg.Data = data
|
||||
pub, cancel := context.WithTimeout(ctx, 10*time.Second)
|
||||
defer cancel()
|
||||
if _, err := js.PublishMsg(pub, msg, jetstream.WithMsgID(id)); err != nil {
|
||||
return fmt.Errorf("saying %s of %s at %s: %w", builds.BuildEndedEvent, e.Module, e.FP, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// versionActed answers a reasks value of builder_publish_again, builder_floor_again or builder_old_form's drop, a hand act said in the log.
|
||||
func versionActed(ctx context.Context, acts versions.Acts, kv jetstream.KeyValue, e jetstream.KeyValueEntry, r build.ReAsk, logf func(string, ...any)) {
|
||||
var err error
|
||||
|
||||
@@ -175,9 +175,10 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1
|
||||
}
|
||||
// Design 47 G §672: the toolchain waits' stream, kept 7 days within 16 MiB, and the bound's setting, 60 minutes.
|
||||
// Design 47 G A-30, §383: and its own findings' events, which its finding writer says.
|
||||
if len(m.Streams) != 2 || m.Streams[0].Name != "waits" || !reflect.DeepEqual(m.Streams[0].Events, []string{"wait-held.**", "wait-ended.**"}) || m.Streams[0].MaxAgeDays != 7 || m.Streams[0].MaxMiB != 16 ||
|
||||
m.Streams[1].Name != "mesh-delivery-builder-findings" || !reflect.DeepEqual(m.Streams[1].Events, []string{"finding-written.*", "finding-cleared.*", "findings-stated.*"}) {
|
||||
t.Errorf("streams = %+v, want waits with wait-held.** and wait-ended.**, 7 days, 16 MiB, and the findings' stream", m.Streams)
|
||||
if len(m.Streams) != 3 || m.Streams[0].Name != "waits" || !reflect.DeepEqual(m.Streams[0].Events, []string{"wait-held.**", "wait-ended.**"}) || m.Streams[0].MaxAgeDays != 7 || m.Streams[0].MaxMiB != 16 ||
|
||||
m.Streams[1].Name != builds.BuildsStream || !reflect.DeepEqual(m.Streams[1].Events, []string{builds.BuildsStreamEvents}) || m.Streams[1].MaxAgeDays != 7 ||
|
||||
m.Streams[2].Name != "mesh-delivery-builder-findings" || !reflect.DeepEqual(m.Streams[2].Events, []string{"finding-written.*", "finding-cleared.*", "findings-stated.*"}) {
|
||||
t.Errorf("streams = %+v, want waits with wait-held.** and wait-ended.**, 7 days, 16 MiB, the builds' stream (build.ended.**, 7 days) and the findings' stream", m.Streams)
|
||||
}
|
||||
if c, ok := m.Config["toolchain-wait-minutes"]; !ok || c.Default != float64(60) || len(m.Config) != 1 {
|
||||
t.Errorf("config = %+v, want toolchain-wait-minutes of 60", m.Config)
|
||||
|
||||
@@ -118,6 +118,7 @@
|
||||
],
|
||||
"streams": [
|
||||
{ "name": "waits", "events": ["wait-held.**", "wait-ended.**"], "max-age-days": 7, "max-mib": 16 },
|
||||
{ "name": "mesh-delivery-builder-builds", "events": ["build.ended.**"], "max-age-days": 7, "max-mib": 16 },
|
||||
{ "name": "mesh-delivery-builder-findings", "events": ["finding-written.*", "finding-cleared.*", "findings-stated.*"], "max-age-days": 7, "max-mib": 16 }
|
||||
],
|
||||
"config": {
|
||||
|
||||
Reference in New Issue
Block a user