jschoubben
175fb45c19
mesh/delivery merged: source.main moved to novox/mesh-host@3e3e43de5214
mesh/check pass 175fb45c19e3 on 3e3e43de5214 | unit 299 run, 1042 kept, 5 skipped | integration 302 run, 1043 kept, 4 skipped | over bound: unit (slow…
586 lines
20 KiB
Go
586 lines
20 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"crypto/ed25519"
|
|
"encoding/base64"
|
|
"encoding/json"
|
|
"fmt"
|
|
"os"
|
|
"path/filepath"
|
|
"sort"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/declaration"
|
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
|
"github.com/novox/mesh-host/internal/agent"
|
|
"github.com/novox/mesh-host/internal/apply"
|
|
"github.com/novox/mesh-host/internal/link"
|
|
"github.com/novox/mesh-host/internal/store"
|
|
)
|
|
|
|
// The node agent's wiring (novox/hq ADR 0429 §4, ADR 0539 §9): the buckets it watches, the haves it writes, the backup
|
|
// it asks and the apply of one module's composition.
|
|
|
|
// bucketWatch is a bucket of the bus as the agent watches it, through the session's link.
|
|
type bucketWatch struct {
|
|
session *link.Session
|
|
bucket string
|
|
}
|
|
|
|
func (w bucketWatch) Watch(ctx context.Context, pattern string) (<-chan agent.Update, error) {
|
|
in, err := w.session.KVWatch(ctx, w.bucket, pattern)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
out := make(chan agent.Update, 64)
|
|
go func() {
|
|
defer close(out)
|
|
for u := range in {
|
|
select {
|
|
case out <- agent.Update{Key: u.Key, Value: u.Value, Revision: u.Revision, Deleted: u.Deleted, Caught: u.Caught}:
|
|
case <-ctx.Done():
|
|
return
|
|
}
|
|
}
|
|
}()
|
|
return out, nil
|
|
}
|
|
|
|
// sessionHaves writes the haves the agent says, through the session.
|
|
type sessionHaves struct{ session *link.Session }
|
|
|
|
func (h sessionHaves) Read(ctx context.Context, module string) (agent.Have, bool) {
|
|
got, ok := h.session.ReadHave(ctx, module)
|
|
if !ok {
|
|
return agent.Have{}, false
|
|
}
|
|
return agent.Have{Module: module, FP: got.FP, Composed: got.Composed, Phase: string(got.Phase), Detail: got.Detail}, true
|
|
}
|
|
|
|
func (h sessionHaves) Write(ctx context.Context, x agent.Have) {
|
|
if x.Module == agent.NodeModule {
|
|
// The node's own key holds its facts and the digest of the composition it runs.
|
|
if x.Phase == "" || x.Phase == string(delivery.PhaseRunning) {
|
|
h.session.NoteNode(ctx, x.Composed)
|
|
}
|
|
return
|
|
}
|
|
phase := x.Phase
|
|
if phase == string(delivery.PhaseRunning) {
|
|
phase = ""
|
|
}
|
|
h.session.NoteModule(ctx, link.ModuleFact{Module: x.Module, FP: x.FP, Composed: x.Composed, Phase: phase,
|
|
Failed: phase == string(delivery.PhaseFailed), Detail: shorten(x.Detail)})
|
|
}
|
|
|
|
func (h sessionHaves) Removed(ctx context.Context, module string) {
|
|
h.session.NoteRemoved(ctx, module, "")
|
|
}
|
|
|
|
// shorten keeps a have's detail within what a have value holds.
|
|
func shorten(s string) string {
|
|
const most = 300
|
|
if len(s) <= most {
|
|
return s
|
|
}
|
|
return s[:most-1] + "…"
|
|
}
|
|
|
|
// backupWithin is how long a backup is given from its ask to say it is done (ADR 0429 §14).
|
|
var backupWithin = 10 * time.Minute
|
|
|
|
// backedUpEvent is where the holder of node-backup says a backup taken by hand ended, also one that could not start.
|
|
const backedUpEvent = "mesh.mod.restic.event.backed-up"
|
|
|
|
// backupNode asks the holder of node-backup on this node to back a module up now, and waits for its event to say how it
|
|
// went: the holder answers `now` when it has started, and says `backed-up` when the backup ends.
|
|
func backupNode(session *link.Session) func(ctx context.Context, module string) agent.Backup {
|
|
return func(ctx context.Context, module string) agent.Backup {
|
|
// Heard before the ask, so an event that comes at once is not lost.
|
|
events, stop, err := session.Hear(backedUpEvent)
|
|
if err != nil {
|
|
return agent.Backup{Words: "the end of the backup cannot be heard: " + err.Error()}
|
|
}
|
|
defer stop()
|
|
// A module this machine backs nothing of up moves without a backup: never held for one that cannot run (issue
|
|
// 1945). An answer that cannot be had or read asks for the backup as before.
|
|
if said, err := session.AskSeat(ctx, "node-backup", "backed-up", map[string]any{"module": module}, 30*time.Second); err == nil {
|
|
if nothing, read := declaresNothingToBackUp(said, module); read && nothing {
|
|
return agent.Backup{Done: true, Words: module + " declares nothing to back up on this machine"}
|
|
}
|
|
}
|
|
asked := time.Now()
|
|
if _, err := session.AskSeat(ctx, "node-backup", "now", map[string]any{"module": module}, 30*time.Second); err != nil {
|
|
return agent.Backup{Words: err.Error()}
|
|
}
|
|
return awaitBackup(ctx, events, session.Membership.Node, module, asked, backupWithin)
|
|
}
|
|
}
|
|
|
|
// awaitBackup waits for the event that says the backup of a module on this node ended, within the bound: done when it
|
|
// says ok, held on its words when it does not, and held when it says nothing within the bound. An event of another node
|
|
// or module, or from before the ask, is not the answer.
|
|
func awaitBackup(ctx context.Context, events <-chan []byte, node, module string, since time.Time, within time.Duration) agent.Backup {
|
|
bound := time.NewTimer(within)
|
|
defer bound.Stop()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return agent.Backup{Words: "the node agent was stopped while the backup ran"}
|
|
case <-bound.C:
|
|
return agent.Backup{Words: fmt.Sprintf("the backup did not say it was done within %s", within)}
|
|
case raw := <-events:
|
|
var e struct {
|
|
Node string `json:"node"`
|
|
Module string `json:"module"`
|
|
OK bool `json:"ok"`
|
|
At time.Time `json:"at"`
|
|
Error string `json:"error"`
|
|
}
|
|
if json.Unmarshal(raw, &e) != nil || e.Node != node || e.Module != module || e.At.Before(since.Add(-time.Second)) {
|
|
continue
|
|
}
|
|
if e.OK {
|
|
return agent.Backup{Done: true}
|
|
}
|
|
words := "the backup failed"
|
|
if e.Error != "" {
|
|
words += ": " + e.Error
|
|
}
|
|
return agent.Backup{Words: words}
|
|
}
|
|
}
|
|
}
|
|
|
|
// declarationOf is the declaration the compositions make together: their resources, and the node's own parts when the
|
|
// node's composition is among them. It is what the node judges and says its health by, and what one module's apply is
|
|
// made of (with a single composition).
|
|
func declarationOf(cs ...delivery.Composed) (*declaration.Declaration, error) {
|
|
d := &declaration.Declaration{Version: declaration.Version}
|
|
for _, c := range cs {
|
|
for _, raw := range c.Resources {
|
|
r, err := declaration.ParseResource(raw)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("a resource of %q: %w", c.Module, err)
|
|
}
|
|
d.Resources = append(d.Resources, r)
|
|
}
|
|
d.Filled = append(d.Filled, c.Filled...)
|
|
if c.Module == "" {
|
|
d.Adoption = c.Own.Adoption
|
|
d.Identities = append(d.Identities, c.Own.Identities...)
|
|
d.Groups = append(d.Groups, c.Own.Groups...)
|
|
d.Libraries = append(d.Libraries, c.Own.Libraries...)
|
|
d.Provisioner = c.Own.Provisioner
|
|
}
|
|
}
|
|
return d, nil
|
|
}
|
|
|
|
func scopeName(c delivery.Composed) string {
|
|
if c.Module == "" {
|
|
return agent.NodeModule
|
|
}
|
|
return c.Module
|
|
}
|
|
|
|
func resourcesOf(cs []delivery.Composed) []declaration.Resource {
|
|
var out []declaration.Resource
|
|
for _, c := range cs {
|
|
for _, raw := range c.Resources {
|
|
if r, err := declaration.ParseResource(raw); err == nil {
|
|
out = append(out, r)
|
|
}
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
// operatorKeysFile is the file beside the node's state that holds the operator's public keys, one ed25519 key in base64 a
|
|
// line (a line that starts with # is a note). The switch's last declaration writes it as a file resource, beside the
|
|
// controller's key the node was enrolled with (novox/hq ADR 0539 §3, §11); a node adopted later has it at its adoption.
|
|
const operatorKeysFile = "operator-keys"
|
|
|
|
// trustedKeys are the keys a composition may be signed by: the controller's, which the node was enrolled with, and the
|
|
// operator's, read from operatorKeysFile each time it is asked, so a key that arrives while the agent runs is trusted at
|
|
// the next composition. A file that cannot be read trusts the controller's alone, and says so.
|
|
func trustedKeys(opts options, controller ed25519.PublicKey) func() []ed25519.PublicKey {
|
|
return func() []ed25519.PublicKey {
|
|
keys := []ed25519.PublicKey{}
|
|
if len(controller) == ed25519.PublicKeySize {
|
|
keys = append(keys, controller)
|
|
}
|
|
raw, err := os.ReadFile(filepath.Join(filepath.Dir(opts.state), operatorKeysFile))
|
|
if err != nil {
|
|
return keys
|
|
}
|
|
for _, line := range strings.Split(string(raw), "\n") {
|
|
line = strings.TrimSpace(line)
|
|
if line == "" || strings.HasPrefix(line, "#") {
|
|
continue
|
|
}
|
|
key, err := base64.StdEncoding.DecodeString(line)
|
|
if err != nil || len(key) != ed25519.PublicKeySize {
|
|
fmt.Fprintf(os.Stderr, "mesh-host: a line of %s is not an ed25519 public key in base64, and is not trusted\n", operatorKeysFile)
|
|
continue
|
|
}
|
|
keys = append(keys, ed25519.PublicKey(key))
|
|
}
|
|
return keys
|
|
}
|
|
}
|
|
|
|
// keptByModule is the resources of the declaration the node kept before the switch, by the module they say; the node's
|
|
// own are those that say none (ADR 0539 §9).
|
|
func keptByModule(opts options, signer ed25519.PublicKey) func() map[string][]json.RawMessage {
|
|
var once sync.Once
|
|
var by map[string][]json.RawMessage
|
|
return func() map[string][]json.RawMessage {
|
|
once.Do(func() {
|
|
raw, err := store.LoadDeclared(store.DeclaredPath(opts.state), signer)
|
|
if err != nil {
|
|
return
|
|
}
|
|
var kept struct {
|
|
Resources []json.RawMessage `json:"resources"`
|
|
}
|
|
if json.Unmarshal(raw, &kept) != nil {
|
|
return
|
|
}
|
|
by = map[string][]json.RawMessage{}
|
|
for _, r := range kept.Resources {
|
|
var head struct {
|
|
Module string `json:"module"`
|
|
}
|
|
_ = json.Unmarshal(r, &head)
|
|
name := head.Module
|
|
if name == "" {
|
|
name = agent.NodeModule
|
|
}
|
|
by[name] = append(by[name], r)
|
|
}
|
|
})
|
|
return by
|
|
}
|
|
}
|
|
|
|
// nodeAgent runs the node agent under the session's links for as long as ctx lasts: it waits for a link, follows the
|
|
// node's wants and compositions on it, and begins again with the next link. A watch it cannot make (a bucket this node's
|
|
// account may not read) is said, and tried again with a growing wait, as the link is.
|
|
func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, session *link.Session, sched *apply.Scheduler,
|
|
say link.Announce, afterApply func(), current *agentNow, sayFacts func(ctx context.Context, ag *agent.Agent)) {
|
|
wait := 2 * time.Second
|
|
for ctx.Err() == nil {
|
|
linked, cancel, err := session.WhenLinked(ctx)
|
|
if err != nil {
|
|
return
|
|
}
|
|
began := time.Now()
|
|
node := session.Membership.Node
|
|
composedBucket, err := delivery.ComposedBucketFor(node)
|
|
if err != nil {
|
|
cancel()
|
|
say("the node agent cannot name this node's composition bucket: " + err.Error())
|
|
return
|
|
}
|
|
var ag *agent.Agent
|
|
env := agent.Env{
|
|
Node: node,
|
|
Wants: bucketWatch{session, delivery.AssignerWantsBucket},
|
|
Composed: bucketWatch{session, composedBucket},
|
|
Haves: sessionHaves{session},
|
|
Kept: keptByModule(opts, signer),
|
|
Recorded: recordedIn(opts.state, say),
|
|
Trusted: trustedKeys(opts, signer),
|
|
Backup: backupNode(session),
|
|
Say: say,
|
|
Apply: func(ctx context.Context, c delivery.Composed, others []delivery.Composed) agent.Result {
|
|
actx, done := applyContext(session.WithLink(ctx))
|
|
res := applyComposition(actx, opts, c, others, sched, say)
|
|
done()
|
|
afterApply()
|
|
sayFacts(ctx, ag)
|
|
return res
|
|
},
|
|
Remove: func(ctx context.Context, module string, others []delivery.Composed) agent.Result {
|
|
actx, done := applyContext(session.WithLink(ctx))
|
|
res := removeComposition(actx, opts, module, others, sched, say)
|
|
done()
|
|
afterApply()
|
|
sayFacts(ctx, ag)
|
|
return res
|
|
},
|
|
}
|
|
ag = agent.New(env)
|
|
current.set(ag)
|
|
err = ag.Run(linked)
|
|
cancel()
|
|
current.set(nil)
|
|
if ctx.Err() != nil {
|
|
return
|
|
}
|
|
if time.Since(began) > 30*time.Second {
|
|
wait = 2 * time.Second
|
|
}
|
|
if err != nil {
|
|
say(fmt.Sprintf("the node agent stopped: %v — beginning again in %s", err, wait))
|
|
}
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-time.After(wait):
|
|
}
|
|
if wait *= 2; wait > 2*time.Minute {
|
|
wait = 2 * time.Minute
|
|
}
|
|
}
|
|
}
|
|
|
|
// agentNow is the agent running now, which the node's facts are built from.
|
|
type agentNow struct {
|
|
mu sync.Mutex
|
|
ag *agent.Agent
|
|
}
|
|
|
|
func (c *agentNow) set(a *agent.Agent) {
|
|
c.mu.Lock()
|
|
c.ag = a
|
|
c.mu.Unlock()
|
|
}
|
|
|
|
func (c *agentNow) get() *agent.Agent {
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
return c.ag
|
|
}
|
|
|
|
// applyComposition applies one composition: a module's, or the node's own.
|
|
func applyComposition(ctx context.Context, opts options, c delivery.Composed, others []delivery.Composed,
|
|
sched *apply.Scheduler, say link.Announce) agent.Result {
|
|
name := scopeName(c)
|
|
d, err := declarationOf(c)
|
|
if err != nil {
|
|
return agent.Result{Failed: true, Detail: shorten(err.Error())}
|
|
}
|
|
if err := refuseEmptyWrites(d); err != nil {
|
|
return agent.Result{Failed: true, Detail: shorten(err.Error())}
|
|
}
|
|
scope := apply.Scope{Module: name, Others: resourcesOf(others)}
|
|
got := applyModuleHeld(ctx, opts, d, &scope, sched, say)
|
|
if got.Refused != "" {
|
|
return agent.Result{Failed: true, Detail: shorten(got.Refused)}
|
|
}
|
|
results := apply.ModuleResults(d, got.Outcome, got.Err)
|
|
if r, ok := results[name]; ok && r.Failed {
|
|
return agent.Result{Failed: true, Detail: shorten(r.Detail)}
|
|
}
|
|
if got.Err != nil {
|
|
return agent.Result{Failed: true, Detail: shorten(got.Err.Error())}
|
|
}
|
|
if got.Probe != nil {
|
|
return agent.Result{Failed: true, Detail: shorten(got.Probe.partial().Error())}
|
|
}
|
|
return agent.Result{}
|
|
}
|
|
|
|
// removeComposition removes one module's own resources and nothing of another's.
|
|
func removeComposition(ctx context.Context, opts options, module string, others []delivery.Composed,
|
|
sched *apply.Scheduler, say link.Announce) agent.Result {
|
|
scope := apply.Scope{Module: module, Others: resourcesOf(others)}
|
|
got := applyModuleHeld(ctx, opts, &declaration.Declaration{Version: declaration.Version}, &scope, sched, say)
|
|
switch {
|
|
case got.Refused != "":
|
|
return agent.Result{Failed: true, Detail: shorten(got.Refused)}
|
|
case got.Err != nil:
|
|
return agent.Result{Failed: true, Detail: shorten(got.Err.Error())}
|
|
}
|
|
return agent.Result{}
|
|
}
|
|
|
|
// busyModules are the modules whose last apply left a container running because it said busy.
|
|
func busyModules() []string {
|
|
busyMu.Lock()
|
|
defer busyMu.Unlock()
|
|
var out []string
|
|
for m, w := range busyBy {
|
|
if w && m != "" {
|
|
out = append(out, m)
|
|
}
|
|
}
|
|
sort.Strings(out)
|
|
return out
|
|
}
|
|
|
|
// keptDeclaration is the declaration the node kept before the switch, verified with the mesh's key when it is given.
|
|
func keptDeclaration(statePath string, signer ed25519.PublicKey) (*declaration.Declaration, error) {
|
|
var raw []byte
|
|
if len(signer) > 0 {
|
|
var err error
|
|
if raw, err = store.LoadDeclared(store.DeclaredPath(statePath), signer); err != nil {
|
|
return nil, err
|
|
}
|
|
} else {
|
|
kept, err := store.ReadDeclared(store.DeclaredPath(statePath))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
raw = kept.Declaration
|
|
}
|
|
return declaration.Parse(raw)
|
|
}
|
|
|
|
// whatRuns is what runs on this node as the declaration of it (novox/hq ADR 0539 §9): the compositions the agent runs
|
|
// and, for every module the agent does not hold yet, what the kept declaration has for it, with the node's own parts
|
|
// (adoption, identities, groups, libraries) from the node's composition once the agent holds it and from the kept
|
|
// declaration until then. What the node judges its health by, and what the filling, the take act and the judgement of
|
|
// directories read.
|
|
func whatRuns(everything []delivery.Composed, holdsNode bool, kept func() (*declaration.Declaration, error)) (*declaration.Declaration, error) {
|
|
d, err := declarationOf(everything...)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if holdsNode {
|
|
return d, nil
|
|
}
|
|
old, err := kept()
|
|
if err != nil {
|
|
if len(everything) > 0 {
|
|
return d, nil
|
|
}
|
|
return nil, err
|
|
}
|
|
d.Adoption, d.Identities, d.Groups, d.Libraries = old.Adoption, old.Identities, old.Groups, old.Libraries
|
|
d.Provisioner, d.Builds = old.Provisioner, old.Builds
|
|
d.Filled = append(d.Filled, old.Filled...)
|
|
return d, nil
|
|
}
|
|
|
|
// runsHere says what runs on this node; the running serve sets it to the agent's view, and a command run by hand reads
|
|
// the kept declaration.
|
|
var runsHere struct {
|
|
mu sync.Mutex
|
|
fn func(statePath string, signer ed25519.PublicKey) (*declaration.Declaration, error)
|
|
}
|
|
|
|
func setRunsHere(fn func(statePath string, signer ed25519.PublicKey) (*declaration.Declaration, error)) {
|
|
runsHere.mu.Lock()
|
|
runsHere.fn = fn
|
|
runsHere.mu.Unlock()
|
|
}
|
|
|
|
// declaredNow is the declaration of what runs on this node now.
|
|
func declaredNow(statePath string, signer ed25519.PublicKey) (*declaration.Declaration, error) {
|
|
runsHere.mu.Lock()
|
|
fn := runsHere.fn
|
|
runsHere.mu.Unlock()
|
|
if fn != nil {
|
|
return fn(statePath, signer)
|
|
}
|
|
return keptDeclaration(statePath, signer)
|
|
}
|
|
|
|
// runsFromAgent is what runs now as the agent in hand sees it.
|
|
func runsFromAgent(current *agentNow) func(string, ed25519.PublicKey) (*declaration.Declaration, error) {
|
|
return func(statePath string, signer ed25519.PublicKey) (*declaration.Declaration, error) {
|
|
kept := func() (*declaration.Declaration, error) { return keptDeclaration(statePath, signer) }
|
|
ag := current.get()
|
|
if ag == nil {
|
|
return kept()
|
|
}
|
|
return whatRuns(ag.Everything(), ag.HoldsNode(), kept)
|
|
}
|
|
}
|
|
|
|
// refuseEmptyWrites refuses a declaration that names a file the provisioner fills (owned at runtime, no content, no
|
|
// bytes, no digest, no sealed value) without saying it is filled: applied, the file would be written empty over what the
|
|
// provisioner wrote (issue 1945, live on 2026-10-25 at 01:22 on ace, g14 and shanks). The apply fails, naming the file,
|
|
// and nothing is written.
|
|
func refuseEmptyWrites(d *declaration.Declaration) error {
|
|
filled := make(map[string]bool, len(d.Filled))
|
|
for _, p := range d.Filled {
|
|
filled[p] = true
|
|
}
|
|
var refused []string
|
|
for _, r := range d.Resources {
|
|
f, ok := r.(*declaration.File)
|
|
if !ok || f.OwnerAtRuntime != declaration.RuntimeOwned || filled[f.Path] {
|
|
continue
|
|
}
|
|
if f.Content == "" && f.Bytes == "" && f.Digest == "" && f.Sealed == "" && f.VaultSealed == nil {
|
|
refused = append(refused, fmt.Sprintf("%s (%s)", f.ID, f.Path))
|
|
}
|
|
}
|
|
if len(refused) > 0 {
|
|
return fmt.Errorf("the composition names %s with no content and does not say the provisioner fills it, so it "+
|
|
"would be written empty; nothing was applied", strings.Join(refused, ", "))
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// recordedIn says the engine's applied state holds a record of every resource of a composition: what the agent believes
|
|
// of a have or the kept declaration only when it does, so a resource removed since is made again (issue 1945). A state
|
|
// that cannot be read says yes: the agent then believes what it was told, as before.
|
|
func recordedIn(statePath string, say func(string)) func(ctx context.Context, c delivery.Composed) bool {
|
|
return func(_ context.Context, c delivery.Composed) bool {
|
|
known, err := store.Load(statePath)
|
|
if err != nil {
|
|
if say != nil {
|
|
say(fmt.Sprintf("the applied state cannot be read, so what runs is believed as told: %v", err))
|
|
}
|
|
return true
|
|
}
|
|
filled := make(map[string]bool, len(c.Filled))
|
|
for _, p := range c.Filled {
|
|
filled[p] = true
|
|
}
|
|
for _, raw := range c.Resources {
|
|
r, err := declaration.ParseResource(raw)
|
|
if err != nil {
|
|
return true
|
|
}
|
|
// A file the provisioner client fills is left to it, and has no record until it writes it.
|
|
if f, ok := r.(*declaration.File); ok && filled[f.Path] {
|
|
continue
|
|
}
|
|
if _, ok := known.Find(r.Identity()); !ok {
|
|
return false
|
|
}
|
|
}
|
|
return true
|
|
}
|
|
}
|
|
|
|
// declaresNothingToBackUp reads node-backup.backed-up's answer for a module: true when this machine backs nothing of it
|
|
// up (no runs, no paths), and whether the answer could be read at all.
|
|
func declaresNothingToBackUp(said json.RawMessage, module string) (nothing bool, read bool) {
|
|
var list []struct {
|
|
Module string `json:"module"`
|
|
Runs int `json:"runs"`
|
|
Paths []string `json:"paths"`
|
|
}
|
|
if err := json.Unmarshal(said, &list); err != nil {
|
|
return false, false
|
|
}
|
|
for _, m := range list {
|
|
if m.Module == module && (m.Runs > 0 || len(m.Paths) > 0) {
|
|
return false, true
|
|
}
|
|
}
|
|
return true, true
|
|
}
|
|
|
|
// applyBound is how long one apply or removal may take: long enough for the slowest image pull and backup step, short
|
|
// enough that a step that hangs cannot hold the node agent for ever.
|
|
const applyBound = 30 * time.Minute
|
|
|
|
// applyContext is the context an apply or a removal runs in: the link's values, never its end, and a bound of its own.
|
|
// The link's context ends when the bus goes, and the bus may go because of the apply itself (it replaces the bus's own
|
|
// container), which must then finish: a container removed and never made again took the mesh down (issue 1945, live on
|
|
// 2026-10-25 at 11:23). applyBound keeps a hung step from blocking the agent for ever.
|
|
func applyContext(ctx context.Context) (context.Context, context.CancelFunc) {
|
|
return context.WithTimeout(context.WithoutCancel(ctx), applyBound)
|
|
}
|