Monorepo for Tangled
spindle/mill: place workflows by engine capabilities
+764
-83
+52
-51
shuttle/src/gen/spindle/agent/v1/spindle.agent.v1.rs
+52
-51
shuttle/src/gen/spindle/agent/v1/spindle.agent.v1.rs
···
2
2
// This file is @generated by prost-build.
3
3
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
4
4
pub struct Hello {
5
-
#[prost(uint32, tag = "1")]
5
+
#[prost(uint32, tag="1")]
6
6
pub protocol_version: u32,
7
-
#[prost(string, tag = "2")]
7
+
#[prost(string, tag="2")]
8
8
pub agent_version: ::prost::alloc::string::String,
9
-
#[prost(string, tag = "3")]
9
+
#[prost(string, tag="3")]
10
10
pub boot_id: ::prost::alloc::string::String,
11
-
#[prost(string, tag = "4")]
11
+
#[prost(string, tag="4")]
12
12
pub nix_version: ::prost::alloc::string::String,
13
13
}
14
14
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
15
15
pub struct Init {
16
-
#[prost(string, tag = "1")]
16
+
#[prost(string, tag="1")]
17
17
pub job_id: ::prost::alloc::string::String,
18
-
#[prost(string, repeated, tag = "2")]
18
+
#[prost(string, repeated, tag="2")]
19
19
pub cache_trusted_public_keys: ::prost::alloc::vec::Vec<::prost::alloc::string::String>,
20
-
#[prost(uint32, tag = "3")]
20
+
#[prost(uint32, tag="3")]
21
21
pub cache_read_proxy_port: u32,
22
-
#[prost(uint32, tag = "4")]
22
+
#[prost(uint32, tag="4")]
23
23
pub cache_upload_proxy_port: u32,
24
-
#[prost(uint32, tag = "5")]
24
+
#[prost(uint32, tag="5")]
25
25
pub dns_proxy_port: u32,
26
26
}
27
27
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
28
28
pub struct ExecStart {
29
-
#[prost(string, repeated, tag = "1")]
29
+
#[prost(string, repeated, tag="1")]
30
30
pub argv: ::prost::alloc::vec::Vec<::prost::alloc::string::String>,
31
-
#[prost(string, repeated, tag = "2")]
31
+
#[prost(string, repeated, tag="2")]
32
32
pub env: ::prost::alloc::vec::Vec<::prost::alloc::string::String>,
33
-
#[prost(string, tag = "3")]
33
+
#[prost(string, tag="3")]
34
34
pub cwd: ::prost::alloc::string::String,
35
-
#[prost(string, tag = "4")]
35
+
#[prost(string, tag="4")]
36
36
pub user: ::prost::alloc::string::String,
37
-
#[prost(uint32, tag = "5")]
37
+
#[prost(uint32, tag="5")]
38
38
pub timeout_seconds: u32,
39
39
}
40
40
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
41
41
pub struct ExecStdout {
42
-
#[prost(string, tag = "1")]
42
+
#[prost(string, tag="1")]
43
43
pub data: ::prost::alloc::string::String,
44
44
}
45
45
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
46
46
pub struct ExecStderr {
47
-
#[prost(string, tag = "1")]
47
+
#[prost(string, tag="1")]
48
48
pub data: ::prost::alloc::string::String,
49
49
}
50
50
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
51
51
pub struct ExecExit {
52
-
#[prost(int32, tag = "1")]
52
+
#[prost(int32, tag="1")]
53
53
pub exit_code: i32,
54
-
#[prost(string, tag = "2")]
54
+
#[prost(string, tag="2")]
55
55
pub error: ::prost::alloc::string::String,
56
56
/// set when the guest killed the step on its own timeout timer, so the host
57
57
/// can classify it as a timeout rather than inferring failure from exit_code.
58
-
#[prost(bool, tag = "3")]
58
+
#[prost(bool, tag="3")]
59
59
pub timed_out: bool,
60
60
}
61
61
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
62
62
pub struct ActivateConfig {
63
-
#[prost(string, tag = "1")]
63
+
#[prost(string, tag="1")]
64
64
pub config_key: ::prost::alloc::string::String,
65
-
#[prost(string, tag = "2")]
65
+
#[prost(string, tag="2")]
66
66
pub base_config_hash: ::prost::alloc::string::String,
67
-
#[prost(string, tag = "3")]
67
+
#[prost(string, tag="3")]
68
68
pub user_config: ::prost::alloc::string::String,
69
-
#[prost(string, tag = "4")]
69
+
#[prost(string, tag="4")]
70
70
pub toplevel: ::prost::alloc::string::String,
71
-
#[prost(uint32, tag = "5")]
71
+
#[prost(uint32, tag="5")]
72
72
pub timeout_seconds: u32,
73
73
}
74
74
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
75
75
pub struct ActivateConfigResult {
76
-
#[prost(string, tag = "1")]
76
+
#[prost(string, tag="1")]
77
77
pub config_key: ::prost::alloc::string::String,
78
-
#[prost(string, tag = "2")]
78
+
#[prost(string, tag="2")]
79
79
pub toplevel: ::prost::alloc::string::String,
80
-
#[prost(string, tag = "3")]
80
+
#[prost(string, tag="3")]
81
81
pub error: ::prost::alloc::string::String,
82
82
}
83
83
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
84
84
pub struct BuiltPaths {
85
-
#[prost(string, repeated, tag = "1")]
85
+
#[prost(string, repeated, tag="1")]
86
86
pub paths: ::prost::alloc::vec::Vec<::prost::alloc::string::String>,
87
-
#[prost(string, tag = "2")]
87
+
#[prost(string, tag="2")]
88
88
pub reason: ::prost::alloc::string::String,
89
89
}
90
90
#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
91
91
pub struct CacheDrain {
92
-
#[prost(uint32, tag = "1")]
92
+
#[prost(uint32, tag="1")]
93
93
pub timeout_seconds: u32,
94
94
}
95
95
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
96
96
pub struct CacheDrainResult {
97
-
#[prost(string, tag = "1")]
97
+
#[prost(string, tag="1")]
98
98
pub error: ::prost::alloc::string::String,
99
-
#[prost(uint32, tag = "2")]
99
+
#[prost(uint32, tag="2")]
100
100
pub cache_queued: u32,
101
-
#[prost(uint32, tag = "3")]
101
+
#[prost(uint32, tag="3")]
102
102
pub cache_active: u32,
103
-
#[prost(uint32, tag = "4")]
103
+
#[prost(uint32, tag="4")]
104
104
pub cache_uploaded: u32,
105
-
#[prost(uint32, tag = "5")]
105
+
#[prost(uint32, tag="5")]
106
106
pub cache_failed: u32,
107
107
}
108
108
#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
109
-
pub struct Poweroff {}
109
+
pub struct Poweroff {
110
+
}
110
111
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
111
112
pub struct PoweroffResult {
112
-
#[prost(string, tag = "1")]
113
+
#[prost(string, tag="1")]
113
114
pub error: ::prost::alloc::string::String,
114
115
}
115
116
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
116
117
pub struct Message {
117
-
#[prost(string, tag = "1")]
118
+
#[prost(string, tag="1")]
118
119
pub id: ::prost::alloc::string::String,
119
-
#[prost(message, optional, tag = "2")]
120
+
#[prost(message, optional, tag="2")]
120
121
pub hello: ::core::option::Option<Hello>,
121
-
#[prost(message, optional, tag = "3")]
122
+
#[prost(message, optional, tag="3")]
122
123
pub init: ::core::option::Option<Init>,
123
-
#[prost(message, optional, tag = "4")]
124
+
#[prost(message, optional, tag="4")]
124
125
pub exec_start: ::core::option::Option<ExecStart>,
125
-
#[prost(message, optional, tag = "5")]
126
+
#[prost(message, optional, tag="5")]
126
127
pub exec_stdout: ::core::option::Option<ExecStdout>,
127
-
#[prost(message, optional, tag = "6")]
128
+
#[prost(message, optional, tag="6")]
128
129
pub exec_stderr: ::core::option::Option<ExecStderr>,
129
-
#[prost(message, optional, tag = "7")]
130
+
#[prost(message, optional, tag="7")]
130
131
pub exec_exit: ::core::option::Option<ExecExit>,
131
-
#[prost(message, optional, tag = "8")]
132
+
#[prost(message, optional, tag="8")]
132
133
pub activate_config: ::core::option::Option<ActivateConfig>,
133
-
#[prost(message, optional, tag = "9")]
134
+
#[prost(message, optional, tag="9")]
134
135
pub activate_config_result: ::core::option::Option<ActivateConfigResult>,
135
-
#[prost(message, optional, tag = "10")]
136
+
#[prost(message, optional, tag="10")]
136
137
pub built_paths: ::core::option::Option<BuiltPaths>,
137
-
#[prost(message, optional, tag = "11")]
138
+
#[prost(message, optional, tag="11")]
138
139
pub cache_drain: ::core::option::Option<CacheDrain>,
139
-
#[prost(message, optional, tag = "12")]
140
+
#[prost(message, optional, tag="12")]
140
141
pub cache_drain_result: ::core::option::Option<CacheDrainResult>,
141
-
#[prost(message, optional, tag = "13")]
142
+
#[prost(message, optional, tag="13")]
142
143
pub poweroff: ::core::option::Option<Poweroff>,
143
-
#[prost(message, optional, tag = "14")]
144
+
#[prost(message, optional, tag="14")]
144
145
pub poweroff_result: ::core::option::Option<PoweroffResult>,
145
146
}
146
147
// @@protoc_insertion_point(module)
+49
spindle/engine/placement.go
+49
spindle/engine/placement.go
···
1
+
package engine
2
+
3
+
import (
4
+
"fmt"
5
+
"slices"
6
+
"strings"
7
+
"unicode"
8
+
9
+
"tangled.org/core/api/tangled"
10
+
"tangled.org/core/spindle/models"
11
+
)
12
+
13
+
// PlacementPlanner derives the engine capabilities a workflow needs. Planners
14
+
// run in the trusted mill process and must not inspect executor-local state.
15
+
type PlacementPlanner interface {
16
+
Requirements(twf tangled.Pipeline_Workflow, tpl tangled.Pipeline) ([]string, error)
17
+
}
18
+
19
+
// CapabilityProvider reports the cached placement capabilities of one local
20
+
// engine instance. Capabilities are opaque to the mill.
21
+
type CapabilityProvider interface {
22
+
Capabilities() []string
23
+
}
24
+
25
+
// WorkflowPlacementValidator performs the definitive immutable compatibility
26
+
// checks before an executor accepts and holds a remote lease.
27
+
type WorkflowPlacementValidator interface {
28
+
ValidateWorkflowPlacement(wf *models.Workflow) error
29
+
}
30
+
31
+
// NormalizeCapabilities validates, sorts, and deduplicates an opaque
32
+
// capability set so snapshots and workflow requirements compare deterministically.
33
+
func NormalizeCapabilities(capabilities []string) ([]string, error) {
34
+
seen := make(map[string]struct{}, len(capabilities))
35
+
normalized := make([]string, 0, len(capabilities))
36
+
for _, capability := range capabilities {
37
+
capability = strings.TrimSpace(capability)
38
+
if capability == "" || strings.ContainsFunc(capability, unicode.IsSpace) {
39
+
return nil, fmt.Errorf("invalid empty or whitespace-containing engine capability %q", capability)
40
+
}
41
+
if _, ok := seen[capability]; ok {
42
+
continue
43
+
}
44
+
seen[capability] = struct{}{}
45
+
normalized = append(normalized, capability)
46
+
}
47
+
slices.Sort(normalized)
48
+
return normalized, nil
49
+
}
+22
spindle/engine/placement_test.go
+22
spindle/engine/placement_test.go
···
1
+
package engine
2
+
3
+
import "testing"
4
+
5
+
func TestNormalizeCapabilities(t *testing.T) {
6
+
got, err := NormalizeCapabilities([]string{" runner/qemu ", "image/nixos", "runner/qemu"})
7
+
if err != nil {
8
+
t.Fatalf("NormalizeCapabilities() error = %v", err)
9
+
}
10
+
want := []string{"image/nixos", "runner/qemu"}
11
+
if len(got) != len(want) || got[0] != want[0] || got[1] != want[1] {
12
+
t.Fatalf("NormalizeCapabilities() = %v, want %v", got, want)
13
+
}
14
+
}
15
+
16
+
func TestNormalizeCapabilitiesRejectsInvalidAtoms(t *testing.T) {
17
+
for _, capabilities := range [][]string{{""}, {" "}, {"image/nixos arm"}, {"image/nixos\tarm"}} {
18
+
if got, err := NormalizeCapabilities(capabilities); err == nil {
19
+
t.Errorf("NormalizeCapabilities(%q) = %v, nil; want error", capabilities, got)
20
+
}
21
+
}
22
+
}
+16
spindle/engine_microvm_linux.go
+16
spindle/engine_microvm_linux.go
···
1
+
//go:build linux
2
+
3
+
package spindle
4
+
5
+
import (
6
+
"context"
7
+
8
+
"tangled.org/core/spindle/config"
9
+
"tangled.org/core/spindle/db"
10
+
"tangled.org/core/spindle/engines/microvm"
11
+
"tangled.org/core/spindle/models"
12
+
)
13
+
14
+
func newMicrovmEngine(ctx context.Context, cfg *config.Config, d *db.DB) (models.Engine, error) {
15
+
return microvm.New(ctx, cfg, d)
16
+
}
+16
spindle/engine_microvm_other.go
+16
spindle/engine_microvm_other.go
···
1
+
//go:build !linux
2
+
3
+
package spindle
4
+
5
+
import (
6
+
"context"
7
+
"fmt"
8
+
9
+
"tangled.org/core/spindle/config"
10
+
"tangled.org/core/spindle/db"
11
+
"tangled.org/core/spindle/models"
12
+
)
13
+
14
+
func newMicrovmEngine(context.Context, *config.Config, *db.DB) (models.Engine, error) {
15
+
return nil, fmt.Errorf("microvm engine is only supported on Linux")
16
+
}
+2
spindle/engines/microvm/agent.go
+2
spindle/engines/microvm/agent.go
+2
spindle/engines/microvm/args.go
+2
spindle/engines/microvm/args.go
+2
spindle/engines/microvm/budget.go
+2
spindle/engines/microvm/budget.go
+2
spindle/engines/microvm/cgroup.go
+2
spindle/engines/microvm/cgroup.go
+2
spindle/engines/microvm/cgroup_oom_test.go
+2
spindle/engines/microvm/cgroup_oom_test.go
+2
spindle/engines/microvm/cgroup_test.go
+2
spindle/engines/microvm/cgroup_test.go
+2
spindle/engines/microvm/dns_proxy.go
+2
spindle/engines/microvm/dns_proxy.go
+2
spindle/engines/microvm/dns_proxy_test.go
+2
spindle/engines/microvm/dns_proxy_test.go
+6
spindle/engines/microvm/engine.go
+6
spindle/engines/microvm/engine.go
···
1
+
//go:build linux
2
+
1
3
package microvm
2
4
3
5
import (
···
49
51
agent *agentHub
50
52
scheduler *engine.ResourceScheduler[Resources]
51
53
cgroupParent *CgroupParent
54
+
budget Resources
55
+
maxWorkflow Resources
52
56
53
57
cleanupMu sync.Mutex
54
58
cleanup map[string][]cleanupFunc
···
96
100
agent: agent,
97
101
scheduler: engine.NewResourceScheduler(budget, max, agingThreshold),
98
102
cgroupParent: cgroupParent,
103
+
budget: budget,
104
+
maxWorkflow: max,
99
105
cleanup: make(map[string][]cleanupFunc),
100
106
}, nil
101
107
}
+2
spindle/engines/microvm/engine_test.go
+2
spindle/engines/microvm/engine_test.go
+5
-7
spindle/engines/microvm/image.go
+5
-7
spindle/engines/microvm/image.go
···
1
+
//go:build linux
2
+
1
3
package microvm
2
4
3
5
import (
···
7
9
"os"
8
10
"path/filepath"
9
11
"strings"
12
+
13
+
"tangled.org/core/spindle/engines/microvm/placement"
10
14
)
11
15
12
16
const imageSpecFileName = "spec.json"
···
202
206
203
207
// check if image name is not a path
204
208
func isPlainImageName(name string) bool {
205
-
if name == "" || name == "." || name == ".." {
206
-
return false
207
-
}
208
-
if filepath.IsAbs(name) || strings.ContainsRune(name, '/') || strings.ContainsRune(name, filepath.Separator) {
209
-
return false
210
-
}
211
-
return true
209
+
return placement.IsPlainImageName(name)
212
210
}
213
211
214
212
// returns candidates, which is either a directory or spec file itself
+2
spindle/engines/microvm/image_test.go
+2
spindle/engines/microvm/image_test.go
+2
spindle/engines/microvm/models.go
+2
spindle/engines/microvm/models.go
+2
spindle/engines/microvm/models_test.go
+2
spindle/engines/microvm/models_test.go
+2
spindle/engines/microvm/networking.go
+2
spindle/engines/microvm/networking.go
+2
spindle/engines/microvm/nixos_toplevel_cache.go
+2
spindle/engines/microvm/nixos_toplevel_cache.go
+66
spindle/engines/microvm/placement/placement.go
+66
spindle/engines/microvm/placement/placement.go
···
1
+
package placement
2
+
3
+
import (
4
+
"fmt"
5
+
"strings"
6
+
"unicode"
7
+
8
+
"gopkg.in/yaml.v3"
9
+
10
+
"tangled.org/core/api/tangled"
11
+
)
12
+
13
+
const DefaultImageCapability = "image/default"
14
+
15
+
type Planner struct{}
16
+
17
+
func (Planner) Requirements(twf tangled.Pipeline_Workflow, _ tangled.Pipeline) ([]string, error) {
18
+
var manifest struct {
19
+
Image string `yaml:"image"`
20
+
}
21
+
if err := yaml.Unmarshal([]byte(twf.Raw), &manifest); err != nil {
22
+
return nil, fmt.Errorf("parse microVM workflow placement: %w", err)
23
+
}
24
+
25
+
image := strings.TrimSpace(manifest.Image)
26
+
if image == "" {
27
+
return []string{DefaultImageCapability}, nil
28
+
}
29
+
capability, err := ImageCapability(image)
30
+
if err != nil {
31
+
return nil, err
32
+
}
33
+
return []string{capability}, nil
34
+
}
35
+
36
+
func ImageCapability(name string) (string, error) {
37
+
name = strings.TrimSpace(name)
38
+
if !IsPlainImageName(name) {
39
+
return "", fmt.Errorf("invalid microVM image name %q: must be a whitespace-free plain name, not a path", name)
40
+
}
41
+
return "image/" + name, nil
42
+
}
43
+
44
+
func IsPlainImageName(name string) bool {
45
+
if name == "" || name == "." || name == ".." {
46
+
return false
47
+
}
48
+
if strings.ContainsAny(name, `/\\`) || strings.ContainsFunc(name, unicode.IsSpace) {
49
+
return false
50
+
}
51
+
return true
52
+
}
53
+
54
+
func IsNativeArchitecture(imageArch, goArch string) bool {
55
+
normalize := func(arch string) string {
56
+
switch arch {
57
+
case "x86_64", "amd64":
58
+
return "amd64"
59
+
case "aarch64", "arm64":
60
+
return "arm64"
61
+
default:
62
+
return arch
63
+
}
64
+
}
65
+
return imageArch != "" && normalize(imageArch) == normalize(goArch)
66
+
}
+78
spindle/engines/microvm/placement/placement_test.go
+78
spindle/engines/microvm/placement/placement_test.go
···
1
+
package placement
2
+
3
+
import (
4
+
"testing"
5
+
6
+
"tangled.org/core/api/tangled"
7
+
)
8
+
9
+
func TestPlannerRequirements(t *testing.T) {
10
+
tests := []struct {
11
+
name string
12
+
raw string
13
+
want string
14
+
wantErr bool
15
+
}{
16
+
{name: "explicit image", raw: "image: nixos\n", want: "image/nixos"},
17
+
{name: "explicit arm image", raw: "image: nixos-aarch64\n", want: "image/nixos-aarch64"},
18
+
{name: "omitted image", raw: "steps: []\n", want: DefaultImageCapability},
19
+
{name: "blank image", raw: "image: ' '\n", want: DefaultImageCapability},
20
+
{name: "path", raw: "image: ../nixos\n", wantErr: true},
21
+
{name: "whitespace", raw: "image: 'nixos arm'\n", wantErr: true},
22
+
{name: "wrong yaml type", raw: "image: [nixos]\n", wantErr: true},
23
+
{name: "malformed yaml", raw: "image: [\n", wantErr: true},
24
+
}
25
+
26
+
planner := Planner{}
27
+
for _, tt := range tests {
28
+
t.Run(tt.name, func(t *testing.T) {
29
+
got, err := planner.Requirements(tangled.Pipeline_Workflow{Raw: tt.raw}, tangled.Pipeline{})
30
+
if tt.wantErr {
31
+
if err == nil {
32
+
t.Fatalf("Requirements() = %v, nil; want error", got)
33
+
}
34
+
return
35
+
}
36
+
if err != nil {
37
+
t.Fatalf("Requirements() error = %v", err)
38
+
}
39
+
if len(got) != 1 || got[0] != tt.want {
40
+
t.Fatalf("Requirements() = %v, want [%s]", got, tt.want)
41
+
}
42
+
})
43
+
}
44
+
}
45
+
46
+
func TestIsPlainImageName(t *testing.T) {
47
+
for _, name := range []string{"nixos", "nixos-aarch64", "release.2026_07"} {
48
+
if !IsPlainImageName(name) {
49
+
t.Errorf("IsPlainImageName(%q) = false, want true", name)
50
+
}
51
+
}
52
+
for _, name := range []string{"", ".", "..", "/nixos", "../nixos", `dir\\nixos`, "nixos arm", "nixos\tarm"} {
53
+
if IsPlainImageName(name) {
54
+
t.Errorf("IsPlainImageName(%q) = true, want false", name)
55
+
}
56
+
}
57
+
}
58
+
59
+
func TestIsNativeArchitecture(t *testing.T) {
60
+
tests := []struct {
61
+
image string
62
+
host string
63
+
want bool
64
+
}{
65
+
{image: "x86_64", host: "amd64", want: true},
66
+
{image: "amd64", host: "amd64", want: true},
67
+
{image: "aarch64", host: "arm64", want: true},
68
+
{image: "arm64", host: "arm64", want: true},
69
+
{image: "x86_64", host: "arm64", want: false},
70
+
{image: "aarch64", host: "amd64", want: false},
71
+
{image: "", host: "amd64", want: false},
72
+
}
73
+
for _, tt := range tests {
74
+
if got := IsNativeArchitecture(tt.image, tt.host); got != tt.want {
75
+
t.Errorf("IsNativeArchitecture(%q, %q) = %v, want %v", tt.image, tt.host, got, tt.want)
76
+
}
77
+
}
78
+
}
+109
spindle/engines/microvm/placement_linux.go
+109
spindle/engines/microvm/placement_linux.go
···
1
+
//go:build linux
2
+
3
+
package microvm
4
+
5
+
import (
6
+
"fmt"
7
+
"os"
8
+
"os/exec"
9
+
"runtime"
10
+
"slices"
11
+
"strings"
12
+
13
+
"tangled.org/core/spindle/engines/microvm/placement"
14
+
"tangled.org/core/spindle/models"
15
+
)
16
+
17
+
func (e *Engine) Capabilities() []string {
18
+
entries, err := os.ReadDir(e.cfg.MicroVMPipelines.ImageDir)
19
+
if err != nil {
20
+
e.l.Error("discover microVM placement capabilities", "err", err)
21
+
return nil
22
+
}
23
+
24
+
seen := make(map[string]struct{}, len(entries)+1)
25
+
capabilities := make([]string, 0, len(entries)+1)
26
+
addImage := func(name string) bool {
27
+
name = strings.TrimSuffix(name, ".json")
28
+
if !placement.IsPlainImageName(name) {
29
+
return false
30
+
}
31
+
capability, err := placement.ImageCapability(name)
32
+
if err != nil {
33
+
return false
34
+
}
35
+
if _, ok := seen[capability]; ok {
36
+
return true
37
+
}
38
+
spec, _, _, err := e.resolveImage(name)
39
+
if err != nil {
40
+
e.l.Debug("microVM image is unavailable for placement", "image", name, "err", err)
41
+
return false
42
+
}
43
+
if err := e.validateImagePlacement(spec); err != nil {
44
+
e.l.Debug("microVM image is incompatible with executor", "image", name, "err", err)
45
+
return false
46
+
}
47
+
seen[capability] = struct{}{}
48
+
capabilities = append(capabilities, capability)
49
+
return true
50
+
}
51
+
52
+
for _, entry := range entries {
53
+
addImage(entry.Name())
54
+
}
55
+
if defaultImage := strings.TrimSpace(e.cfg.MicroVMPipelines.DefaultImage); defaultImage != "" && addImage(defaultImage) {
56
+
capabilities = append(capabilities, placement.DefaultImageCapability)
57
+
}
58
+
slices.Sort(capabilities)
59
+
return capabilities
60
+
}
61
+
62
+
func (e *Engine) ValidateWorkflowPlacement(wf *models.Workflow) error {
63
+
state, ok := wf.Data.(*workflowState)
64
+
if !ok || state == nil {
65
+
return fmt.Errorf("microVM workflow state is not initialized")
66
+
}
67
+
return e.validateImagePlacement(state.ImageSpec)
68
+
}
69
+
70
+
func (e *Engine) validateImagePlacement(spec ImageSpec) error {
71
+
if err := spec.Validate(); err != nil {
72
+
return err
73
+
}
74
+
if !placement.IsNativeArchitecture(spec.Arch, runtime.GOARCH) {
75
+
return fmt.Errorf("microVM image architecture %q is not native to executor architecture %q", spec.Arch, runtime.GOARCH)
76
+
}
77
+
if err := spec.validateImageFiles(); err != nil {
78
+
return err
79
+
}
80
+
runner, err := runnerFor(spec.RunnerType)
81
+
if err != nil {
82
+
return err
83
+
}
84
+
if err := runner.Validate(spec, e.cfg.MicroVMPipelines.EnableKVM); err != nil {
85
+
return err
86
+
}
87
+
if len(spec.Volumes) > 0 {
88
+
if _, err := exec.LookPath("mkfs.ext4"); err != nil {
89
+
return fmt.Errorf("required host command %q not found in PATH: %w", "mkfs.ext4", err)
90
+
}
91
+
for _, volume := range spec.Volumes {
92
+
if volume.ReadOnly {
93
+
return fmt.Errorf("read-only microvm volume %q is not supported yet", volume.Image)
94
+
}
95
+
if volume.FSType != "ext4" {
96
+
return fmt.Errorf("microvm volume %q uses unsupported fsType %q", volume.Image, volume.FSType)
97
+
}
98
+
if volume.ImageType != "" && volume.ImageType != "raw" {
99
+
return fmt.Errorf("microvm volume %q uses unsupported imageType %q", volume.Image, volume.ImageType)
100
+
}
101
+
}
102
+
}
103
+
104
+
request := resourcesForImage(spec)
105
+
if !request.Fits(e.budget) || !request.Fits(e.maxWorkflow) {
106
+
return fmt.Errorf("microVM image resources exceed executor limits: request=%v budget=%v max=%v", request, e.budget, e.maxWorkflow)
107
+
}
108
+
return nil
109
+
}
+2
spindle/engines/microvm/qemu.go
+2
spindle/engines/microvm/qemu.go
+2
spindle/engines/microvm/read_cache_proxy.go
+2
spindle/engines/microvm/read_cache_proxy.go
+2
spindle/engines/microvm/read_cache_proxy_test.go
+2
spindle/engines/microvm/read_cache_proxy_test.go
+2
spindle/engines/microvm/runner.go
+2
spindle/engines/microvm/runner.go
+2
spindle/engines/microvm/upload_cache_http.go
+2
spindle/engines/microvm/upload_cache_http.go
+2
spindle/engines/microvm/upload_cache_narinfo.go
+2
spindle/engines/microvm/upload_cache_narinfo.go
+2
spindle/engines/microvm/upload_cache_nix_store.go
+2
spindle/engines/microvm/upload_cache_nix_store.go
+2
spindle/engines/microvm/upload_cache_nix_store_test.go
+2
spindle/engines/microvm/upload_cache_nix_store_test.go
+2
spindle/engines/microvm/upload_cache_proxy.go
+2
spindle/engines/microvm/upload_cache_proxy.go
+2
spindle/engines/microvm/upload_cache_proxy_test.go
+2
spindle/engines/microvm/upload_cache_proxy_test.go
+99
spindle/mill/capability_test.go
+99
spindle/mill/capability_test.go
···
1
+
package mill
2
+
3
+
import (
4
+
"errors"
5
+
"io"
6
+
"log/slog"
7
+
"testing"
8
+
9
+
"tangled.org/core/api/tangled"
10
+
"tangled.org/core/spindle/models"
11
+
12
+
millproto "tangled.org/core/spindle/mill/proto"
13
+
)
14
+
15
+
type testPlacementPlanner struct {
16
+
requirements []string
17
+
err error
18
+
}
19
+
20
+
func (p testPlacementPlanner) Requirements(tangled.Pipeline_Workflow, tangled.Pipeline) ([]string, error) {
21
+
return p.requirements, p.err
22
+
}
23
+
24
+
func TestMillEngineStoresCanonicalPlacementRequirements(t *testing.T) {
25
+
m := New(slog.New(slog.NewTextHandler(io.Discard, nil)), Config{})
26
+
eng := NewEngineWithPlanner("microvm", m, testPlacementPlanner{
27
+
requirements: []string{"runner/qemu", "image/nixos", "runner/qemu"},
28
+
})
29
+
30
+
wf, err := eng.InitWorkflow(tangled.Pipeline_Workflow{Name: "build"}, tangled.Pipeline{})
31
+
if err != nil {
32
+
t.Fatalf("InitWorkflow() error = %v", err)
33
+
}
34
+
got := wf.Data.(*millWorkflowState).RequiredCapabilities
35
+
want := []string{"image/nixos", "runner/qemu"}
36
+
if len(got) != len(want) || got[0] != want[0] || got[1] != want[1] {
37
+
t.Fatalf("required capabilities = %v, want %v", got, want)
38
+
}
39
+
}
40
+
41
+
func TestMillEngineRejectsInvalidPlacementRequirements(t *testing.T) {
42
+
m := New(slog.New(slog.NewTextHandler(io.Discard, nil)), Config{})
43
+
plannerErr := errors.New("invalid image")
44
+
eng := NewEngineWithPlanner("microvm", m, testPlacementPlanner{err: plannerErr})
45
+
if _, err := eng.InitWorkflow(tangled.Pipeline_Workflow{}, tangled.Pipeline{}); !errors.Is(err, plannerErr) {
46
+
t.Fatalf("InitWorkflow() error = %v, want wrapped %v", err, plannerErr)
47
+
}
48
+
49
+
eng = NewEngineWithPlanner("microvm", m, testPlacementPlanner{requirements: []string{"image/nixos arm"}})
50
+
if _, err := eng.InitWorkflow(tangled.Pipeline_Workflow{}, tangled.Pipeline{}); err == nil {
51
+
t.Fatal("InitWorkflow() accepted whitespace-containing capability")
52
+
}
53
+
}
54
+
55
+
func TestRankCandidatesIntersectsLabelsAndEngineCapabilities(t *testing.T) {
56
+
m := New(slog.New(slog.NewTextHandler(io.Discard, nil)), Config{})
57
+
incapable := addCandidateSession(t, m, "low-load-incapable", []string{"region/eu"}, 0.05, nil)
58
+
incapable.snapshot.Engines["dummy"].Capabilities = []string{"image/nixos-aarch64"}
59
+
wrongRegion := addCandidateSession(t, m, "wrong-region", []string{"region/us"}, 0.10, nil)
60
+
wrongRegion.snapshot.Engines["dummy"].Capabilities = []string{"image/nixos", "runner/qemu"}
61
+
partial := addCandidateSession(t, m, "partial", []string{"region/eu"}, 0.20, nil)
62
+
partial.snapshot.Engines["dummy"].Capabilities = []string{"image/nixos"}
63
+
eligible := addCandidateSession(t, m, "eligible", []string{"region/eu"}, 0.70, nil)
64
+
eligible.snapshot.Engines["dummy"].Capabilities = []string{"runner/qemu", "image/nixos"}
65
+
66
+
got := m.rankCandidates("dummy", []string{"region/eu"}, []string{"image/nixos", "runner/qemu"})
67
+
assertRankedNodes(t, got, []string{"eligible"})
68
+
}
69
+
70
+
func TestBidWithMissingEngineCapabilitySendsNoReserve(t *testing.T) {
71
+
m := New(slog.New(slog.NewTextHandler(io.Discard, nil)), Config{})
72
+
reserveSent := false
73
+
sess := addCandidateSession(t, m, "arm", nil, 0.05, scriptedEncoder(func(msg *millproto.Message) error {
74
+
if msg.GetReserveSeat() != nil {
75
+
reserveSent = true
76
+
}
77
+
return nil
78
+
}))
79
+
sess.snapshot.Engines["dummy"].Capabilities = []string{"image/nixos-aarch64"}
80
+
wf := testWorkflow("build")
81
+
wf.Data.(*millWorkflowState).RequiredCapabilities = []string{"image/nixos"}
82
+
83
+
lease, err := m.bid(t.Context(), "dummy", wf)
84
+
if err != nil {
85
+
t.Fatalf("bid() error = %v", err)
86
+
}
87
+
if lease != nil {
88
+
t.Fatalf("bid() lease = %v, want nil", lease)
89
+
}
90
+
if reserveSent {
91
+
t.Fatal("bid() sent ReserveSeat to an engine missing a required capability")
92
+
}
93
+
}
94
+
95
+
func TestRequiredCapabilitiesWithoutMillStateAreEmpty(t *testing.T) {
96
+
if got := requiredCapabilities(&models.Workflow{}); got != nil {
97
+
t.Fatalf("requiredCapabilities() = %v, want nil", got)
98
+
}
99
+
}
+36
-13
spindle/mill/engine.go
+36
-13
spindle/mill/engine.go
···
2
2
3
3
import (
4
4
"context"
5
+
"fmt"
5
6
"log/slog"
6
7
"time"
7
8
···
15
16
// engines use it). InitWorkflow parses nothing here: it just carries the raw
16
17
// pipeline/workflow forward so the executor can run the real InitWorkflow later.
17
18
type millWorkflowState struct {
18
-
TargetEngine string
19
-
RawWorkflow tangled.Pipeline_Workflow
20
-
RawPipeline tangled.Pipeline
21
-
Wid models.WorkflowId // stamped at placement (InitWorkflow can't see it)
22
-
Lease *RemoteLease
19
+
TargetEngine string
20
+
RawWorkflow tangled.Pipeline_Workflow
21
+
RawPipeline tangled.Pipeline
22
+
RequiredCapabilities []string
23
+
Wid models.WorkflowId // stamped at placement (InitWorkflow can't see it)
24
+
Lease *RemoteLease
23
25
}
24
26
25
27
// Engine is the mill's stand-in for a real engine, registered under the real
26
28
// engine names ("microvm", "nixery"). All registered names share one Mill.
27
29
type Engine struct {
28
-
name string
29
-
mill *Mill
30
-
l *slog.Logger
30
+
name string
31
+
mill *Mill
32
+
planner engine.PlacementPlanner
33
+
l *slog.Logger
31
34
}
32
35
33
-
// NewEngine returns a mill engine view for one engine name.
36
+
// NewEngine returns a mill engine view for one engine name without
37
+
// engine-derived placement requirements.
34
38
func NewEngine(name string, mill *Mill) *Engine {
35
-
return &Engine{name: name, mill: mill, l: mill.l.With("engine", "mill:"+name)}
39
+
return NewEngineWithPlanner(name, mill, nil)
40
+
}
41
+
42
+
// NewEngineWithPlanner returns a mill engine view whose trusted planner
43
+
// derives requirements for capability-aware placement.
44
+
func NewEngineWithPlanner(name string, mill *Mill, planner engine.PlacementPlanner) *Engine {
45
+
return &Engine{name: name, mill: mill, planner: planner, l: mill.l.With("engine", "mill:"+name)}
36
46
}
37
47
38
48
// InitWorkflow returns a synthetic one-step workflow so processPipeline injects
39
49
// TANGLED_* env and marks pending normally. The real InitWorkflow runs later on
40
50
// the executor inside ReserveSeat (intended: it runs twice).
41
51
func (e *Engine) InitWorkflow(twf tangled.Pipeline_Workflow, tpl tangled.Pipeline) (*models.Workflow, error) {
52
+
var required []string
53
+
if e.planner != nil {
54
+
var err error
55
+
required, err = e.planner.Requirements(twf, tpl)
56
+
if err != nil {
57
+
return nil, fmt.Errorf("derive %s placement requirements: %w", e.name, err)
58
+
}
59
+
required, err = engine.NormalizeCapabilities(required)
60
+
if err != nil {
61
+
return nil, fmt.Errorf("derive %s placement requirements: %w", e.name, err)
62
+
}
63
+
}
42
64
return &models.Workflow{
43
65
Name: twf.Name,
44
66
Environment: map[string]string{},
45
67
Steps: []models.Step{remoteStep{}},
46
68
Data: &millWorkflowState{
47
-
TargetEngine: e.name,
48
-
RawWorkflow: twf,
49
-
RawPipeline: tpl,
69
+
TargetEngine: e.name,
70
+
RawWorkflow: twf,
71
+
RawPipeline: tpl,
72
+
RequiredCapabilities: required,
50
73
},
51
74
}, nil
52
75
}
+106
spindle/mill/executor/capability_test.go
+106
spindle/mill/executor/capability_test.go
···
1
+
package executor
2
+
3
+
import (
4
+
"context"
5
+
"encoding/json"
6
+
"errors"
7
+
"io"
8
+
"log/slog"
9
+
"testing"
10
+
11
+
"tangled.org/core/api/tangled"
12
+
millv1 "tangled.org/core/spindle/mill/proto/gen"
13
+
"tangled.org/core/spindle/models"
14
+
)
15
+
16
+
type placementEngine struct {
17
+
*fakeEngine
18
+
capabilities []string
19
+
validationErr error
20
+
validated bool
21
+
}
22
+
23
+
func (e *placementEngine) Capabilities() []string {
24
+
return e.capabilities
25
+
}
26
+
27
+
func (e *placementEngine) ValidateWorkflowPlacement(*models.Workflow) error {
28
+
e.validated = true
29
+
return e.validationErr
30
+
}
31
+
32
+
func TestPushSnapshotAdvertisesCanonicalPerEngineCapabilities(t *testing.T) {
33
+
enc := newCaptureEncoder()
34
+
eng := &placementEngine{
35
+
fakeEngine: &fakeEngine{},
36
+
capabilities: []string{"runner/qemu", "image/nixos", "runner/qemu"},
37
+
}
38
+
e := &Executor{
39
+
l: slog.New(slog.NewTextHandler(io.Discard, nil)),
40
+
enc: enc,
41
+
seats: 2,
42
+
engines: map[string]models.Engine{"microvm": eng},
43
+
active: make(map[string]*reservation),
44
+
}
45
+
46
+
e.pushSnapshot()
47
+
availability := (<-enc.messages).GetNodeSnapshot().GetEngines()["microvm"]
48
+
if availability == nil {
49
+
t.Fatal("snapshot omitted microvm availability")
50
+
}
51
+
got := availability.GetCapabilities()
52
+
want := []string{"image/nixos", "runner/qemu"}
53
+
if len(got) != len(want) || got[0] != want[0] || got[1] != want[1] {
54
+
t.Fatalf("snapshot capabilities = %v, want %v", got, want)
55
+
}
56
+
}
57
+
58
+
func TestHandleReserveValidatesPlacementBeforeAcquiringSlot(t *testing.T) {
59
+
enc := newCaptureEncoder()
60
+
validationErr := errors.New("image architecture is not native")
61
+
eng := &placementEngine{fakeEngine: &fakeEngine{}, validationErr: validationErr}
62
+
e := &Executor{
63
+
l: slog.New(slog.NewTextHandler(io.Discard, nil)),
64
+
enc: enc,
65
+
seats: 1,
66
+
engines: map[string]models.Engine{"microvm": eng},
67
+
active: make(map[string]*reservation),
68
+
}
69
+
twf, err := json.Marshal(tangled.Pipeline_Workflow{Name: "build"})
70
+
if err != nil {
71
+
t.Fatal(err)
72
+
}
73
+
tpl, err := json.Marshal(tangled.Pipeline{})
74
+
if err != nil {
75
+
t.Fatal(err)
76
+
}
77
+
78
+
e.handleReserve(context.Background(), &millv1.ReserveSeat{
79
+
LeaseId: "lease-1",
80
+
TargetEngine: "microvm",
81
+
RawWorkflowJson: string(twf),
82
+
RawPipelineJson: string(tpl),
83
+
Knot: "k",
84
+
Rkey: "r",
85
+
})
86
+
87
+
result := (<-enc.messages).GetReserveResult()
88
+
if result == nil {
89
+
t.Fatal("handleReserve() did not send ReserveResult")
90
+
}
91
+
if result.GetAccepted() {
92
+
t.Fatal("handleReserve() accepted placement validation failure")
93
+
}
94
+
if result.GetRejectClass() != millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE {
95
+
t.Fatalf("reject class = %v, want incompatible", result.GetRejectClass())
96
+
}
97
+
if !eng.validated {
98
+
t.Fatal("placement validator was not called")
99
+
}
100
+
if eng.acquireCalled {
101
+
t.Fatal("slot acquisition ran after placement validation failed")
102
+
}
103
+
if len(e.active) != 0 {
104
+
t.Fatalf("active reservations = %d, want 0", len(e.active))
105
+
}
106
+
}
+19
-3
spindle/mill/executor/executor.go
+19
-3
spindle/mill/executor/executor.go
···
264
264
reject("init workflow: "+err.Error(), millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE)
265
265
return
266
266
}
267
+
if validator, ok := realEngine.(engine.WorkflowPlacementValidator); ok {
268
+
if err := validator.ValidateWorkflowPlacement(wf); err != nil {
269
+
reject("validate workflow placement: "+err.Error(), millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE)
270
+
return
271
+
}
272
+
}
267
273
// the job skipped processPipeline, so inject TANGLED_* env here.
268
274
if wf.Environment == nil {
269
275
wf.Environment = make(map[string]string)
···
480
486
}
481
487
available := !draining && active < e.seats
482
488
engines := make(map[string]*millv1.EngineAvailability, len(e.engines))
483
-
for name := range e.engines {
489
+
for name, realEngine := range e.engines {
490
+
var capabilities []string
491
+
if provider, ok := realEngine.(engine.CapabilityProvider); ok {
492
+
var err error
493
+
capabilities, err = engine.NormalizeCapabilities(provider.Capabilities())
494
+
if err != nil {
495
+
e.l.Error("engine advertised invalid placement capabilities", "engine", name, "err", err)
496
+
capabilities = nil
497
+
}
498
+
}
484
499
engines[name] = &millv1.EngineAvailability{
485
-
Available: available,
486
-
Load: map[string]float64{"slots": load},
500
+
Available: available,
501
+
Load: map[string]float64{"slots": load},
502
+
Capabilities: capabilities,
487
503
}
488
504
}
489
505
e.send(&millproto.Message{NodeSnapshot: &millv1.NodeSnapshot{
+14
-2
spindle/mill/mill.go
+14
-2
spindle/mill/mill.go
···
283
283
}
284
284
285
285
requiredLabels := requiredLabels(wf)
286
-
candidates := m.rankCandidates(engineName, requiredLabels)
286
+
requiredCapabilities := requiredCapabilities(wf)
287
+
candidates := m.rankCandidates(engineName, requiredLabels, requiredCapabilities)
287
288
if len(candidates) == 0 {
288
289
return nil, nil
289
290
}
···
390
391
return winner.lease, nil
391
392
}
392
393
393
-
func (m *Mill) rankCandidates(engineName string, requiredLabels []string) []*millSession {
394
+
func (m *Mill) rankCandidates(engineName string, requiredLabels, requiredCapabilities []string) []*millSession {
394
395
m.mu.Lock()
395
396
defer m.mu.Unlock()
396
397
···
412
413
continue
413
414
}
414
415
if !hasLabels(s.labels, requiredLabels) {
416
+
continue
417
+
}
418
+
if !hasLabels(ea.GetCapabilities(), requiredCapabilities) {
415
419
continue
416
420
}
417
421
worst, sum := loadScore(ea.GetLoad())
···
456
460
return nil
457
461
}
458
462
return st.RawWorkflow.RunsOn
463
+
}
464
+
465
+
func requiredCapabilities(wf *models.Workflow) []string {
466
+
st, ok := wf.Data.(*millWorkflowState)
467
+
if !ok || st == nil {
468
+
return nil
469
+
}
470
+
return st.RequiredCapabilities
459
471
}
460
472
461
473
func hasLabels(labels []string, required []string) bool {
+1
-1
spindle/mill/mill_test.go
+1
-1
spindle/mill/mill_test.go
+14
-3
spindle/mill/proto/gen/mill.pb.go
+14
-3
spindle/mill/proto/gen/mill.pb.go
···
190
190
state protoimpl.MessageState `protogen:"open.v1"`
191
191
Available bool `protobuf:"varint,1,opt,name=available,proto3" json:"available,omitempty"`
192
192
// engine-defined load metrics. keys are opaque to the mill.
193
-
Load map[string]float64 `protobuf:"bytes,2,rep,name=load,proto3" json:"load,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"fixed64,2,opt,name=value"`
193
+
Load map[string]float64 `protobuf:"bytes,2,rep,name=load,proto3" json:"load,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"fixed64,2,opt,name=value"`
194
+
// engine-defined placement facts. keys are opaque to the mill and matched
195
+
// exactly against requirements from the trusted engine planner.
196
+
Capabilities []string `protobuf:"bytes,3,rep,name=capabilities,proto3" json:"capabilities,omitempty"`
194
197
unknownFields protoimpl.UnknownFields
195
198
sizeCache protoimpl.SizeCache
196
199
}
···
235
238
func (x *EngineAvailability) GetLoad() map[string]float64 {
236
239
if x != nil {
237
240
return x.Load
241
+
}
242
+
return nil
243
+
}
244
+
245
+
func (x *EngineAvailability) GetCapabilities() []string {
246
+
if x != nil {
247
+
return x.Capabilities
238
248
}
239
249
return nil
240
250
}
···
1154
1164
"\x06labels\x18\x03 \x03(\tR\x06labels\"'\n" +
1155
1165
"\x06Resume\x12\x1d\n" +
1156
1166
"\n" +
1157
-
"ack_offset\x18\x01 \x01(\x04R\tackOffset\"\xae\x01\n" +
1167
+
"ack_offset\x18\x01 \x01(\x04R\tackOffset\"\xd2\x01\n" +
1158
1168
"\x12EngineAvailability\x12\x1c\n" +
1159
1169
"\tavailable\x18\x01 \x01(\bR\tavailable\x12A\n" +
1160
-
"\x04load\x18\x02 \x03(\v2-.spindle.mill.v1.EngineAvailability.LoadEntryR\x04load\x1a7\n" +
1170
+
"\x04load\x18\x02 \x03(\v2-.spindle.mill.v1.EngineAvailability.LoadEntryR\x04load\x12\"\n" +
1171
+
"\fcapabilities\x18\x03 \x03(\tR\fcapabilities\x1a7\n" +
1161
1172
"\tLoadEntry\x12\x10\n" +
1162
1173
"\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" +
1163
1174
"\x05value\x18\x02 \x01(\x01R\x05value:\x028\x01\"\xe0\x01\n" +
+3
spindle/mill/proto/spindle/mill/v1/mill.proto
+3
spindle/mill/proto/spindle/mill/v1/mill.proto
···
31
31
bool available = 1;
32
32
// engine-defined load metrics. keys are opaque to the mill.
33
33
map<string, double> load = 2;
34
+
// engine-defined placement facts. keys are opaque to the mill and matched
35
+
// exactly against requirements from the trusted engine planner.
36
+
repeated string capabilities = 3;
34
37
}
35
38
36
39
// NodeSnapshot is pushed on connect, periodically, and right after any state
+3
-3
spindle/server.go
+3
-3
spindle/server.go
···
33
33
"tangled.org/core/spindle/db"
34
34
"tangled.org/core/spindle/engine"
35
35
"tangled.org/core/spindle/engines/dummy"
36
-
"tangled.org/core/spindle/engines/microvm"
36
+
microvmplacement "tangled.org/core/spindle/engines/microvm/placement"
37
37
"tangled.org/core/spindle/engines/nixery"
38
38
"tangled.org/core/spindle/git"
39
39
"tangled.org/core/spindle/mill"
···
384
384
})
385
385
engines = map[string]models.Engine{
386
386
"nixery": mill.NewEngine("nixery", m),
387
-
"microvm": mill.NewEngine("microvm", m),
387
+
"microvm": mill.NewEngineWithPlanner("microvm", m, microvmplacement.Planner{}),
388
388
"dummy": mill.NewEngine("dummy", m),
389
389
}
390
390
} else {
···
393
393
if err != nil {
394
394
return err
395
395
}
396
-
microvmEng, err := microvm.New(ctx, cfg, d)
396
+
microvmEng, err := newMicrovmEngine(ctx, cfg, d)
397
397
if err != nil {
398
398
return err
399
399
}