diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go new file mode 100644 index 00000000..a931e9eb --- /dev/null +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go @@ -0,0 +1,51 @@ +// Command mesh-delivery-pipeline is the pipeline runner, holder of the seat mesh-pipeline-runner, a foundation part +// whose kept data are in its manifest's old sections (design 47 G §21, §27, §101). It names no module, branch or +// repository of its own (§7): its own module is the one its bus account names. +package main + +import ( + "context" + "fmt" + "io" + "log" + "os" + "os/signal" + "syscall" + + seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner" + "git.novox.be/novox/mesh-sdk/go/process" +) + +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, "pipeline runner:", err) + os.Exit(1) + } +} + +// run is the program: it returns when ctx ends, or at once when the bus or the state pipelines cannot be used. +func run(ctx context.Context, getenv func(string) string, logOut io.Writer) error { + cred, err := process.ReadBrokerAccount(getenv) + if err != nil { + return err + } + logger := log.New(logOut, cred.Module+": ", log.LstdFlags) + conn, js, err := cred.Dial() + if err != nil { + return err + } + defer conn.Close() + // The state pipelines, `_pipelines`, holds every pipeline by compare-and-set (§27): without it the + // runner keeps nothing, so it stops. + if _, err := process.EssentialBucket(ctx, js, "pipeline runner", ownBucket(cred.Module, seat.StatePipelines), "keep a pipeline"); err != nil { + return err + } + logger.Printf("holding %s", seat.Seat) + <-ctx.Done() + return nil +} + +// ownBucket is a state of the runner's old state section on the bus, `_`. +func ownBucket(module, state string) string { return module + "_" + state } diff --git a/modules/mesh-delivery-pipeline/go.mod b/modules/mesh-delivery-pipeline/go.mod new file mode 100644 index 00000000..511f96a5 --- /dev/null +++ b/modules/mesh-delivery-pipeline/go.mod @@ -0,0 +1,17 @@ +module mesh-delivery-pipeline + +go 1.26.0 + +require ( + git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-sdk/go v0.0.0-00010101000000-000000000000 + github.com/nats-io/nats.go v1.54.0 +) + +require ( + github.com/klauspost/compress v1.20.1 // indirect + github.com/nats-io/nkeys v0.4.16 // indirect + github.com/nats-io/nuid v1.0.1 // indirect + golang.org/x/crypto v0.57.0 // indirect + golang.org/x/sys v0.48.0 // indirect +) diff --git a/modules/mesh-delivery-pipeline/go.sum b/modules/mesh-delivery-pipeline/go.sum new file mode 100644 index 00000000..0daef11f --- /dev/null +++ b/modules/mesh-delivery-pipeline/go.sum @@ -0,0 +1,12 @@ +github.com/klauspost/compress v1.20.1 h1:T7kKElXUMXrUJ2E9QhQhxFtcK5rPyLdsGZvdbLMPdiQ= +github.com/klauspost/compress v1.20.1/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= +github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA= +github.com/nats-io/nats.go v1.54.0/go.mod h1:y+DZoD1oBOYfZTU681eTUiUjI0vbqYGixNVFHcjHJ0k= +github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg= +github.com/nats-io/nkeys v0.4.16/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs= +github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw= +github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= +golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M= +golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA= +golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo= +golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= diff --git a/modules/mesh-delivery-pipeline/manifest_test.go b/modules/mesh-delivery-pipeline/manifest_test.go new file mode 100644 index 00000000..59aa7378 --- /dev/null +++ b/modules/mesh-delivery-pipeline/manifest_test.go @@ -0,0 +1,77 @@ +package pipeline_test + +import ( + "os" + "slices" + "testing" + + seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner" + "git.novox.be/novox/mesh-sdk/go/catalogue" +) + +// TestManifest holds the runner's manifest to its design lines, row by row. +func TestManifest(t *testing.T) { + raw, err := os.ReadFile("module.json") + if err != nil { + t.Fatal(err) + } + m, err := catalogue.Decode(raw) + if err != nil { + t.Fatalf("the manifest does not read: %v", err) + } + var states, streams, objects []string + for _, s := range m.State { + states = append(states, s.Name) + } + for _, s := range m.Streams { + streams = append(streams, s.Name) + } + for _, o := range m.Objects { + objects = append(objects, o.Name) + } + var program catalogue.Artifact + if m.Build == nil { + t.Fatal("the manifest builds nothing") + } + for _, a := range m.Build.Artifacts { + if a.Name == "program" { + program = a + } + } + rows := []struct { + name string + ok bool + }{ + // RN1/AC1 (§21): a foundation part asks no template; its kept data are in old sections. + {"no requires", len(m.Requires) == 0}, + {"claims the seat serving every verb of its definition", len(m.Claims) == 1 && m.Claims[0].Name == seat.Seat && + equalSets(m.Claims[0].Serves, seat.Verbs())}, + // §27, §85, §330, §355, §77: the old state sections. + {"old state sections", equalSets(states, []string{seat.StatePipelines, seat.StateKept, "aged", "findings", "first-views"})}, + // §27, §85: the step log on an old stream, its outputs in an old objects section. + {"old streams section", equalSets(streams, []string{seat.StateStepLog})}, + {"step-log 128 MiB", len(m.Streams) == 1 && m.Streams[0].MaxMiB == 128}, + {"old objects section", equalSets(objects, []string{seat.StateStepOutputs})}, + {"step-outputs 1 GiB", len(m.Objects) == 1 && m.Objects[0].MaxMiB == 1024}, + // §333, §54, §44, §53, §376, §24, §37, §4, §77: what the runner hears. + {"consumes", equalSets(m.Consumes, []string{ + "mesh-delivery-indexer.take-in.taken", "mesh-builds.package-published", "mesh-builds.toolchain-agreed", + "operator-channel.decided", "node-build-go.step-ended", "node-build-node.step-ended", + "node-build-image.step-ended", "mesh-wants.step-ended", "mesh-catalog.registration-changed", + "mesh-catalog.catalog-revision"})}, + // RN1/AC5 (§95, §130): the new assigner is delivered before the runner. + {"program needs the assigner's program", slices.Equal(program.Needs, []string{"mesh-delivery-assigner:program"})}, + } + for _, r := range rows { + if !r.ok { + t.Errorf("%s: not as the design says", r.name) + } + } +} + +func equalSets(a, b []string) bool { + x, y := slices.Clone(a), slices.Clone(b) + slices.Sort(x) + slices.Sort(y) + return slices.Equal(x, y) +} diff --git a/modules/mesh-delivery-pipeline/module.json b/modules/mesh-delivery-pipeline/module.json new file mode 100644 index 00000000..02802158 --- /dev/null +++ b/modules/mesh-delivery-pipeline/module.json @@ -0,0 +1,100 @@ +{ + "module": "mesh-delivery-pipeline", + "version": "1", + "slug": "pipeline", + "claims": [ + { + "name": "mesh-pipeline-runner", + "scope": "mesh", + "serves": [ + "run", + "start", + "show", + "list", + "stands", + "why-carried", + "kept", + "review", + "rerun", + "accept", + "cancel", + "ask-again" + ] + } + ], + "consumes": [ + "mesh-delivery-indexer.take-in.taken", + "mesh-builds.package-published", + "mesh-builds.toolchain-agreed", + "operator-channel.decided", + "node-build-go.step-ended", + "node-build-node.step-ended", + "node-build-image.step-ended", + "mesh-wants.step-ended", + "mesh-catalog.registration-changed", + "mesh-catalog.catalog-revision" + ], + "state": [ + { "name": "pipelines", "history": 1 }, + { "name": "kept", "history": 1 }, + { "name": "aged", "history": 1 }, + { "name": "findings", "history": 1 }, + { "name": "first-views", "history": 1 } + ], + "streams": [ + { "name": "step-log", "events": ["step-log.**"], "max-mib": 128 } + ], + "objects": [ + { "name": "step-outputs", "max-mib": 1024 } + ], + "own-secrets": { + "broker": "${dir:state}/broker" + }, + "secrets-owner": "mesh-delivery-pipeline", + "resources": [ + { + "id": "account", + "type": "user", + "name": "mesh-delivery-pipeline", + "shell": "/usr/bin/nologin", + "home": "/var/lib/mesh-delivery-pipeline" + }, + { + "id": "state", + "type": "directory", + "mode": "0700", + "owner": "mesh-delivery-pipeline", + "place": ".", + "unknown": "program" + }, + { + "id": "pipeline", + "type": "process", + "name": "mesh-delivery-pipeline", + "artifact": "program", + "run": [ + "./mesh-delivery-pipeline" + ], + "env": { + "MESH_BROKER_FILE": "${dir:state}/broker" + }, + "user": "mesh-delivery-pipeline", + "health": { + "kind": "unit" + } + } + ], + "build": { + "artifacts": [ + { + "name": "program", + "kind": "bundle", + "language": "go", + "system": "arch", + "from": "cmd/mesh-delivery-pipeline", + "binary": "mesh-delivery-pipeline", + "needs": ["mesh-delivery-assigner:program"] + } + ] + } +}