diff --git a/spindle/config/config.go b/spindle/config/config.go index 3bae65d1..661e35f7 100644 --- a/spindle/config/config.go +++ b/spindle/config/config.go @@ -62,7 +62,7 @@ type S3 struct { } type MicroVMPipelines struct { - ImageDir string `env:"IMAGE_DIR, required"` + ImageDir string `env:"IMAGE_DIR"` OverlayDir string `env:"OVERLAY_DIR, default="` // where microVM temporary disks will live DefaultImage string `env:"DEFAULT_IMAGE, default=nixos-x86_64"` AgentPort uint32 `env:"AGENT_PORT, default=10240"` diff --git a/spindle/config/config_test.go b/spindle/config/config_test.go new file mode 100644 index 00000000..12a42e39 --- /dev/null +++ b/spindle/config/config_test.go @@ -0,0 +1,20 @@ +package config + +import ( + "context" + "testing" +) + +func TestLoadAllowsUnconfiguredMicroVMEngine(t *testing.T) { + t.Setenv("SPINDLE_SERVER_HOSTNAME", "spindle.example.com") + t.Setenv("SPINDLE_SERVER_OWNER", "did:web:spindle.example.com") + t.Setenv("SPINDLE_MICROVM_PIPELINES_IMAGE_DIR", "") + + cfg, err := Load(context.Background()) + if err != nil { + t.Fatal(err) + } + if cfg.MicroVMPipelines.ImageDir != "" { + t.Fatalf("image directory = %q, want empty", cfg.MicroVMPipelines.ImageDir) + } +} diff --git a/spindle/engines/microvm/engine.go b/spindle/engines/microvm/engine.go index 8574f0f9..469e17f7 100644 --- a/spindle/engines/microvm/engine.go +++ b/spindle/engines/microvm/engine.go @@ -46,6 +46,7 @@ type Engine struct { l *slog.Logger cfg *config.Config db *db.DB + agentMu sync.Mutex agent *agentHub scheduler *engine.ResourceScheduler[Resources] cgroupParent *CgroupParent @@ -70,18 +71,11 @@ func (s Step) Kind() models.StepKind { return s.kind } func New(ctx context.Context, cfg *config.Config, d *db.DB) (*Engine, error) { l := log.FromContext(ctx).With("component", "engine.microvm") - port := cfg.MicroVMPipelines.AgentPort - if port == 0 { - port = agentproto.DefaultPort - } - agent, err := newAgentHub(port, l) - if err != nil { - return nil, err - } budget, max, agingThreshold := newVMBudgetConfig(cfg.MicroVMPipelines) l.Info("initialized microVM workflow budget", "budget", budget.String(), "maxWorkflow", max.String(), "agingThreshold", agingThreshold) var cgroupParent *CgroupParent + var err error if cfg.MicroVMPipelines.EnableCgroups { cgroupParent, err = initCgroupParent(cfg.MicroVMPipelines.CgroupParent, cfg.MicroVMPipelines.CgroupSupervisorMemoryMinMiB, l) if err != nil { @@ -93,13 +87,32 @@ func New(ctx context.Context, cfg *config.Config, d *db.DB) (*Engine, error) { l: l, cfg: cfg, db: d, - agent: agent, scheduler: engine.NewResourceScheduler(budget, max, agingThreshold), cgroupParent: cgroupParent, cleanup: make(map[string][]cleanupFunc), }, nil } +func (e *Engine) ensureAgentHub() (*agentHub, error) { + e.agentMu.Lock() + defer e.agentMu.Unlock() + + if e.agent != nil { + return e.agent, nil + } + + port := e.cfg.MicroVMPipelines.AgentPort + if port == 0 { + port = agentproto.DefaultPort + } + agent, err := newAgentHub(port, e.l) + if err != nil { + return nil, err + } + e.agent = agent + return agent, nil +} + func (e *Engine) InitWorkflow(twf tangled.Pipeline_Workflow, tpl tangled.Pipeline) (*models.Workflow, error) { swf := &models.Workflow{} var dwf manifestWorkflow @@ -210,7 +223,11 @@ func (e *Engine) SetupWorkflow(ctx context.Context, wid models.WorkflowId, wf *m if err != nil { return err } - connCh, unregister, err := e.agent.expect(cid) + agent, err := e.ensureAgentHub() + if err != nil { + return err + } + connCh, unregister, err := agent.expect(cid) if err != nil { return err } diff --git a/spindle/engines/microvm/engine_test.go b/spindle/engines/microvm/engine_test.go index 0585121c..13465cd5 100644 --- a/spindle/engines/microvm/engine_test.go +++ b/spindle/engines/microvm/engine_test.go @@ -1,6 +1,7 @@ package microvm import ( + "context" "encoding/json" "log/slog" "os" @@ -36,6 +37,16 @@ func testEngine(t *testing.T, imageDir string) *Engine { } } +func TestNewDefersAgentHubUntilWorkflowSetup(t *testing.T) { + e, err := New(context.Background(), &config.Config{}, nil) + if err != nil { + t.Fatal(err) + } + if e.agent != nil { + t.Fatal("agent hub started during engine initialization") + } +} + func TestInitWorkflowRejectsConfigOnNonNixOSImage(t *testing.T) { dir := t.TempDir() writeTestImageSpec(t, dir, "alpine", validImageSpec()) diff --git a/spindle/engines/microvm/image.go b/spindle/engines/microvm/image.go index 89db1552..efe3b6be 100644 --- a/spindle/engines/microvm/image.go +++ b/spindle/engines/microvm/image.go @@ -181,7 +181,12 @@ func (e *Engine) resolveImage(name string) (ImageSpec, string, string, error) { return ImageSpec{}, "", "", fmt.Errorf("invalid microVM image name %q: must be a plain name, not a path", name) } - candidates := imageCandidates(e.cfg.MicroVMPipelines.ImageDir, name) + imageDir := strings.TrimSpace(e.cfg.MicroVMPipelines.ImageDir) + if imageDir == "" { + return ImageSpec{}, "", "", fmt.Errorf("microVM workflows require SPINDLE_MICROVM_PIPELINES_IMAGE_DIR") + } + + candidates := imageCandidates(imageDir, name) for _, candidate := range candidates { path, ok, err := imageSpecPath(candidate) if err != nil { diff --git a/spindle/engines/microvm/image_test.go b/spindle/engines/microvm/image_test.go index 3000005c..3c5f66f0 100644 --- a/spindle/engines/microvm/image_test.go +++ b/spindle/engines/microvm/image_test.go @@ -77,6 +77,14 @@ func TestResolveImageDirectoryMissingSpec(t *testing.T) { } } +func TestResolveImageRequiresImageDirOnlyWhenUsed(t *testing.T) { + e := testEngine(t, "") + _, _, _, err := e.resolveImage("nixos") + if err == nil || !strings.Contains(err.Error(), "SPINDLE_MICROVM_PIPELINES_IMAGE_DIR") { + t.Fatalf("missing image directory should error when resolving an image, got: %v", err) + } +} + func TestResolveImageRejectsPaths(t *testing.T) { e := testEngine(t, t.TempDir()) for _, name := range []string{"/etc/passwd", "../evil", "sub/evil", "..", "."} {