alice.pds.demo.boltless.dev
This repository has no description
spindle/engines/microvm: init agent hub lazily
Signed-off-by: dawn <dawn@tangled.org>
author
dawn
date
(Jul 17, 2026, 11:53 AM +0300)
commit
769505a5
769505a5800131172e39a33c1d37694479f3c030
parent
c5406c96
c5406c96b38abf3be5e4e515124d7492ede2e8ce
change-id
nowsytzv
nowsytzvpoktxzvzsvmvsptwonzmqvxt
+38
-10
+27
-10
spindle/engines/microvm/engine.go
+27
-10
spindle/engines/microvm/engine.go
···
46
46
l *slog.Logger
47
47
cfg *config.Config
48
48
db *db.DB
49
+
agentMu sync.Mutex
49
50
agent *agentHub
50
51
scheduler *engine.ResourceScheduler[Resources]
51
52
cgroupParent *CgroupParent
···
70
71
71
72
func New(ctx context.Context, cfg *config.Config, d *db.DB) (*Engine, error) {
72
73
l := log.FromContext(ctx).With("component", "engine.microvm")
73
-
port := cfg.MicroVMPipelines.AgentPort
74
-
if port == 0 {
75
-
port = agentproto.DefaultPort
76
-
}
77
-
agent, err := newAgentHub(port, l)
78
-
if err != nil {
79
-
return nil, err
80
-
}
81
74
budget, max, agingThreshold := newVMBudgetConfig(cfg.MicroVMPipelines)
82
75
l.Info("initialized microVM workflow budget", "budget", budget.String(), "maxWorkflow", max.String(), "agingThreshold", agingThreshold)
83
76
84
77
var cgroupParent *CgroupParent
78
+
var err error
85
79
if cfg.MicroVMPipelines.EnableCgroups {
86
80
cgroupParent, err = initCgroupParent(cfg.MicroVMPipelines.CgroupParent, cfg.MicroVMPipelines.CgroupSupervisorMemoryMinMiB, l)
87
81
if err != nil {
···
93
87
l: l,
94
88
cfg: cfg,
95
89
db: d,
96
-
agent: agent,
97
90
scheduler: engine.NewResourceScheduler(budget, max, agingThreshold),
98
91
cgroupParent: cgroupParent,
99
92
cleanup: make(map[string][]cleanupFunc),
100
93
}, nil
94
+
}
95
+
96
+
func (e *Engine) ensureAgentHub() (*agentHub, error) {
97
+
e.agentMu.Lock()
98
+
defer e.agentMu.Unlock()
99
+
100
+
if e.agent != nil {
101
+
return e.agent, nil
102
+
}
103
+
104
+
port := e.cfg.MicroVMPipelines.AgentPort
105
+
if port == 0 {
106
+
port = agentproto.DefaultPort
107
+
}
108
+
agent, err := newAgentHub(port, e.l)
109
+
if err != nil {
110
+
return nil, err
111
+
}
112
+
e.agent = agent
113
+
return agent, nil
101
114
}
102
115
103
116
func (e *Engine) InitWorkflow(twf tangled.Pipeline_Workflow, tpl tangled.Pipeline) (*models.Workflow, error) {
···
210
223
if err != nil {
211
224
return err
212
225
}
213
-
connCh, unregister, err := e.agent.expect(cid)
226
+
agent, err := e.ensureAgentHub()
227
+
if err != nil {
228
+
return err
229
+
}
230
+
connCh, unregister, err := agent.expect(cid)
214
231
if err != nil {
215
232
return err
216
233
}
+11
spindle/engines/microvm/engine_test.go
+11
spindle/engines/microvm/engine_test.go
···
1
1
package microvm
2
2
3
3
import (
4
+
"context"
4
5
"encoding/json"
5
6
"log/slog"
6
7
"os"
···
33
34
DefaultImage: "alpine",
34
35
},
35
36
},
37
+
}
38
+
}
39
+
40
+
func TestNewDefersAgentHubUntilWorkflowSetup(t *testing.T) {
41
+
e, err := New(context.Background(), &config.Config{}, nil)
42
+
if err != nil {
43
+
t.Fatal(err)
44
+
}
45
+
if e.agent != nil {
46
+
t.Fatal("agent hub started during engine initialization")
36
47
}
37
48
}
38
49