Monorepo for Tangled
spindle/{config,engines/microvm}: make engine fully lazy
Signed-off-by: dawn <dawn@tangled.org>
+73
-12
+1
-1
spindle/config/config.go
+1
-1
spindle/config/config.go
···
62
62
}
63
63
64
64
type MicroVMPipelines struct {
65
-
ImageDir string `env:"IMAGE_DIR, required"`
65
+
ImageDir string `env:"IMAGE_DIR"`
66
66
OverlayDir string `env:"OVERLAY_DIR, default="` // where microVM temporary disks will live
67
67
DefaultImage string `env:"DEFAULT_IMAGE, default=nixos-x86_64"`
68
68
AgentPort uint32 `env:"AGENT_PORT, default=10240"`
+20
spindle/config/config_test.go
+20
spindle/config/config_test.go
···
1
+
package config
2
+
3
+
import (
4
+
"context"
5
+
"testing"
6
+
)
7
+
8
+
func TestLoadAllowsUnconfiguredMicroVMEngine(t *testing.T) {
9
+
t.Setenv("SPINDLE_SERVER_HOSTNAME", "spindle.example.com")
10
+
t.Setenv("SPINDLE_SERVER_OWNER", "did:web:spindle.example.com")
11
+
t.Setenv("SPINDLE_MICROVM_PIPELINES_IMAGE_DIR", "")
12
+
13
+
cfg, err := Load(context.Background())
14
+
if err != nil {
15
+
t.Fatal(err)
16
+
}
17
+
if cfg.MicroVMPipelines.ImageDir != "" {
18
+
t.Fatalf("image directory = %q, want empty", cfg.MicroVMPipelines.ImageDir)
19
+
}
20
+
}
+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
+6
-1
spindle/engines/microvm/image.go
+6
-1
spindle/engines/microvm/image.go
···
181
181
return ImageSpec{}, "", "", fmt.Errorf("invalid microVM image name %q: must be a plain name, not a path", name)
182
182
}
183
183
184
-
candidates := imageCandidates(e.cfg.MicroVMPipelines.ImageDir, name)
184
+
imageDir := strings.TrimSpace(e.cfg.MicroVMPipelines.ImageDir)
185
+
if imageDir == "" {
186
+
return ImageSpec{}, "", "", fmt.Errorf("microVM workflows require SPINDLE_MICROVM_PIPELINES_IMAGE_DIR")
187
+
}
188
+
189
+
candidates := imageCandidates(imageDir, name)
185
190
for _, candidate := range candidates {
186
191
path, ok, err := imageSpecPath(candidate)
187
192
if err != nil {
+8
spindle/engines/microvm/image_test.go
+8
spindle/engines/microvm/image_test.go
···
77
77
}
78
78
}
79
79
80
+
func TestResolveImageRequiresImageDirOnlyWhenUsed(t *testing.T) {
81
+
e := testEngine(t, "")
82
+
_, _, _, err := e.resolveImage("nixos")
83
+
if err == nil || !strings.Contains(err.Error(), "SPINDLE_MICROVM_PIPELINES_IMAGE_DIR") {
84
+
t.Fatalf("missing image directory should error when resolving an image, got: %v", err)
85
+
}
86
+
}
87
+
80
88
func TestResolveImageRejectsPaths(t *testing.T) {
81
89
e := testEngine(t, t.TempDir())
82
90
for _, name := range []string{"/etc/passwd", "../evil", "sub/evil", "..", "."} {