openvibers/openvibe.link Commits
S1: Extend protocol.WorkerCapabilities and populate it from the worker… (#14)
Changes
@@@ -237,8 +237,10 @@ A Node shutting down sends `job_exit` (`stopped`) also for a job acked but not s`job_cancel` for an id that is not running (unknown, refused or ended) and `job_exit_ack` for an unknown id areignored, with no answer: an `ack` keyed by a job id means the job was accepted.--A Node that runs jobs lists its classes in `status.capabilities.worker` as `{"runtime_classes": ["function"]}`; the--key is absent while it runs none, and `worker` is reserved as a plugin name.++A Node that runs jobs lists its classes and effective local worker caps in `status.capabilities.worker`:++`{"runtime_classes":["function"],"max_jobs":1,"max_ttl_ms":600000,"max_wall_ms":300000,"max_cpu_ms":300000,"max_mem_bytes":536870912,"max_output_bytes":1048576}`.++The limits are `worker.caps` with defaults applied; the key is absent while the worker is off, and `worker` is++reserved as a plugin name.### Fault codes@@@ -815,7 +815,10 @@ func (n *Node) sendStatus() {}}if c := n.jobs.runtimeClasses(); len(c) > 0 {-- s.Capabilities[protocol.CapWorker] = protocol.WorkerCapabilities{RuntimeClasses: c}++ caps := n.opt.Config.Worker.Caps.WithDefaults()++ s.Capabilities[protocol.CapWorker] = protocol.WorkerCapabilities{RuntimeClasses: c, MaxJobs: caps.MaxJobs,++ MaxTTLMS: caps.MaxTTLMS, MaxWallMS: caps.MaxWallMS, MaxCPUMS: caps.MaxCPUMS,++ MaxMemBytes: caps.MaxMemBytes, MaxOutputBytes: caps.MaxOutputBytes}}if n.pub != nil {s.Video = n.pub.State()@@@ -10,6 +10,7 @@ import ("log/slog""os""path/filepath"++ "reflect""slices""strings""testing"@@@ -97,13 +98,22 @@ func startWorkerNode(t *testing.T) *env {e := &env{t: t, srv: srv, node: n, paths: paths, cancel: cancel, done: done}t.Cleanup(e.stop)e.conn = e.nextConn()-- _, err = e.conn.Expect(protocol.TypeStatus, 10*time.Second, func(m protocol.Message) bool {++ f, err := e.conn.Expect(protocol.TypeStatus, 10*time.Second, func(m protocol.Message) bool {_, ok := m.(protocol.Status).Capabilities[protocol.CapWorker]return ok})if err != nil {t.Fatal(err)}++ caps := cfg.Worker.Caps.WithDefaults()++ want := map[string]any{"runtime_classes": []any{protocol.ClassFunction}, "max_jobs": float64(caps.MaxJobs),++ "max_ttl_ms": float64(caps.MaxTTLMS), "max_wall_ms": float64(caps.MaxWallMS),++ "max_cpu_ms": float64(caps.MaxCPUMS), "max_mem_bytes": float64(caps.MaxMemBytes),++ "max_output_bytes": float64(caps.MaxOutputBytes)}++ got := f.Msg.(protocol.Status).Capabilities[protocol.CapWorker]++ if !reflect.DeepEqual(got, want) {++ t.Fatalf("worker capability = %#v, want %#v", got, want)++ }return e}@@@ -393,9 +393,15 @@ type JobExit struct {Usage JobExitUsage `json:"usage"`}--// WorkerCapabilities is status.capabilities[CapWorker]: the runtime classes this Node runs.++// WorkerCapabilities is status.capabilities[CapWorker]: the runtime classes this Node runs and its local job caps.type WorkerCapabilities struct {RuntimeClasses []string `json:"runtime_classes"`++ MaxJobs int `json:"max_jobs"`++ MaxTTLMS int64 `json:"max_ttl_ms"`++ MaxWallMS int64 `json:"max_wall_ms"`++ MaxCPUMS int64 `json:"max_cpu_ms"`++ MaxMemBytes int64 `json:"max_mem_bytes"`++ MaxOutputBytes int64 `json:"max_output_bytes"`}type DriverStatus struct {
@@ -237,8 +237,10 @@ A Node shutting down sends `job_exit` (`stopped`) also for a job acked but not s
@@ -237,8 +237,10 @@ A Node shutting down sends `job_exit` (`stopped`) also for a job acked but not s
`job_cancel` for an id that is not running (unknown, refused or ended) and `job_exit_ack` for an unknown id are
`job_cancel` for an id that is not running (unknown, refused or ended) and `job_exit_ack` for an unknown id are
ignored, with no answer: an `ack` keyed by a job id means the job was accepted.
ignored, with no answer: an `ack` keyed by a job id means the job was accepted.
-A Node that runs jobs lists its classes in `status.capabilities.worker` as `{"runtime_classes": ["function"]}`; the
-key is absent while it runs none, and `worker` is reserved as a plugin name.
+A Node that runs jobs lists its classes and effective local worker caps in `status.capabilities.worker`:
+`{"runtime_classes":["function"],"max_jobs":1,"max_ttl_ms":600000,"max_wall_ms":300000,"max_cpu_ms":300000,"max_mem_bytes":536870912,"max_output_bytes":1048576}`.
+The limits are `worker.caps` with defaults applied; the key is absent while the worker is off, and `worker` is
+reserved as a plugin name.
### Fault codes
### Fault codes
@@ -815,7 +815,10 @@ func (n *Node) sendStatus() {
@@ -815,7 +815,10 @@ func (n *Node) sendStatus() {
}
}
}
}
if c := n.jobs.runtimeClasses(); len(c) > 0 {
if c := n.jobs.runtimeClasses(); len(c) > 0 {
- s.Capabilities[protocol.CapWorker] = protocol.WorkerCapabilities{RuntimeClasses: c}
+ caps := n.opt.Config.Worker.Caps.WithDefaults()
+ s.Capabilities[protocol.CapWorker] = protocol.WorkerCapabilities{RuntimeClasses: c, MaxJobs: caps.MaxJobs,
+ MaxTTLMS: caps.MaxTTLMS, MaxWallMS: caps.MaxWallMS, MaxCPUMS: caps.MaxCPUMS,
+ MaxMemBytes: caps.MaxMemBytes, MaxOutputBytes: caps.MaxOutputBytes}
}
}
if n.pub != nil {
if n.pub != nil {
s.Video = n.pub.State()
s.Video = n.pub.State()
@@ -10,6 +10,7 @@ import (
@@ -10,6 +10,7 @@ import (
"log/slog"
"log/slog"
"os"
"os"
"path/filepath"
"path/filepath"
+ "reflect"
"slices"
"slices"
"strings"
"strings"
"testing"
"testing"
@@ -97,13 +98,22 @@ func startWorkerNode(t *testing.T) *env {
@@ -97,13 +98,22 @@ func startWorkerNode(t *testing.T) *env {
e := &env{t: t, srv: srv, node: n, paths: paths, cancel: cancel, done: done}
e := &env{t: t, srv: srv, node: n, paths: paths, cancel: cancel, done: done}
t.Cleanup(e.stop)
t.Cleanup(e.stop)
e.conn = e.nextConn()
e.conn = e.nextConn()
- _, err = e.conn.Expect(protocol.TypeStatus, 10*time.Second, func(m protocol.Message) bool {
+ f, err := e.conn.Expect(protocol.TypeStatus, 10*time.Second, func(m protocol.Message) bool {
_, ok := m.(protocol.Status).Capabilities[protocol.CapWorker]
_, ok := m.(protocol.Status).Capabilities[protocol.CapWorker]
return ok
return ok
})
})
if err != nil {
if err != nil {
t.Fatal(err)
t.Fatal(err)
}
}
+ caps := cfg.Worker.Caps.WithDefaults()
+ want := map[string]any{"runtime_classes": []any{protocol.ClassFunction}, "max_jobs": float64(caps.MaxJobs),
+ "max_ttl_ms": float64(caps.MaxTTLMS), "max_wall_ms": float64(caps.MaxWallMS),
+ "max_cpu_ms": float64(caps.MaxCPUMS), "max_mem_bytes": float64(caps.MaxMemBytes),
+ "max_output_bytes": float64(caps.MaxOutputBytes)}
+ got := f.Msg.(protocol.Status).Capabilities[protocol.CapWorker]
+ if !reflect.DeepEqual(got, want) {
+ t.Fatalf("worker capability = %#v, want %#v", got, want)
+ }
return e
return e
}
}
@@ -393,9 +393,15 @@ type JobExit struct {
@@ -393,9 +393,15 @@ type JobExit struct {
Usage JobExitUsage `json:"usage"`
Usage JobExitUsage `json:"usage"`
}
}
-// WorkerCapabilities is status.capabilities[CapWorker]: the runtime classes this Node runs.
+// WorkerCapabilities is status.capabilities[CapWorker]: the runtime classes this Node runs and its local job caps.
type WorkerCapabilities struct {
type WorkerCapabilities struct {
RuntimeClasses []string `json:"runtime_classes"`
RuntimeClasses []string `json:"runtime_classes"`
+ MaxJobs int `json:"max_jobs"`
+ MaxTTLMS int64 `json:"max_ttl_ms"`
+ MaxWallMS int64 `json:"max_wall_ms"`
+ MaxCPUMS int64 `json:"max_cpu_ms"`
+ MaxMemBytes int64 `json:"max_mem_bytes"`
+ MaxOutputBytes int64 `json:"max_output_bytes"`
}
}
type DriverStatus struct {
type DriverStatus struct {