openvibers/openvibe.link Commits
Advertise worker capability names in the Node status descriptor (#19)
Changes
@@@ -245,10 +245,14 @@ 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 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.++A Node that runs jobs lists its classes, the reserved Fabric capability names they advertise, and its effective local++worker caps in `status.capabilities.worker`:++`{"capabilities":["worker:function"],"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}`.++`capabilities` holds one `worker:<class>` name per entry in `runtime_classes` (the `platform.resource-offer@1`++capabilities namespace OpenVibe.Run routes on; the names are reserved in OpenVibe.Contracts and no schema change++was needed), so only classes this Node actually runs are advertised: `worker:function` today, `worker:code` once the++`code` class ships. 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@@@ -39,11 +39,12 @@ the server. The worker is **off by default**; off, the Node refuses every job `ceach clamped to them (the stricter value wins, as with `max_command_ms`). `max_output_bytes` caps the stdout bytesone job sends (counted on the wire, after invalid UTF-8 became U+FFFD); `max_jobs` how many run at once.--The class `function` is advertised in `status.capabilities.worker` only when `enabled` is true, the OS is Linux and a--**boot-time probe** passes: it starts `/bin/sh` the way a job is started and checks that the process got user,--network and PID namespaces of its own, that its CPU limit can be set and that its CPU and memory can be read. If the--probe fails the Node logs why (`worker off: …`) and refuses every job, exactly as with the worker off. On Ubuntu 24.04--and later, unprivileged user namespaces are restricted by AppArmor; a Node running as root with `run_as` is not++The class `function` — advertised as the reserved Fabric capability `worker:function` in `status.capabilities.worker`++(the `platform.resource-offer@1` namespace OpenVibe.Run routes on) — is offered only when `enabled` is true, the OS is++Linux and a **boot-time probe** passes: it starts `/bin/sh` the way a job is started and checks that the process got++user, network and PID namespaces of its own, that its CPU limit can be set and that its CPU and memory can be read. If++the probe fails the Node logs why (`worker off: …`) and refuses every job, exactly as with the worker off. On Ubuntu++24.04 and later, unprivileged user namespaces are restricted by AppArmor; a Node running as root with `run_as` is notaffected.## How a job runs@@@ -816,7 +816,7 @@ func (n *Node) sendStatus() {}if c := n.jobs.runtimeClasses(); len(c) > 0 {caps := n.opt.Config.Worker.Caps.WithDefaults()-- s.Capabilities[protocol.CapWorker] = protocol.WorkerCapabilities{RuntimeClasses: c, MaxJobs: caps.MaxJobs,++ s.Capabilities[protocol.CapWorker] = protocol.WorkerCapabilities{Capabilities: protocol.WorkerCapabilityNames(c), RuntimeClasses: c, MaxJobs: caps.MaxJobs,MaxTTLMS: caps.MaxTTLMS, MaxWallMS: caps.MaxWallMS, MaxCPUMS: caps.MaxCPUMS,MaxMemBytes: caps.MaxMemBytes, MaxOutputBytes: caps.MaxOutputBytes}}@@@ -106,7 +106,7 @@ func startWorkerNode(t *testing.T) *env {t.Fatal(err)}caps := cfg.Worker.Caps.WithDefaults()-- want := map[string]any{"runtime_classes": []any{protocol.ClassFunction}, "max_jobs": float64(caps.MaxJobs),++ want := map[string]any{"capabilities": []any{"worker:function"}, "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)}@@@ -88,6 +88,20 @@ const NetDeny = "deny"// present only when the Node runs at least one class.const CapWorker = "worker"++// WorkerCapPrefix is the namespace of the reserved Fabric capability names ([email protected])++// a Node's runtime classes map to, one name per class: "worker:function", "worker:code", …. A name matches the++// offer's capabilities pattern ^[a-z][a-z0-9-]*:[a-z0-9.-]+$ with no schema change needed.++const WorkerCapPrefix = "worker:"++++// WorkerCapabilityNames returns the reserved Fabric capability names for the given runtime classes, in order.++func WorkerCapabilityNames(classes []string) []string {++ names := make([]string, 0, len(classes))++ for _, c := range classes {++ names = append(names, WorkerCapPrefix+c)++ }++ return names++}++// Job refusal reasons, carried in nack.message (nack.fault_code is bad_value, bad_frame or unsupported).const (JobBadID = "missing or malformed job id"@@@ -395,6 +409,7 @@ type JobExit struct {// WorkerCapabilities is status.capabilities[CapWorker]: the runtime classes this Node runs and its local job caps.type WorkerCapabilities struct {++ Capabilities []string `json:"capabilities"` // the reserved Fabric names these classes advertise: worker:function, …RuntimeClasses []string `json:"runtime_classes"`MaxJobs int `json:"max_jobs"`MaxTTLMS int64 `json:"max_ttl_ms"`@@@ -248,6 +248,21 @@ func TestJobFrameFixtures(t *testing.T) {}}++func TestWorkerCapabilityNames(t *testing.T) {++ for _, tc := range []struct {++ classes []string++ want []string++ }{++ {nil, []string{}},++ {[]string{ClassFunction}, []string{"worker:function"}},++ {[]string{ClassFunction, ClassCode}, []string{"worker:function", "worker:code"}},++ } {++ if got := WorkerCapabilityNames(tc.classes); !reflect.DeepEqual(got, tc.want) {++ t.Errorf("WorkerCapabilityNames(%v) = %v, want %v", tc.classes, got, tc.want)++ }++ }++}++func TestJobCheck(t *testing.T) {const id = "job_01JAB2C3D4E5F6G7H8J9K0MNPQ"valid := func() Job {
@@ -245,10 +245,14 @@ A Node shutting down sends `job_exit` (`stopped`) also for a job acked but not s
@@ -245,10 +245,14 @@ 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 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.
+A Node that runs jobs lists its classes, the reserved Fabric capability names they advertise, and its effective local
+worker caps in `status.capabilities.worker`:
+`{"capabilities":["worker:function"],"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}`.
+`capabilities` holds one `worker:<class>` name per entry in `runtime_classes` (the `platform.resource-offer@1`
+capabilities namespace OpenVibe.Run routes on; the names are reserved in OpenVibe.Contracts and no schema change
+was needed), so only classes this Node actually runs are advertised: `worker:function` today, `worker:code` once the
+`code` class ships. 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
@@ -39,11 +39,12 @@ the server. The worker is **off by default**; off, the Node refuses every job `c
@@ -39,11 +39,12 @@ the server. The worker is **off by default**; off, the Node refuses every job `c
each clamped to them (the stricter value wins, as with `max_command_ms`). `max_output_bytes` caps the stdout bytes
each clamped to them (the stricter value wins, as with `max_command_ms`). `max_output_bytes` caps the stdout bytes
one job sends (counted on the wire, after invalid UTF-8 became U+FFFD); `max_jobs` how many run at once.
one job sends (counted on the wire, after invalid UTF-8 became U+FFFD); `max_jobs` how many run at once.
-The class `function` is advertised in `status.capabilities.worker` only when `enabled` is true, the OS is Linux and a
-**boot-time probe** passes: it starts `/bin/sh` the way a job is started and checks that the process got user,
-network and PID namespaces of its own, that its CPU limit can be set and that its CPU and memory can be read. If the
-probe fails the Node logs why (`worker off: …`) and refuses every job, exactly as with the worker off. On Ubuntu 24.04
-and later, unprivileged user namespaces are restricted by AppArmor; a Node running as root with `run_as` is not
+The class `function` — advertised as the reserved Fabric capability `worker:function` in `status.capabilities.worker`
+(the `platform.resource-offer@1` namespace OpenVibe.Run routes on) — is offered only when `enabled` is true, the OS is
+Linux and a **boot-time probe** passes: it starts `/bin/sh` the way a job is started and checks that the process got
+user, network and PID namespaces of its own, that its CPU limit can be set and that its CPU and memory can be read. If
+the probe fails the Node logs why (`worker off: …`) and refuses every job, exactly as with the worker off. On Ubuntu
+24.04 and later, unprivileged user namespaces are restricted by AppArmor; a Node running as root with `run_as` is not
affected.
affected.
## How a job runs
## How a job runs
@@ -816,7 +816,7 @@ func (n *Node) sendStatus() {
@@ -816,7 +816,7 @@ func (n *Node) sendStatus() {
}
}
if c := n.jobs.runtimeClasses(); len(c) > 0 {
if c := n.jobs.runtimeClasses(); len(c) > 0 {
caps := n.opt.Config.Worker.Caps.WithDefaults()
caps := n.opt.Config.Worker.Caps.WithDefaults()
- s.Capabilities[protocol.CapWorker] = protocol.WorkerCapabilities{RuntimeClasses: c, MaxJobs: caps.MaxJobs,
+ s.Capabilities[protocol.CapWorker] = protocol.WorkerCapabilities{Capabilities: protocol.WorkerCapabilityNames(c), RuntimeClasses: c, MaxJobs: caps.MaxJobs,
MaxTTLMS: caps.MaxTTLMS, MaxWallMS: caps.MaxWallMS, MaxCPUMS: caps.MaxCPUMS,
MaxTTLMS: caps.MaxTTLMS, MaxWallMS: caps.MaxWallMS, MaxCPUMS: caps.MaxCPUMS,
MaxMemBytes: caps.MaxMemBytes, MaxOutputBytes: caps.MaxOutputBytes}
MaxMemBytes: caps.MaxMemBytes, MaxOutputBytes: caps.MaxOutputBytes}
}
}
@@ -106,7 +106,7 @@ func startWorkerNode(t *testing.T) *env {
@@ -106,7 +106,7 @@ func startWorkerNode(t *testing.T) *env {
t.Fatal(err)
t.Fatal(err)
}
}
caps := cfg.Worker.Caps.WithDefaults()
caps := cfg.Worker.Caps.WithDefaults()
- want := map[string]any{"runtime_classes": []any{protocol.ClassFunction}, "max_jobs": float64(caps.MaxJobs),
+ want := map[string]any{"capabilities": []any{"worker:function"}, "runtime_classes": []any{protocol.ClassFunction}, "max_jobs": float64(caps.MaxJobs),
"max_ttl_ms": float64(caps.MaxTTLMS), "max_wall_ms": float64(caps.MaxWallMS),
"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_cpu_ms": float64(caps.MaxCPUMS), "max_mem_bytes": float64(caps.MaxMemBytes),
"max_output_bytes": float64(caps.MaxOutputBytes)}
"max_output_bytes": float64(caps.MaxOutputBytes)}
@@ -88,6 +88,20 @@ const NetDeny = "deny"
@@ -88,6 +88,20 @@ const NetDeny = "deny"
// present only when the Node runs at least one class.
// present only when the Node runs at least one class.
const CapWorker = "worker"
const CapWorker = "worker"
+// WorkerCapPrefix is the namespace of the reserved Fabric capability names ([email protected])
+// a Node's runtime classes map to, one name per class: "worker:function", "worker:code", …. A name matches the
+// offer's capabilities pattern ^[a-z][a-z0-9-]*:[a-z0-9.-]+$ with no schema change needed.
+const WorkerCapPrefix = "worker:"
+
+// WorkerCapabilityNames returns the reserved Fabric capability names for the given runtime classes, in order.
+func WorkerCapabilityNames(classes []string) []string {
+ names := make([]string, 0, len(classes))
+ for _, c := range classes {
+ names = append(names, WorkerCapPrefix+c)
+ }
+ return names
+}
+
// Job refusal reasons, carried in nack.message (nack.fault_code is bad_value, bad_frame or unsupported).
// Job refusal reasons, carried in nack.message (nack.fault_code is bad_value, bad_frame or unsupported).
const (
const (
JobBadID = "missing or malformed job id"
JobBadID = "missing or malformed job id"
@@ -395,6 +409,7 @@ type JobExit struct {
@@ -395,6 +409,7 @@ type JobExit struct {
// WorkerCapabilities is status.capabilities[CapWorker]: the runtime classes this Node runs and its local job caps.
// WorkerCapabilities is status.capabilities[CapWorker]: the runtime classes this Node runs and its local job caps.
type WorkerCapabilities struct {
type WorkerCapabilities struct {
+ Capabilities []string `json:"capabilities"` // the reserved Fabric names these classes advertise: worker:function, …
RuntimeClasses []string `json:"runtime_classes"`
RuntimeClasses []string `json:"runtime_classes"`
MaxJobs int `json:"max_jobs"`
MaxJobs int `json:"max_jobs"`
MaxTTLMS int64 `json:"max_ttl_ms"`
MaxTTLMS int64 `json:"max_ttl_ms"`
@@ -248,6 +248,21 @@ func TestJobFrameFixtures(t *testing.T) {
@@ -248,6 +248,21 @@ func TestJobFrameFixtures(t *testing.T) {
}
}
}
}
+func TestWorkerCapabilityNames(t *testing.T) {
+ for _, tc := range []struct {
+ classes []string
+ want []string
+ }{
+ {nil, []string{}},
+ {[]string{ClassFunction}, []string{"worker:function"}},
+ {[]string{ClassFunction, ClassCode}, []string{"worker:function", "worker:code"}},
+ } {
+ if got := WorkerCapabilityNames(tc.classes); !reflect.DeepEqual(got, tc.want) {
+ t.Errorf("WorkerCapabilityNames(%v) = %v, want %v", tc.classes, got, tc.want)
+ }
+ }
+}
+
func TestJobCheck(t *testing.T) {
func TestJobCheck(t *testing.T) {
const id = "job_01JAB2C3D4E5F6G7H8J9K0MNPQ"
const id = "job_01JAB2C3D4E5F6G7H8J9K0MNPQ"
valid := func() Job {
valid := func() Job {