Monorepo for Tangled
spindle/mill: implement load-based scheduling and label matching
+678
-259
+74
-61
spindle/mill/executor/executor.go
+74
-61
spindle/mill/executor/executor.go
···
3
3
import (
4
4
"context"
5
5
"encoding/json"
6
+
"errors"
6
7
"fmt"
7
8
"log/slog"
8
9
"maps"
9
10
"net/http"
10
11
"runtime"
12
+
"strings"
11
13
"sync"
12
14
"time"
13
15
···
34
36
35
37
type Executor struct {
36
38
millURL string
37
-
secret string
39
+
token string
38
40
nodeID string
39
41
seats int
42
+
labels []string
40
43
41
44
engines map[string]models.Engine
42
45
db *db.DB
···
50
53
connMu sync.Mutex
51
54
enc messageEncoder
52
55
53
-
mu sync.Mutex
54
-
active map[string]*reservation
55
-
draining bool
56
-
activeCount int
56
+
mu sync.Mutex
57
+
active map[string]*reservation
58
+
draining bool
57
59
}
58
60
59
61
type reservation struct {
···
75
77
Encode(*millproto.Message) error
76
78
}
77
79
78
-
type reservationCleanup struct {
79
-
stopTail func()
80
-
slot engine.WorkflowSlot
81
-
}
82
-
83
-
func (c reservationCleanup) run() {
84
-
if c.stopTail != nil {
85
-
c.stopTail()
86
-
}
87
-
if c.slot != nil {
88
-
c.slot.Release()
89
-
}
90
-
}
91
-
92
80
func New(cfg *config.Config, engines map[string]models.Engine, d *db.DB, n *notifier.Notifier, l *slog.Logger) *Executor {
93
81
seats := cfg.Mill.Seats
94
82
if seats <= 0 {
95
83
seats = defaultSeats
96
84
}
85
+
labels := normalizeLabels(cfg.Mill.Labels)
97
86
return &Executor{
98
87
millURL: cfg.Mill.URL,
99
-
secret: cfg.Mill.SharedSecret,
88
+
token: cfg.Mill.SharedSecret,
100
89
nodeID: cfg.Server.Hostname,
101
90
seats: seats,
91
+
labels: labels,
102
92
engines: engines,
103
93
db: d,
104
94
n: n,
···
133
123
134
124
func (e *Executor) runSession(ctx context.Context) error {
135
125
header := http.Header{}
136
-
if e.secret != "" {
137
-
header.Set("Authorization", "Bearer "+e.secret)
126
+
if e.token != "" {
127
+
header.Set("Authorization", "Bearer "+e.token)
138
128
}
139
129
conn, _, err := websocket.DefaultDialer.DialContext(ctx, e.millURL, header)
140
130
if err != nil {
···
151
141
NodeId: e.nodeID,
152
142
Engines: e.engineNames(),
153
143
Arch: runtime.GOARCH,
144
+
Labels: e.labels,
154
145
LastOffset: e.relay.lastOffset(),
155
146
}}
156
147
if err := enc.Encode(hello); err != nil {
···
229
220
// --- reserve / commit / release / cancel ----------------------------------
230
221
231
222
func (e *Executor) handleReserve(ctx context.Context, rs *millv1.ReserveSeat) {
232
-
reject := func(reason string) {
223
+
reject := func(reason string, class millv1.RejectClass) {
233
224
e.send(&millproto.Message{ReserveResult: &millv1.ReserveResult{
234
225
LeaseId: rs.GetLeaseId(),
235
226
Accepted: false,
236
227
RejectReason: reason,
228
+
RejectClass: class,
237
229
}})
238
230
}
239
231
···
241
233
draining := e.draining
242
234
e.mu.Unlock()
243
235
if draining {
244
-
reject("draining")
236
+
reject("draining", millv1.RejectClass_REJECT_CLASS_TRANSIENT)
245
237
return
246
238
}
247
239
248
240
realEngine, ok := e.engines[rs.GetTargetEngine()]
249
241
if !ok {
250
-
reject("unknown engine " + rs.GetTargetEngine())
242
+
reject("unknown engine "+rs.GetTargetEngine(), millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE)
251
243
return
252
244
}
253
245
slotter, ok := realEngine.(engine.WorkflowSlotter)
254
246
if !ok {
255
-
reject("engine does not support workflow slots")
247
+
reject("engine does not support workflow slots", millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE)
256
248
return
257
249
}
258
250
259
251
var twf tangled.Pipeline_Workflow
260
252
if err := json.Unmarshal([]byte(rs.GetRawWorkflowJson()), &twf); err != nil {
261
-
reject("bad workflow json")
253
+
reject("bad workflow json", millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE)
262
254
return
263
255
}
264
256
var tpl tangled.Pipeline
265
257
if err := json.Unmarshal([]byte(rs.GetRawPipelineJson()), &tpl); err != nil {
266
-
reject("bad pipeline json")
258
+
reject("bad pipeline json", millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE)
267
259
return
268
260
}
269
261
···
272
264
273
265
wf, err := realEngine.InitWorkflow(twf, tpl)
274
266
if err != nil {
275
-
reject("init workflow: " + err.Error())
267
+
reject("init workflow: "+err.Error(), millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE)
276
268
return
277
269
}
278
270
// the job skipped processPipeline, so inject TANGLED_* env here.
···
284
276
// NoWait: the executor doesn't queue locally; the mill owns the backlog.
285
277
slot, err := slotter.AcquireWorkflowSlot(ctx, wid, wf, engine.NoWait)
286
278
if err != nil {
287
-
reject(err.Error())
279
+
class := millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE
280
+
if errors.Is(err, engine.ErrNoWorkflowSlots) {
281
+
class = millv1.RejectClass_REJECT_CLASS_TRANSIENT
282
+
}
283
+
reject(err.Error(), class)
288
284
return
289
285
}
290
286
···
305
301
306
302
e.mu.Lock()
307
303
e.active[res.leaseID] = res
308
-
e.activeCount++
309
304
e.mu.Unlock()
310
305
311
306
e.send(&millproto.Message{ReserveResult: &millv1.ReserveResult{
312
307
LeaseId: rs.GetLeaseId(),
313
308
Accepted: true,
314
-
Score: e.bidScore(),
315
309
}})
316
310
e.pushSnapshot()
317
311
}
···
355
349
if !ok {
356
350
return
357
351
}
358
-
cleanup.run()
352
+
cleanup()
359
353
e.pushSnapshot()
360
354
}
361
355
···
369
363
res.cancelled = true
370
364
cancel := res.cancel
371
365
committed := res.committed
372
-
var cleanup reservationCleanup
366
+
var cleanup func()
373
367
if !committed {
374
368
cleanup = e.removeReservationLocked(res, true)
375
369
}
···
381
375
// committed jobs clean up when the observe loop sees the terminal row; an
382
376
// uncommitted cancel still needs the slot released.
383
377
if !committed {
384
-
cleanup.run()
378
+
cleanup()
385
379
e.pushSnapshot()
386
380
}
387
381
}
···
392
386
return
393
387
}
394
388
e.l.Warn("reservation expired before commit", "lease", leaseID)
395
-
cleanup.run()
389
+
cleanup()
396
390
e.pushSnapshot()
397
391
}
398
392
399
-
func (e *Executor) takeUncommittedReservation(leaseID string, releaseSlot bool) (reservationCleanup, bool) {
393
+
func (e *Executor) takeUncommittedReservation(leaseID string, releaseSlot bool) (func(), bool) {
400
394
e.mu.Lock()
401
395
defer e.mu.Unlock()
402
396
res := e.active[leaseID]
403
397
if res == nil || res.committed {
404
-
return reservationCleanup{}, false
398
+
return func() {}, false
405
399
}
406
400
return e.removeReservationLocked(res, releaseSlot), true
407
401
}
408
402
409
-
func (e *Executor) finishReservation(res *reservation) (reservationCleanup, bool, bool) {
403
+
func (e *Executor) finishReservation(res *reservation) (func(), bool, bool) {
410
404
e.mu.Lock()
411
405
defer e.mu.Unlock()
412
406
if e.active[res.leaseID] != res {
413
-
return reservationCleanup{}, false, false
407
+
return func() {}, false, false
414
408
}
415
409
cancelled := res.cancelled
416
410
return e.removeReservationLocked(res, false), cancelled, true
417
411
}
418
412
419
-
func (e *Executor) removeReservationLocked(res *reservation, releaseSlot bool) reservationCleanup {
413
+
func (e *Executor) removeReservationLocked(res *reservation, releaseSlot bool) func() {
420
414
delete(e.active, res.leaseID)
421
-
if e.activeCount > 0 {
422
-
e.activeCount--
423
-
}
424
415
if res.ttlTimer != nil {
425
416
res.ttlTimer.Stop()
426
417
res.ttlTimer = nil
427
418
}
428
-
cleanup := reservationCleanup{stopTail: res.stopTail}
419
+
stopTail := res.stopTail
429
420
res.stopTail = nil
421
+
slot := res.slot
430
422
if releaseSlot {
431
-
cleanup.slot = res.slot
432
423
res.slot = nil
433
424
}
434
-
return cleanup
425
+
return func() {
426
+
if stopTail != nil {
427
+
stopTail()
428
+
}
429
+
if releaseSlot && slot != nil {
430
+
slot.Release()
431
+
}
432
+
}
433
+
}
434
+
435
+
func normalizeLabels(labels []string) []string {
436
+
seen := make(map[string]struct{}, len(labels))
437
+
out := make([]string, 0, len(labels))
438
+
for _, label := range labels {
439
+
label = strings.TrimSpace(label)
440
+
if label == "" {
441
+
continue
442
+
}
443
+
if _, ok := seen[label]; ok {
444
+
continue
445
+
}
446
+
seen[label] = struct{}{}
447
+
out = append(out, label)
448
+
}
449
+
return out
435
450
}
436
451
437
452
// --- snapshots -------------------------------------------------------------
···
458
473
459
474
func (e *Executor) pushSnapshot() {
460
475
e.mu.Lock()
461
-
free := uint32(0)
462
-
if !e.draining {
463
-
if n := e.seats - e.activeCount; n > 0 {
464
-
free = uint32(n)
465
-
}
466
-
}
476
+
draining := e.draining
477
+
active := len(e.active)
467
478
e.mu.Unlock()
468
479
469
-
engines := make(map[string]*millv1.EngineSnapshot, len(e.engines))
480
+
load := 0.0
481
+
if e.seats > 0 {
482
+
load = float64(active) / float64(e.seats)
483
+
}
484
+
available := !draining && active < e.seats
485
+
engines := make(map[string]*millv1.EngineAvailability, len(e.engines))
470
486
for name := range e.engines {
471
-
engines[name] = &millv1.EngineSnapshot{FreeSeats: free}
487
+
engines[name] = &millv1.EngineAvailability{
488
+
Available: available,
489
+
Load: map[string]float64{"slots": load},
490
+
}
472
491
}
473
492
e.send(&millproto.Message{NodeSnapshot: &millv1.NodeSnapshot{
474
493
NodeId: e.nodeID,
···
481
500
e.draining = true
482
501
e.mu.Unlock()
483
502
e.pushSnapshot()
484
-
}
485
-
486
-
func (e *Executor) bidScore() float64 {
487
-
e.mu.Lock()
488
-
defer e.mu.Unlock()
489
-
return float64(e.seats - e.activeCount)
490
503
}
491
504
492
505
func (e *Executor) engineNames() []string {
+2
-3
spindle/mill/executor/observe.go
+2
-3
spindle/mill/executor/observe.go
+2
-25
spindle/mill/executor/reserved.go
+2
-25
spindle/mill/executor/reserved.go
···
4
4
"context"
5
5
"fmt"
6
6
"sync"
7
-
"time"
8
7
9
-
"tangled.org/core/api/tangled"
10
8
"tangled.org/core/spindle/engine"
11
9
"tangled.org/core/spindle/models"
12
-
"tangled.org/core/spindle/secrets"
13
10
)
14
11
15
12
// reservedEngine wraps a real engine so that the slot acquired up-front during
···
19
16
//
20
17
// This is the only change to the execution path on an executor.
21
18
type reservedEngine struct {
22
-
inner models.Engine
19
+
models.Engine
23
20
slot engine.WorkflowSlot
24
21
once sync.Once
25
22
}
···
27
24
// newReservedEngine returns a wrapper around inner that hands back slot exactly
28
25
// once from AcquireWorkflowSlot.
29
26
func newReservedEngine(inner models.Engine, slot engine.WorkflowSlot) models.Engine {
30
-
return &reservedEngine{inner: inner, slot: slot}
31
-
}
32
-
33
-
func (e *reservedEngine) InitWorkflow(twf tangled.Pipeline_Workflow, tpl tangled.Pipeline) (*models.Workflow, error) {
34
-
return e.inner.InitWorkflow(twf, tpl)
35
-
}
36
-
37
-
func (e *reservedEngine) SetupWorkflow(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, wfLogger models.WorkflowLogger) error {
38
-
return e.inner.SetupWorkflow(ctx, wid, wf, wfLogger)
39
-
}
40
-
41
-
func (e *reservedEngine) WorkflowTimeout() time.Duration {
42
-
return e.inner.WorkflowTimeout()
43
-
}
44
-
45
-
func (e *reservedEngine) DestroyWorkflow(ctx context.Context, wid models.WorkflowId) error {
46
-
return e.inner.DestroyWorkflow(ctx, wid)
47
-
}
48
-
49
-
func (e *reservedEngine) RunStep(ctx context.Context, wid models.WorkflowId, w *models.Workflow, idx int, secrets []secrets.UnlockedSecret, wfLogger models.WorkflowLogger) error {
50
-
return e.inner.RunStep(ctx, wid, w, idx, secrets, wfLogger)
27
+
return &reservedEngine{Engine: inner, slot: slot}
51
28
}
52
29
53
30
// AcquireWorkflowSlot hands back the pre-acquired slot exactly once. The slot
+125
-44
spindle/mill/mill.go
+125
-44
spindle/mill/mill.go
···
6
6
"errors"
7
7
"fmt"
8
8
"log/slog"
9
-
"sort"
9
+
"slices"
10
+
"strings"
10
11
"sync"
11
12
"time"
12
13
···
272
273
return nil, err
273
274
}
274
275
275
-
candidates := m.rankCandidates(engineName)
276
+
requiredLabels := requiredLabels(wf)
277
+
candidates := m.rankCandidates(engineName, requiredLabels)
276
278
if len(candidates) == 0 {
277
279
return nil, nil
278
-
}
279
-
if len(candidates) > m.cfg.TopK {
280
-
candidates = candidates[:m.cfg.TopK]
281
280
}
282
281
283
282
bidCtx, cancel := context.WithTimeout(ctx, m.cfg.BidTimeout)
284
283
defer cancel()
285
284
286
285
type bidResult struct {
287
-
sess *millSession
288
-
lease *RemoteLease
289
-
score float64
286
+
sess *millSession
287
+
lease *RemoteLease
288
+
rank int
289
+
incompatible bool
290
+
reason string
290
291
}
291
292
results := make(chan bidResult, len(candidates))
292
-
var wg sync.WaitGroup
293
-
for _, sess := range candidates {
294
-
wg.Add(1)
295
-
go func(sess *millSession) {
296
-
defer wg.Done()
297
-
leaseID := m.nextLeaseID()
298
-
lease := newLease(leaseID, sess.nodeID, engineName)
299
-
wid := m.jobWid(wf)
300
-
msg := &millproto.Message{ReserveSeat: &millv1.ReserveSeat{
301
-
LeaseId: leaseID,
302
-
TargetEngine: engineName,
303
-
RawPipelineJson: rawPipeline,
304
-
RawWorkflowJson: rawWorkflow,
305
-
Knot: wid.Knot,
306
-
Rkey: wid.Rkey,
307
-
TtlSeconds: uint32(m.cfg.ReconnectGrace / time.Second),
308
-
}}
309
-
resp, err := sess.request(bidCtx, leaseID, msg)
310
-
if err != nil {
293
+
ask := func(rank int, sess *millSession) {
294
+
leaseID := m.nextLeaseID()
295
+
lease := newLease(leaseID, sess.nodeID, engineName)
296
+
wid := m.jobWid(wf)
297
+
msg := &millproto.Message{ReserveSeat: &millv1.ReserveSeat{
298
+
LeaseId: leaseID,
299
+
TargetEngine: engineName,
300
+
RawPipelineJson: rawPipeline,
301
+
RawWorkflowJson: rawWorkflow,
302
+
Knot: wid.Knot,
303
+
Rkey: wid.Rkey,
304
+
TtlSeconds: uint32(m.cfg.ReconnectGrace / time.Second),
305
+
}}
306
+
resp, err := sess.request(bidCtx, leaseID, msg)
307
+
if err != nil {
308
+
results <- bidResult{}
309
+
return
310
+
}
311
+
rr := resp.GetReserveResult()
312
+
if rr == nil {
313
+
results <- bidResult{}
314
+
return
315
+
}
316
+
if !rr.GetAccepted() {
317
+
if rr.GetRejectClass() == millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE {
318
+
results <- bidResult{sess: sess, rank: rank, incompatible: true, reason: rr.GetRejectReason()}
311
319
return
312
320
}
313
-
rr := resp.GetReserveResult()
314
-
if rr == nil || !rr.GetAccepted() {
315
-
return
316
-
}
317
-
results <- bidResult{sess: sess, lease: lease, score: rr.GetScore()}
318
-
}(sess)
321
+
results <- bidResult{}
322
+
return
323
+
}
324
+
results <- bidResult{sess: sess, lease: lease, rank: rank}
319
325
}
320
-
wg.Wait()
321
-
close(results)
326
+
window := m.cfg.TopK
327
+
if window <= 0 {
328
+
window = len(candidates)
329
+
}
330
+
next := 0
331
+
inFlight := 0
332
+
for next < len(candidates) && inFlight < window {
333
+
inFlight++
334
+
go ask(next, candidates[next])
335
+
next++
336
+
}
322
337
323
338
var winner *bidResult
324
339
var losers []*RemoteLease
325
-
for r := range results {
326
-
if winner == nil || r.score > winner.score {
340
+
var incompatible []string
341
+
softReject := false
342
+
for inFlight > 0 {
343
+
r := <-results
344
+
inFlight--
345
+
if r.incompatible {
346
+
if r.reason != "" {
347
+
incompatible = append(incompatible, r.reason)
348
+
}
349
+
} else if r.lease == nil {
350
+
softReject = true
351
+
}
352
+
if r.lease == nil {
353
+
for winner == nil && next < len(candidates) && inFlight < window {
354
+
inFlight++
355
+
go ask(next, candidates[next])
356
+
next++
357
+
}
358
+
continue
359
+
}
360
+
if winner == nil || r.rank < winner.rank {
327
361
if winner != nil {
328
362
losers = append(losers, winner.lease)
329
363
}
···
339
373
}
340
374
341
375
if winner == nil {
376
+
if len(incompatible) > 0 && !softReject {
377
+
return nil, fmt.Errorf("no compatible executor for %s: %s", engineName, strings.Join(incompatible, "; "))
378
+
}
342
379
return nil, nil
343
380
}
344
381
return winner.lease, nil
345
382
}
346
383
347
-
func (m *Mill) rankCandidates(engineName string) []*millSession {
384
+
func (m *Mill) rankCandidates(engineName string, requiredLabels []string) []*millSession {
348
385
m.mu.Lock()
349
386
defer m.mu.Unlock()
350
387
351
388
type ranked struct {
352
389
sess *millSession
353
-
seats uint32
390
+
worst float64
391
+
sum float64
354
392
}
355
393
var rs []ranked
356
394
for _, s := range m.sessions {
···
360
398
if s.snapshot == nil {
361
399
continue
362
400
}
363
-
es, ok := s.snapshot.GetEngines()[engineName]
364
-
if !ok || es.GetFreeSeats() == 0 {
401
+
ea, ok := s.snapshot.GetEngines()[engineName]
402
+
if !ok || !ea.GetAvailable() {
365
403
continue
366
404
}
367
-
rs = append(rs, ranked{sess: s, seats: es.GetFreeSeats()})
405
+
if !hasLabels(s.labels, requiredLabels) {
406
+
continue
407
+
}
408
+
worst, sum := loadScore(ea.GetLoad())
409
+
rs = append(rs, ranked{sess: s, worst: worst, sum: sum})
368
410
}
369
-
sort.SliceStable(rs, func(i, j int) bool { return rs[i].seats > rs[j].seats })
411
+
slices.SortStableFunc(rs, func(a, b ranked) int {
412
+
if a.worst < b.worst {
413
+
return -1
414
+
}
415
+
if a.worst > b.worst {
416
+
return 1
417
+
}
418
+
if a.sum < b.sum {
419
+
return -1
420
+
}
421
+
if a.sum > b.sum {
422
+
return 1
423
+
}
424
+
return 0
425
+
})
370
426
371
427
out := make([]*millSession, len(rs))
372
428
for i := range rs {
···
375
431
return out
376
432
}
377
433
378
-
// --- commit + wait (RunStep) ----------------------------------------------
434
+
func loadScore(load map[string]float64) (worst, sum float64) {
435
+
for _, v := range load {
436
+
if v > worst {
437
+
worst = v
438
+
}
439
+
sum += v
440
+
}
441
+
return worst, sum
442
+
}
443
+
444
+
func requiredLabels(wf *models.Workflow) []string {
445
+
st, ok := wf.Data.(*millWorkflowState)
446
+
if !ok || st == nil {
447
+
return nil
448
+
}
449
+
return st.RawWorkflow.RunsOn
450
+
}
451
+
452
+
func hasLabels(labels []string, required []string) bool {
453
+
for _, want := range required {
454
+
if !slices.Contains(labels, want) {
455
+
return false
456
+
}
457
+
}
458
+
return true
459
+
}
379
460
380
461
func (m *Mill) commitAndWait(ctx context.Context, wf *models.Workflow, unlocked []secrets.UnlockedSecret) error {
381
462
st, ok := wf.Data.(*millWorkflowState)
+292
spindle/mill/mill_test.go
+292
spindle/mill/mill_test.go
···
5
5
"errors"
6
6
"io"
7
7
"log/slog"
8
+
"strings"
8
9
"testing"
9
10
"time"
10
11
···
33
34
}
34
35
}
35
36
37
+
func testWorkflowWithRunsOn(name string, runsOn []string) *models.Workflow {
38
+
wf := testWorkflow(name)
39
+
wf.Data.(*millWorkflowState).RawWorkflow.RunsOn = runsOn
40
+
return wf
41
+
}
42
+
43
+
func addCandidateSession(t *testing.T, m *Mill, nodeID string, labels []string, load float64, enc messageEncoder) *millSession {
44
+
t.Helper()
45
+
if enc == nil {
46
+
enc = scriptedEncoder(func(*millproto.Message) error { return nil })
47
+
}
48
+
sess := newSession(nodeID, enc, slog.New(slog.NewTextHandler(io.Discard, nil)))
49
+
sess.labels = labels
50
+
sess.snapshot = &millv1.NodeSnapshot{
51
+
NodeId: nodeID,
52
+
Engines: map[string]*millv1.EngineAvailability{
53
+
"dummy": {Available: load < 1.0, Load: map[string]float64{"slots": load}},
54
+
},
55
+
}
56
+
m.mu.Lock()
57
+
m.sessions[nodeID] = sess
58
+
m.mu.Unlock()
59
+
return sess
60
+
}
61
+
62
+
func assertRankedNodes(t *testing.T, got []*millSession, want []string) {
63
+
t.Helper()
64
+
if len(got) != len(want) {
65
+
t.Fatalf("rankCandidates() returned %d candidates, want %d: got %v want %v", len(got), len(want), sessionIDs(got), want)
66
+
}
67
+
for i := range want {
68
+
if got[i].nodeID != want[i] {
69
+
t.Fatalf("rankCandidates()[%d] = %q, want %q; full order got %v want %v", i, got[i].nodeID, want[i], sessionIDs(got), want)
70
+
}
71
+
}
72
+
}
73
+
74
+
func sessionIDs(sessions []*millSession) []string {
75
+
out := make([]string, len(sessions))
76
+
for i, sess := range sessions {
77
+
out[i] = sess.nodeID
78
+
}
79
+
return out
80
+
}
81
+
82
+
func sameStringMultiset(a, b []string) bool {
83
+
if len(a) != len(b) {
84
+
return false
85
+
}
86
+
counts := make(map[string]int, len(a))
87
+
for _, s := range a {
88
+
counts[s]++
89
+
}
90
+
for _, s := range b {
91
+
if counts[s] == 0 {
92
+
return false
93
+
}
94
+
counts[s]--
95
+
}
96
+
return true
97
+
}
98
+
99
+
type reserveReply struct {
100
+
accepted bool
101
+
rejectClass millv1.RejectClass
102
+
reason string
103
+
}
104
+
105
+
func addReplyingCandidateSession(t *testing.T, m *Mill, nodeID string, labels []string, load float64, asked chan<- string, reply reserveReply) *millSession {
106
+
t.Helper()
107
+
var sess *millSession
108
+
sess = addCandidateSession(t, m, nodeID, labels, load, scriptedEncoder(func(msg *millproto.Message) error {
109
+
rs := msg.GetReserveSeat()
110
+
if rs == nil {
111
+
return nil
112
+
}
113
+
if asked != nil {
114
+
asked <- nodeID
115
+
}
116
+
sess.deliver(rs.GetLeaseId(), &millproto.Message{ReserveResult: &millv1.ReserveResult{
117
+
LeaseId: rs.GetLeaseId(),
118
+
Accepted: reply.accepted,
119
+
RejectReason: reply.reason,
120
+
RejectClass: reply.rejectClass,
121
+
}})
122
+
return nil
123
+
}))
124
+
return sess
125
+
}
126
+
127
+
func drainAsked(ch <-chan string) []string {
128
+
var out []string
129
+
for {
130
+
select {
131
+
case nodeID := <-ch:
132
+
out = append(out, nodeID)
133
+
default:
134
+
return out
135
+
}
136
+
}
137
+
}
138
+
36
139
func TestCommitRetriesAfterSessionCloseBeforeCommitted(t *testing.T) {
37
140
l := slog.New(slog.NewTextHandler(io.Discard, nil))
38
141
m := New(l, Config{BidTimeout: 25 * time.Millisecond, ReconnectGrace: time.Second})
···
135
238
_, err := m.place(ctx, "dummy", wid, wf)
136
239
if err != context.DeadlineExceeded {
137
240
t.Fatalf("place() error = %v, want DeadlineExceeded", err)
241
+
}
242
+
}
243
+
244
+
func TestRankCandidatesFiltersRequiredLabelsWithANDSemantics(t *testing.T) {
245
+
m := New(slog.New(slog.NewTextHandler(io.Discard, nil)), Config{})
246
+
addCandidateSession(t, m, "linux-high", []string{"linux"}, 0.0, nil)
247
+
addCandidateSession(t, m, "linux-arm", []string{"linux", "arm64"}, 0.25, nil)
248
+
addCandidateSession(t, m, "unlabeled", nil, 0.5, nil)
249
+
addCandidateSession(t, m, "linux-arm-gpu", []string{"linux", "arm64", "gpu"}, 0.75, nil)
250
+
addCandidateSession(t, m, "linux-arm-full", []string{"linux", "arm64"}, 1.0, nil)
251
+
252
+
tests := []struct {
253
+
name string
254
+
requiredLabels []string
255
+
want []string
256
+
}{
257
+
{
258
+
name: "no required labels keeps old capacity ranking",
259
+
want: []string{"linux-high", "linux-arm", "unlabeled", "linux-arm-gpu"},
260
+
},
261
+
{
262
+
name: "single required label includes every candidate carrying it",
263
+
requiredLabels: []string{"linux"},
264
+
want: []string{"linux-high", "linux-arm", "linux-arm-gpu"},
265
+
},
266
+
{
267
+
name: "all required labels must be present",
268
+
requiredLabels: []string{"linux", "arm64"},
269
+
want: []string{"linux-arm", "linux-arm-gpu"},
270
+
},
271
+
{
272
+
name: "one missing required label excludes the candidate",
273
+
requiredLabels: []string{"linux", "arm64", "gpu"},
274
+
want: []string{"linux-arm-gpu"},
275
+
},
276
+
{
277
+
name: "unknown required label leaves no candidate",
278
+
requiredLabels: []string{"linux", "arm64", "metal"},
279
+
},
280
+
}
281
+
282
+
for _, tt := range tests {
283
+
t.Run(tt.name, func(t *testing.T) {
284
+
assertRankedNodes(t, m.rankCandidates("dummy", tt.requiredLabels), tt.want)
285
+
})
286
+
}
287
+
}
288
+
289
+
func TestPlaceWithMissingRequiredLabelsStaysPendingWithoutReserve(t *testing.T) {
290
+
l := slog.New(slog.NewTextHandler(io.Discard, nil))
291
+
m := New(l, Config{BidTimeout: 10 * time.Millisecond})
292
+
reserveSent := make(chan struct{}, 1)
293
+
addCandidateSession(t, m, "linux-only", []string{"linux"}, 0.75, scriptedEncoder(func(msg *millproto.Message) error {
294
+
if msg.GetReserveSeat() != nil {
295
+
select {
296
+
case reserveSent <- struct{}{}:
297
+
default:
298
+
}
299
+
}
300
+
return nil
301
+
}))
302
+
wf := testWorkflowWithRunsOn("build", []string{"linux", "arm64"})
303
+
wid := models.WorkflowId{PipelineId: models.PipelineId{Knot: "k", Rkey: "r"}, Name: "build"}
304
+
305
+
ctx, cancel := context.WithTimeout(context.Background(), 120*time.Millisecond)
306
+
defer cancel()
307
+
_, err := m.place(ctx, "dummy", wid, wf)
308
+
if err != context.DeadlineExceeded {
309
+
t.Fatalf("place() error = %v, want DeadlineExceeded while job remains pending", err)
310
+
}
311
+
select {
312
+
case <-reserveSent:
313
+
t.Fatal("place() sent ReserveSeat to executor missing a required label")
314
+
default:
315
+
}
316
+
}
317
+
318
+
func TestPlaceKeepsAskingBelowTopKAfterIncompatibleRejects(t *testing.T) {
319
+
l := slog.New(slog.NewTextHandler(io.Discard, nil))
320
+
m := New(l, Config{TopK: 2, BidTimeout: time.Second})
321
+
asked := make(chan string, 4)
322
+
addReplyingCandidateSession(t, m, "wrong-label", []string{"linux"}, 0.0, asked, reserveReply{
323
+
accepted: false,
324
+
rejectClass: millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE,
325
+
reason: "wrong-label should not be asked",
326
+
})
327
+
addReplyingCandidateSession(t, m, "incompatible-a", []string{"linux", "arm64"}, 0.0, asked, reserveReply{
328
+
accepted: false,
329
+
rejectClass: millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE,
330
+
reason: "no runner",
331
+
})
332
+
addReplyingCandidateSession(t, m, "incompatible-b", []string{"linux", "arm64"}, 0.25, asked, reserveReply{
333
+
accepted: false,
334
+
rejectClass: millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE,
335
+
reason: "bad image",
336
+
})
337
+
addReplyingCandidateSession(t, m, "compatible-below-window", []string{"linux", "arm64"}, 0.5, asked, reserveReply{
338
+
accepted: true,
339
+
})
340
+
341
+
wf := testWorkflowWithRunsOn("build", []string{"linux", "arm64"})
342
+
wid := models.WorkflowId{PipelineId: models.PipelineId{Knot: "k", Rkey: "r"}, Name: "build"}
343
+
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
344
+
defer cancel()
345
+
slot, err := m.place(ctx, "dummy", wid, wf)
346
+
if err != nil {
347
+
t.Fatalf("place() error = %v, want compatible lower-ranked executor", err)
348
+
}
349
+
defer slot.Release()
350
+
351
+
ms, ok := slot.(*millSlot)
352
+
if !ok {
353
+
t.Fatalf("place() slot type = %T, want *millSlot", slot)
354
+
}
355
+
if ms.lease.nodeID != "compatible-below-window" {
356
+
t.Fatalf("place() chose node %q, want compatible-below-window", ms.lease.nodeID)
357
+
}
358
+
if got, want := drainAsked(asked), []string{"incompatible-a", "incompatible-b", "compatible-below-window"}; !sameStringMultiset(got, want) {
359
+
t.Fatalf("ReserveSeat asked nodes = %v, want %v", got, want)
360
+
}
361
+
}
362
+
363
+
func TestPlaceKeepsTransientOnlyRejectsPending(t *testing.T) {
364
+
l := slog.New(slog.NewTextHandler(io.Discard, nil))
365
+
m := New(l, Config{TopK: 1, BidTimeout: 10 * time.Millisecond})
366
+
asked := make(chan string, 2)
367
+
addReplyingCandidateSession(t, m, "busy-a", []string{"linux", "arm64"}, 0.5, asked, reserveReply{
368
+
accepted: false,
369
+
rejectClass: millv1.RejectClass_REJECT_CLASS_TRANSIENT,
370
+
reason: "draining",
371
+
})
372
+
addReplyingCandidateSession(t, m, "busy-b", []string{"linux", "arm64"}, 0.75, asked, reserveReply{
373
+
accepted: false,
374
+
rejectClass: millv1.RejectClass_REJECT_CLASS_TRANSIENT,
375
+
reason: "no slot",
376
+
})
377
+
378
+
wf := testWorkflowWithRunsOn("build", []string{"linux", "arm64"})
379
+
wid := models.WorkflowId{PipelineId: models.PipelineId{Knot: "k", Rkey: "r"}, Name: "build"}
380
+
ctx, cancel := context.WithTimeout(context.Background(), 120*time.Millisecond)
381
+
defer cancel()
382
+
_, err := m.place(ctx, "dummy", wid, wf)
383
+
if err != context.DeadlineExceeded {
384
+
t.Fatalf("place() error = %v, want DeadlineExceeded while transient rejects leave job pending", err)
385
+
}
386
+
if got, want := drainAsked(asked), []string{"busy-a", "busy-b"}; !sameStringMultiset(got, want) {
387
+
t.Fatalf("ReserveSeat asked nodes = %v, want %v", got, want)
388
+
}
389
+
}
390
+
391
+
func TestPlaceReportsOnlyEligiblePermanentIncompatibleRejects(t *testing.T) {
392
+
l := slog.New(slog.NewTextHandler(io.Discard, nil))
393
+
m := New(l, Config{TopK: 1, BidTimeout: time.Second})
394
+
asked := make(chan string, 3)
395
+
addReplyingCandidateSession(t, m, "wrong-label", []string{"linux"}, 0.25, asked, reserveReply{
396
+
accepted: false,
397
+
rejectClass: millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE,
398
+
reason: "wrong-label should not be asked",
399
+
})
400
+
addReplyingCandidateSession(t, m, "incompatible-a", []string{"linux", "arm64"}, 0.5, asked, reserveReply{
401
+
accepted: false,
402
+
rejectClass: millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE,
403
+
reason: "no qemu-system-aarch64",
404
+
})
405
+
addReplyingCandidateSession(t, m, "incompatible-b", []string{"linux", "arm64"}, 0.75, asked, reserveReply{
406
+
accepted: false,
407
+
rejectClass: millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE,
408
+
reason: "image arch unsupported",
409
+
})
410
+
411
+
wf := testWorkflowWithRunsOn("build", []string{"linux", "arm64"})
412
+
wid := models.WorkflowId{PipelineId: models.PipelineId{Knot: "k", Rkey: "r"}, Name: "build"}
413
+
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
414
+
defer cancel()
415
+
_, err := m.place(ctx, "dummy", wid, wf)
416
+
if err == nil {
417
+
t.Fatal("place() error = nil, want permanent incompatible failure")
418
+
}
419
+
errText := err.Error()
420
+
for _, want := range []string{"no compatible executor for dummy", "no qemu-system-aarch64", "image arch unsupported"} {
421
+
if !strings.Contains(errText, want) {
422
+
t.Fatalf("place() error = %q, want it to contain %q", errText, want)
423
+
}
424
+
}
425
+
if strings.Contains(errText, "wrong-label should not be asked") {
426
+
t.Fatalf("place() error = %q, included a missing-label candidate as incompatible", errText)
427
+
}
428
+
if got, want := drainAsked(asked), []string{"incompatible-a", "incompatible-b"}; !sameStringMultiset(got, want) {
429
+
t.Fatalf("ReserveSeat asked nodes = %v, want %v", got, want)
138
430
}
139
431
}
140
432
+163
-112
spindle/mill/proto/gen/mill.pb.go
+163
-112
spindle/mill/proto/gen/mill.pb.go
···
22
22
_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
23
23
)
24
24
25
+
type RejectClass int32
26
+
27
+
const (
28
+
RejectClass_REJECT_CLASS_UNSPECIFIED RejectClass = 0
29
+
RejectClass_REJECT_CLASS_TRANSIENT RejectClass = 1
30
+
RejectClass_REJECT_CLASS_INCOMPATIBLE RejectClass = 2
31
+
)
32
+
33
+
// Enum value maps for RejectClass.
34
+
var (
35
+
RejectClass_name = map[int32]string{
36
+
0: "REJECT_CLASS_UNSPECIFIED",
37
+
1: "REJECT_CLASS_TRANSIENT",
38
+
2: "REJECT_CLASS_INCOMPATIBLE",
39
+
}
40
+
RejectClass_value = map[string]int32{
41
+
"REJECT_CLASS_UNSPECIFIED": 0,
42
+
"REJECT_CLASS_TRANSIENT": 1,
43
+
"REJECT_CLASS_INCOMPATIBLE": 2,
44
+
}
45
+
)
46
+
47
+
func (x RejectClass) Enum() *RejectClass {
48
+
p := new(RejectClass)
49
+
*p = x
50
+
return p
51
+
}
52
+
53
+
func (x RejectClass) String() string {
54
+
return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x))
55
+
}
56
+
57
+
func (RejectClass) Descriptor() protoreflect.EnumDescriptor {
58
+
return file_spindle_mill_v1_mill_proto_enumTypes[0].Descriptor()
59
+
}
60
+
61
+
func (RejectClass) Type() protoreflect.EnumType {
62
+
return &file_spindle_mill_v1_mill_proto_enumTypes[0]
63
+
}
64
+
65
+
func (x RejectClass) Number() protoreflect.EnumNumber {
66
+
return protoreflect.EnumNumber(x)
67
+
}
68
+
69
+
// Deprecated: Use RejectClass.Descriptor instead.
70
+
func (RejectClass) EnumDescriptor() ([]byte, []int) {
71
+
return file_spindle_mill_v1_mill_proto_rawDescGZIP(), []int{0}
72
+
}
73
+
25
74
// Hello is the first frame an executor sends after dialing the mill. It
26
75
// carries the static traits of the node plus a resume hint.
27
76
type Hello struct {
···
32
81
NodeId string `protobuf:"bytes,2,opt,name=node_id,json=nodeId,proto3" json:"node_id,omitempty"`
33
82
// engine names this node can run ("microvm", "nixery").
34
83
Engines []string `protobuf:"bytes,3,rep,name=engines,proto3" json:"engines,omitempty"`
35
-
// GOARCH of the node, so the mill won't place arch-incompatible jobs.
84
+
// GOARCH of the node, retained as an informational trait for logs.
36
85
Arch string `protobuf:"bytes,4,opt,name=arch,proto3" json:"arch,omitempty"`
37
86
// the highest relay offset the executor believes it has sent; a resume hint.
38
-
LastOffset uint64 `protobuf:"varint,5,opt,name=last_offset,json=lastOffset,proto3" json:"last_offset,omitempty"`
87
+
LastOffset uint64 `protobuf:"varint,5,opt,name=last_offset,json=lastOffset,proto3" json:"last_offset,omitempty"`
88
+
// opaque operator-defined labels used for runs_on matching.
89
+
Labels []string `protobuf:"bytes,6,rep,name=labels,proto3" json:"labels,omitempty"`
39
90
unknownFields protoimpl.UnknownFields
40
91
sizeCache protoimpl.SizeCache
41
92
}
···
105
156
return 0
106
157
}
107
158
159
+
func (x *Hello) GetLabels() []string {
160
+
if x != nil {
161
+
return x.Labels
162
+
}
163
+
return nil
164
+
}
165
+
108
166
// Resume is the mill's reply to Hello. The executor replays every buffered
109
167
// relay entry with offset strictly greater than ack_offset before sending new
110
168
// ones.
···
152
210
return 0
153
211
}
154
212
155
-
// EngineSnapshot is the changing per-engine state on a node. free_seats is a
156
-
// coarse "can you take more" hint; 0 also means draining (the mill treats a
157
-
// draining node as a full one until it leaves). The resource fields are coarse
158
-
// budget headroom for ranking only; the executor makes the real yes/no call in
159
-
// ReserveResult.
160
-
type EngineSnapshot struct {
161
-
state protoimpl.MessageState `protogen:"open.v1"`
162
-
FreeSeats uint32 `protobuf:"varint,1,opt,name=free_seats,json=freeSeats,proto3" json:"free_seats,omitempty"`
163
-
FreeMemoryMib int64 `protobuf:"varint,2,opt,name=free_memory_mib,json=freeMemoryMib,proto3" json:"free_memory_mib,omitempty"`
164
-
FreeVcpus int64 `protobuf:"varint,3,opt,name=free_vcpus,json=freeVcpus,proto3" json:"free_vcpus,omitempty"`
165
-
FreeDiskMib int64 `protobuf:"varint,4,opt,name=free_disk_mib,json=freeDiskMib,proto3" json:"free_disk_mib,omitempty"`
213
+
// EngineAvailability is the changing per-engine state on a node. The executor
214
+
// reports opaque load metrics (higher = more loaded); the mill ranks by the
215
+
// worst load and uses the sum as a tie-breaker. The mill does not know what
216
+
// the metric keys mean.
217
+
type EngineAvailability struct {
218
+
state protoimpl.MessageState `protogen:"open.v1"`
219
+
Available bool `protobuf:"varint,1,opt,name=available,proto3" json:"available,omitempty"`
220
+
// engine-defined load metrics. keys are opaque to the mill.
221
+
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"`
166
222
unknownFields protoimpl.UnknownFields
167
223
sizeCache protoimpl.SizeCache
168
224
}
169
225
170
-
func (x *EngineSnapshot) Reset() {
171
-
*x = EngineSnapshot{}
226
+
func (x *EngineAvailability) Reset() {
227
+
*x = EngineAvailability{}
172
228
mi := &file_spindle_mill_v1_mill_proto_msgTypes[2]
173
229
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
174
230
ms.StoreMessageInfo(mi)
175
231
}
176
232
177
-
func (x *EngineSnapshot) String() string {
233
+
func (x *EngineAvailability) String() string {
178
234
return protoimpl.X.MessageStringOf(x)
179
235
}
180
236
181
-
func (*EngineSnapshot) ProtoMessage() {}
237
+
func (*EngineAvailability) ProtoMessage() {}
182
238
183
-
func (x *EngineSnapshot) ProtoReflect() protoreflect.Message {
239
+
func (x *EngineAvailability) ProtoReflect() protoreflect.Message {
184
240
mi := &file_spindle_mill_v1_mill_proto_msgTypes[2]
185
241
if x != nil {
186
242
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
···
192
248
return mi.MessageOf(x)
193
249
}
194
250
195
-
// Deprecated: Use EngineSnapshot.ProtoReflect.Descriptor instead.
196
-
func (*EngineSnapshot) Descriptor() ([]byte, []int) {
251
+
// Deprecated: Use EngineAvailability.ProtoReflect.Descriptor instead.
252
+
func (*EngineAvailability) Descriptor() ([]byte, []int) {
197
253
return file_spindle_mill_v1_mill_proto_rawDescGZIP(), []int{2}
198
254
}
199
255
200
-
func (x *EngineSnapshot) GetFreeSeats() uint32 {
201
-
if x != nil {
202
-
return x.FreeSeats
203
-
}
204
-
return 0
205
-
}
206
-
207
-
func (x *EngineSnapshot) GetFreeMemoryMib() int64 {
208
-
if x != nil {
209
-
return x.FreeMemoryMib
210
-
}
211
-
return 0
212
-
}
213
-
214
-
func (x *EngineSnapshot) GetFreeVcpus() int64 {
256
+
func (x *EngineAvailability) GetAvailable() bool {
215
257
if x != nil {
216
-
return x.FreeVcpus
258
+
return x.Available
217
259
}
218
-
return 0
260
+
return false
219
261
}
220
262
221
-
func (x *EngineSnapshot) GetFreeDiskMib() int64 {
263
+
func (x *EngineAvailability) GetLoad() map[string]float64 {
222
264
if x != nil {
223
-
return x.FreeDiskMib
265
+
return x.Load
224
266
}
225
-
return 0
267
+
return nil
226
268
}
227
269
228
270
// NodeSnapshot is pushed on connect, periodically, and right after any state
229
271
// change (reserve, commit, terminal).
230
272
type NodeSnapshot struct {
231
-
state protoimpl.MessageState `protogen:"open.v1"`
232
-
NodeId string `protobuf:"bytes,1,opt,name=node_id,json=nodeId,proto3" json:"node_id,omitempty"`
233
-
Seq uint64 `protobuf:"varint,2,opt,name=seq,proto3" json:"seq,omitempty"`
234
-
Engines map[string]*EngineSnapshot `protobuf:"bytes,3,rep,name=engines,proto3" json:"engines,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"`
273
+
state protoimpl.MessageState `protogen:"open.v1"`
274
+
NodeId string `protobuf:"bytes,1,opt,name=node_id,json=nodeId,proto3" json:"node_id,omitempty"`
275
+
Seq uint64 `protobuf:"varint,2,opt,name=seq,proto3" json:"seq,omitempty"`
276
+
Engines map[string]*EngineAvailability `protobuf:"bytes,3,rep,name=engines,proto3" json:"engines,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"`
235
277
unknownFields protoimpl.UnknownFields
236
278
sizeCache protoimpl.SizeCache
237
279
}
···
280
322
return 0
281
323
}
282
324
283
-
func (x *NodeSnapshot) GetEngines() map[string]*EngineSnapshot {
325
+
func (x *NodeSnapshot) GetEngines() map[string]*EngineAvailability {
284
326
if x != nil {
285
327
return x.Engines
286
328
}
···
387
429
388
430
// ReserveResult is the executor's accept/reject for a ReserveSeat.
389
431
type ReserveResult struct {
390
-
state protoimpl.MessageState `protogen:"open.v1"`
391
-
LeaseId string `protobuf:"bytes,1,opt,name=lease_id,json=leaseId,proto3" json:"lease_id,omitempty"`
392
-
Accepted bool `protobuf:"varint,2,opt,name=accepted,proto3" json:"accepted,omitempty"`
393
-
RejectReason string `protobuf:"bytes,3,opt,name=reject_reason,json=rejectReason,proto3" json:"reject_reason,omitempty"`
394
-
// optional bid score the mill ranks accepted leases by (higher is better).
395
-
Score float64 `protobuf:"fixed64,4,opt,name=score,proto3" json:"score,omitempty"`
432
+
state protoimpl.MessageState `protogen:"open.v1"`
433
+
LeaseId string `protobuf:"bytes,1,opt,name=lease_id,json=leaseId,proto3" json:"lease_id,omitempty"`
434
+
Accepted bool `protobuf:"varint,2,opt,name=accepted,proto3" json:"accepted,omitempty"`
435
+
RejectReason string `protobuf:"bytes,3,opt,name=reject_reason,json=rejectReason,proto3" json:"reject_reason,omitempty"`
436
+
RejectClass RejectClass `protobuf:"varint,4,opt,name=reject_class,json=rejectClass,proto3,enum=spindle.mill.v1.RejectClass" json:"reject_class,omitempty"`
396
437
unknownFields protoimpl.UnknownFields
397
438
sizeCache protoimpl.SizeCache
398
439
}
···
448
489
return ""
449
490
}
450
491
451
-
func (x *ReserveResult) GetScore() float64 {
492
+
func (x *ReserveResult) GetRejectClass() RejectClass {
452
493
if x != nil {
453
-
return x.Score
494
+
return x.RejectClass
454
495
}
455
-
return 0
496
+
return RejectClass_REJECT_CLASS_UNSPECIFIED
456
497
}
457
498
458
499
// Secret is a single unlocked secret, sent only inside CommitLease.
···
1134
1175
1135
1176
const file_spindle_mill_v1_mill_proto_rawDesc = "" +
1136
1177
"\n" +
1137
-
"\x1aspindle/mill/v1/mill.proto\x12\x0fspindle.mill.v1\x1a\x1bbuf/validate/validate.proto\"\xa3\x01\n" +
1178
+
"\x1aspindle/mill/v1/mill.proto\x12\x0fspindle.mill.v1\x1a\x1bbuf/validate/validate.proto\"\xbb\x01\n" +
1138
1179
"\x05Hello\x12)\n" +
1139
1180
"\x10protocol_version\x18\x01 \x01(\rR\x0fprotocolVersion\x12 \n" +
1140
1181
"\anode_id\x18\x02 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\x06nodeId\x12\x18\n" +
1141
1182
"\aengines\x18\x03 \x03(\tR\aengines\x12\x12\n" +
1142
1183
"\x04arch\x18\x04 \x01(\tR\x04arch\x12\x1f\n" +
1143
1184
"\vlast_offset\x18\x05 \x01(\x04R\n" +
1144
-
"lastOffset\"'\n" +
1185
+
"lastOffset\x12\x16\n" +
1186
+
"\x06labels\x18\x06 \x03(\tR\x06labels\"'\n" +
1145
1187
"\x06Resume\x12\x1d\n" +
1146
1188
"\n" +
1147
-
"ack_offset\x18\x01 \x01(\x04R\tackOffset\"\x9a\x01\n" +
1148
-
"\x0eEngineSnapshot\x12\x1d\n" +
1149
-
"\n" +
1150
-
"free_seats\x18\x01 \x01(\rR\tfreeSeats\x12&\n" +
1151
-
"\x0ffree_memory_mib\x18\x02 \x01(\x03R\rfreeMemoryMib\x12\x1d\n" +
1152
-
"\n" +
1153
-
"free_vcpus\x18\x03 \x01(\x03R\tfreeVcpus\x12\"\n" +
1154
-
"\rfree_disk_mib\x18\x04 \x01(\x03R\vfreeDiskMib\"\xdc\x01\n" +
1189
+
"ack_offset\x18\x01 \x01(\x04R\tackOffset\"\xae\x01\n" +
1190
+
"\x12EngineAvailability\x12\x1c\n" +
1191
+
"\tavailable\x18\x01 \x01(\bR\tavailable\x12A\n" +
1192
+
"\x04load\x18\x02 \x03(\v2-.spindle.mill.v1.EngineAvailability.LoadEntryR\x04load\x1a7\n" +
1193
+
"\tLoadEntry\x12\x10\n" +
1194
+
"\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" +
1195
+
"\x05value\x18\x02 \x01(\x01R\x05value:\x028\x01\"\xe0\x01\n" +
1155
1196
"\fNodeSnapshot\x12\x17\n" +
1156
1197
"\anode_id\x18\x01 \x01(\tR\x06nodeId\x12\x10\n" +
1157
1198
"\x03seq\x18\x02 \x01(\x04R\x03seq\x12D\n" +
1158
-
"\aengines\x18\x03 \x03(\v2*.spindle.mill.v1.NodeSnapshot.EnginesEntryR\aengines\x1a[\n" +
1199
+
"\aengines\x18\x03 \x03(\v2*.spindle.mill.v1.NodeSnapshot.EnginesEntryR\aengines\x1a_\n" +
1159
1200
"\fEnginesEntry\x12\x10\n" +
1160
-
"\x03key\x18\x01 \x01(\tR\x03key\x125\n" +
1161
-
"\x05value\x18\x02 \x01(\v2\x1f.spindle.mill.v1.EngineSnapshotR\x05value:\x028\x01\"\x80\x02\n" +
1201
+
"\x03key\x18\x01 \x01(\tR\x03key\x129\n" +
1202
+
"\x05value\x18\x02 \x01(\v2#.spindle.mill.v1.EngineAvailabilityR\x05value:\x028\x01\"\x80\x02\n" +
1162
1203
"\vReserveSeat\x12\"\n" +
1163
1204
"\blease_id\x18\x01 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\aleaseId\x12,\n" +
1164
1205
"\rtarget_engine\x18\x02 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\ftargetEngine\x12*\n" +
···
1167
1208
"\x04knot\x18\x05 \x01(\tR\x04knot\x12\x12\n" +
1168
1209
"\x04rkey\x18\x06 \x01(\tR\x04rkey\x12\x1f\n" +
1169
1210
"\vttl_seconds\x18\a \x01(\rR\n" +
1170
-
"ttlSeconds\"\x8a\x01\n" +
1211
+
"ttlSeconds\"\xb5\x01\n" +
1171
1212
"\rReserveResult\x12\"\n" +
1172
1213
"\blease_id\x18\x01 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\aleaseId\x12\x1a\n" +
1173
1214
"\baccepted\x18\x02 \x01(\bR\baccepted\x12#\n" +
1174
-
"\rreject_reason\x18\x03 \x01(\tR\frejectReason\x12\x14\n" +
1175
-
"\x05score\x18\x04 \x01(\x01R\x05score\"0\n" +
1215
+
"\rreject_reason\x18\x03 \x01(\tR\frejectReason\x12?\n" +
1216
+
"\freject_class\x18\x04 \x01(\x0e2\x1c.spindle.mill.v1.RejectClassR\vrejectClass\"0\n" +
1176
1217
"\x06Secret\x12\x10\n" +
1177
1218
"\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" +
1178
1219
"\x05value\x18\x02 \x01(\tR\x05value\"d\n" +
···
1234
1275
"\fstatus_event\n" +
1235
1276
"\blog_line\n" +
1236
1277
"\x0eattempt_result\n" +
1237
-
"\x03ack\x10\x01B0Z.tangled.org/core/spindle/mill/proto/gen;millv1b\x06proto3"
1278
+
"\x03ack\x10\x01*f\n" +
1279
+
"\vRejectClass\x12\x1c\n" +
1280
+
"\x18REJECT_CLASS_UNSPECIFIED\x10\x00\x12\x1a\n" +
1281
+
"\x16REJECT_CLASS_TRANSIENT\x10\x01\x12\x1d\n" +
1282
+
"\x19REJECT_CLASS_INCOMPATIBLE\x10\x02B0Z.tangled.org/core/spindle/mill/proto/gen;millv1b\x06proto3"
1238
1283
1239
1284
var (
1240
1285
file_spindle_mill_v1_mill_proto_rawDescOnce sync.Once
···
1248
1293
return file_spindle_mill_v1_mill_proto_rawDescData
1249
1294
}
1250
1295
1251
-
var file_spindle_mill_v1_mill_proto_msgTypes = make([]protoimpl.MessageInfo, 17)
1296
+
var file_spindle_mill_v1_mill_proto_enumTypes = make([]protoimpl.EnumInfo, 1)
1297
+
var file_spindle_mill_v1_mill_proto_msgTypes = make([]protoimpl.MessageInfo, 18)
1252
1298
var file_spindle_mill_v1_mill_proto_goTypes = []any{
1253
-
(*Hello)(nil), // 0: spindle.mill.v1.Hello
1254
-
(*Resume)(nil), // 1: spindle.mill.v1.Resume
1255
-
(*EngineSnapshot)(nil), // 2: spindle.mill.v1.EngineSnapshot
1256
-
(*NodeSnapshot)(nil), // 3: spindle.mill.v1.NodeSnapshot
1257
-
(*ReserveSeat)(nil), // 4: spindle.mill.v1.ReserveSeat
1258
-
(*ReserveResult)(nil), // 5: spindle.mill.v1.ReserveResult
1259
-
(*Secret)(nil), // 6: spindle.mill.v1.Secret
1260
-
(*CommitLease)(nil), // 7: spindle.mill.v1.CommitLease
1261
-
(*Committed)(nil), // 8: spindle.mill.v1.Committed
1262
-
(*ReleaseLease)(nil), // 9: spindle.mill.v1.ReleaseLease
1263
-
(*CancelAttempt)(nil), // 10: spindle.mill.v1.CancelAttempt
1264
-
(*StatusEvent)(nil), // 11: spindle.mill.v1.StatusEvent
1265
-
(*LogLine)(nil), // 12: spindle.mill.v1.LogLine
1266
-
(*AttemptResult)(nil), // 13: spindle.mill.v1.AttemptResult
1267
-
(*Ack)(nil), // 14: spindle.mill.v1.Ack
1268
-
(*Message)(nil), // 15: spindle.mill.v1.Message
1269
-
nil, // 16: spindle.mill.v1.NodeSnapshot.EnginesEntry
1299
+
(RejectClass)(0), // 0: spindle.mill.v1.RejectClass
1300
+
(*Hello)(nil), // 1: spindle.mill.v1.Hello
1301
+
(*Resume)(nil), // 2: spindle.mill.v1.Resume
1302
+
(*EngineAvailability)(nil), // 3: spindle.mill.v1.EngineAvailability
1303
+
(*NodeSnapshot)(nil), // 4: spindle.mill.v1.NodeSnapshot
1304
+
(*ReserveSeat)(nil), // 5: spindle.mill.v1.ReserveSeat
1305
+
(*ReserveResult)(nil), // 6: spindle.mill.v1.ReserveResult
1306
+
(*Secret)(nil), // 7: spindle.mill.v1.Secret
1307
+
(*CommitLease)(nil), // 8: spindle.mill.v1.CommitLease
1308
+
(*Committed)(nil), // 9: spindle.mill.v1.Committed
1309
+
(*ReleaseLease)(nil), // 10: spindle.mill.v1.ReleaseLease
1310
+
(*CancelAttempt)(nil), // 11: spindle.mill.v1.CancelAttempt
1311
+
(*StatusEvent)(nil), // 12: spindle.mill.v1.StatusEvent
1312
+
(*LogLine)(nil), // 13: spindle.mill.v1.LogLine
1313
+
(*AttemptResult)(nil), // 14: spindle.mill.v1.AttemptResult
1314
+
(*Ack)(nil), // 15: spindle.mill.v1.Ack
1315
+
(*Message)(nil), // 16: spindle.mill.v1.Message
1316
+
nil, // 17: spindle.mill.v1.EngineAvailability.LoadEntry
1317
+
nil, // 18: spindle.mill.v1.NodeSnapshot.EnginesEntry
1270
1318
}
1271
1319
var file_spindle_mill_v1_mill_proto_depIdxs = []int32{
1272
-
16, // 0: spindle.mill.v1.NodeSnapshot.engines:type_name -> spindle.mill.v1.NodeSnapshot.EnginesEntry
1273
-
6, // 1: spindle.mill.v1.CommitLease.secrets:type_name -> spindle.mill.v1.Secret
1274
-
0, // 2: spindle.mill.v1.Message.hello:type_name -> spindle.mill.v1.Hello
1275
-
1, // 3: spindle.mill.v1.Message.resume:type_name -> spindle.mill.v1.Resume
1276
-
3, // 4: spindle.mill.v1.Message.node_snapshot:type_name -> spindle.mill.v1.NodeSnapshot
1277
-
4, // 5: spindle.mill.v1.Message.reserve_seat:type_name -> spindle.mill.v1.ReserveSeat
1278
-
5, // 6: spindle.mill.v1.Message.reserve_result:type_name -> spindle.mill.v1.ReserveResult
1279
-
7, // 7: spindle.mill.v1.Message.commit_lease:type_name -> spindle.mill.v1.CommitLease
1280
-
8, // 8: spindle.mill.v1.Message.committed:type_name -> spindle.mill.v1.Committed
1281
-
9, // 9: spindle.mill.v1.Message.release_lease:type_name -> spindle.mill.v1.ReleaseLease
1282
-
10, // 10: spindle.mill.v1.Message.cancel_attempt:type_name -> spindle.mill.v1.CancelAttempt
1283
-
11, // 11: spindle.mill.v1.Message.status_event:type_name -> spindle.mill.v1.StatusEvent
1284
-
12, // 12: spindle.mill.v1.Message.log_line:type_name -> spindle.mill.v1.LogLine
1285
-
13, // 13: spindle.mill.v1.Message.attempt_result:type_name -> spindle.mill.v1.AttemptResult
1286
-
14, // 14: spindle.mill.v1.Message.ack:type_name -> spindle.mill.v1.Ack
1287
-
2, // 15: spindle.mill.v1.NodeSnapshot.EnginesEntry.value:type_name -> spindle.mill.v1.EngineSnapshot
1288
-
16, // [16:16] is the sub-list for method output_type
1289
-
16, // [16:16] is the sub-list for method input_type
1290
-
16, // [16:16] is the sub-list for extension type_name
1291
-
16, // [16:16] is the sub-list for extension extendee
1292
-
0, // [0:16] is the sub-list for field type_name
1320
+
17, // 0: spindle.mill.v1.EngineAvailability.load:type_name -> spindle.mill.v1.EngineAvailability.LoadEntry
1321
+
18, // 1: spindle.mill.v1.NodeSnapshot.engines:type_name -> spindle.mill.v1.NodeSnapshot.EnginesEntry
1322
+
0, // 2: spindle.mill.v1.ReserveResult.reject_class:type_name -> spindle.mill.v1.RejectClass
1323
+
7, // 3: spindle.mill.v1.CommitLease.secrets:type_name -> spindle.mill.v1.Secret
1324
+
1, // 4: spindle.mill.v1.Message.hello:type_name -> spindle.mill.v1.Hello
1325
+
2, // 5: spindle.mill.v1.Message.resume:type_name -> spindle.mill.v1.Resume
1326
+
4, // 6: spindle.mill.v1.Message.node_snapshot:type_name -> spindle.mill.v1.NodeSnapshot
1327
+
5, // 7: spindle.mill.v1.Message.reserve_seat:type_name -> spindle.mill.v1.ReserveSeat
1328
+
6, // 8: spindle.mill.v1.Message.reserve_result:type_name -> spindle.mill.v1.ReserveResult
1329
+
8, // 9: spindle.mill.v1.Message.commit_lease:type_name -> spindle.mill.v1.CommitLease
1330
+
9, // 10: spindle.mill.v1.Message.committed:type_name -> spindle.mill.v1.Committed
1331
+
10, // 11: spindle.mill.v1.Message.release_lease:type_name -> spindle.mill.v1.ReleaseLease
1332
+
11, // 12: spindle.mill.v1.Message.cancel_attempt:type_name -> spindle.mill.v1.CancelAttempt
1333
+
12, // 13: spindle.mill.v1.Message.status_event:type_name -> spindle.mill.v1.StatusEvent
1334
+
13, // 14: spindle.mill.v1.Message.log_line:type_name -> spindle.mill.v1.LogLine
1335
+
14, // 15: spindle.mill.v1.Message.attempt_result:type_name -> spindle.mill.v1.AttemptResult
1336
+
15, // 16: spindle.mill.v1.Message.ack:type_name -> spindle.mill.v1.Ack
1337
+
3, // 17: spindle.mill.v1.NodeSnapshot.EnginesEntry.value:type_name -> spindle.mill.v1.EngineAvailability
1338
+
18, // [18:18] is the sub-list for method output_type
1339
+
18, // [18:18] is the sub-list for method input_type
1340
+
18, // [18:18] is the sub-list for extension type_name
1341
+
18, // [18:18] is the sub-list for extension extendee
1342
+
0, // [0:18] is the sub-list for field type_name
1293
1343
}
1294
1344
1295
1345
func init() { file_spindle_mill_v1_mill_proto_init() }
···
1302
1352
File: protoimpl.DescBuilder{
1303
1353
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
1304
1354
RawDescriptor: unsafe.Slice(unsafe.StringData(file_spindle_mill_v1_mill_proto_rawDesc), len(file_spindle_mill_v1_mill_proto_rawDesc)),
1305
-
NumEnums: 0,
1306
-
NumMessages: 17,
1355
+
NumEnums: 1,
1356
+
NumMessages: 18,
1307
1357
NumExtensions: 0,
1308
1358
NumServices: 0,
1309
1359
},
1310
1360
GoTypes: file_spindle_mill_v1_mill_proto_goTypes,
1311
1361
DependencyIndexes: file_spindle_mill_v1_mill_proto_depIdxs,
1362
+
EnumInfos: file_spindle_mill_v1_mill_proto_enumTypes,
1312
1363
MessageInfos: file_spindle_mill_v1_mill_proto_msgTypes,
1313
1364
}.Build()
1314
1365
File_spindle_mill_v1_mill_proto = out.File
+19
-14
spindle/mill/proto/spindle/mill/v1/mill.proto
+19
-14
spindle/mill/proto/spindle/mill/v1/mill.proto
···
15
15
string node_id = 2 [(buf.validate.field).string.min_len = 1];
16
16
// engine names this node can run ("microvm", "nixery").
17
17
repeated string engines = 3;
18
-
// GOARCH of the node, so the mill won't place arch-incompatible jobs.
18
+
// GOARCH of the node, retained as an informational trait for logs.
19
19
string arch = 4;
20
20
// the highest relay offset the executor believes it has sent; a resume hint.
21
21
uint64 last_offset = 5;
22
+
// opaque operator-defined labels used for runs_on matching.
23
+
repeated string labels = 6;
22
24
}
23
25
24
26
// Resume is the mill's reply to Hello. The executor replays every buffered
···
28
30
uint64 ack_offset = 1;
29
31
}
30
32
31
-
// EngineSnapshot is the changing per-engine state on a node. free_seats is a
32
-
// coarse "can you take more" hint; 0 also means draining (the mill treats a
33
-
// draining node as a full one until it leaves). The resource fields are coarse
34
-
// budget headroom for ranking only; the executor makes the real yes/no call in
35
-
// ReserveResult.
36
-
message EngineSnapshot {
37
-
uint32 free_seats = 1;
38
-
int64 free_memory_mib = 2;
39
-
int64 free_vcpus = 3;
40
-
int64 free_disk_mib = 4;
33
+
// EngineAvailability is the changing per-engine state on a node. The executor
34
+
// reports opaque load metrics (higher = more loaded); the mill ranks by the
35
+
// worst load and uses the sum as a tie-breaker. The mill does not know what
36
+
// the metric keys mean.
37
+
message EngineAvailability {
38
+
bool available = 1;
39
+
// engine-defined load metrics. keys are opaque to the mill.
40
+
map<string, double> load = 2;
41
41
}
42
42
43
43
// NodeSnapshot is pushed on connect, periodically, and right after any state
···
45
45
message NodeSnapshot {
46
46
string node_id = 1;
47
47
uint64 seq = 2;
48
-
map<string, EngineSnapshot> engines = 3;
48
+
map<string, EngineAvailability> engines = 3;
49
49
}
50
50
51
51
// ReserveSeat asks an executor to hold a seat for a job. Zero secrets ride this
···
64
64
uint32 ttl_seconds = 7;
65
65
}
66
66
67
+
enum RejectClass {
68
+
REJECT_CLASS_UNSPECIFIED = 0;
69
+
REJECT_CLASS_TRANSIENT = 1;
70
+
REJECT_CLASS_INCOMPATIBLE = 2;
71
+
}
72
+
67
73
// ReserveResult is the executor's accept/reject for a ReserveSeat.
68
74
message ReserveResult {
69
75
string lease_id = 1 [(buf.validate.field).string.min_len = 1];
70
76
bool accepted = 2;
71
77
string reject_reason = 3;
72
-
// optional bid score the mill ranks accepted leases by (higher is better).
73
-
double score = 4;
78
+
RejectClass reject_class = 4;
74
79
}
75
80
76
81
// Secret is a single unlocked secret, sent only inside CommitLease.