openvibers/openvibe.link Commits
Advertise worker:function only when a function is declared (#21)
Changes
@@@ -251,8 +251,9 @@ worker caps in `status.capabilities.worker`:`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 changewas 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.++`code` class ships. The limits are `worker.caps` with defaults applied; the key is absent while the worker is off or++declares no `functions` (a probe that passes with none advertises nothing, so Fabric places no job this Node would++refuse), and `worker` is reserved as a plugin name.### Fault codes@@@ -41,8 +41,8 @@ the server. The worker is **off by default**; off, the Node refuses every job `cThe 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++Linux, at least one `functions` entry is declared 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. Ifthe probe fails the Node logs why (`worker off: …`) and refuses every job, exactly as with the worker off. On Ubuntu24.04 and later, unprivileged user namespaces are restricted by AppArmor; a Node running as root with `run_as` is notaffected.@@@ -155,9 +155,11 @@ func New(opt Options) (*Node, error) {} else {n.worker = ww.SetStopped(stopFault(n.latch.State()))-- // The only class advertised (status.capabilities.worker), and only with a worker to run it; with none,-- // every job is refused "class not available".-- n.jobs.admit, n.jobs.classes = w.Admit, []string{protocol.ClassFunction}++ n.jobs.admit = w.Admit++ // Advertise function jobs only when the worker has a function to run.++ if len(opt.Config.Worker.Functions) > 0 {++ n.jobs.classes = []string{protocol.ClassFunction}++ }opt.Log.Info("worker on", "functions", len(opt.Config.Worker.Functions))}}@@@ -58,6 +58,23 @@ func startWorkerNode(t *testing.T) *env {if err != nil {t.Fatal(err)}++ cfg := config.WorkerConfig{Enabled: true, AllowSameUser: true, Caps: config.WorkerCaps{MaxMemBytes: 8 << 30},++ Functions: []config.FunctionConfig{{Name: "hello", Version: "1.0.0", Command: []string{exe, "-test.run=^TestWorkerHelperProcess$", "--", "job"}}}}++ e, status := startWorkerNodeWith(t, cfg, true)++ caps := cfg.Caps.WithDefaults()++ 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)}++ got := status.Capabilities[protocol.CapWorker]++ if !reflect.DeepEqual(got, want) {++ t.Fatalf("worker capability = %#v, want %#v", got, want)++ }++ return e++}++++func startWorkerNodeWith(t *testing.T, workerCfg config.WorkerConfig, probeShouldPass bool) (*env, protocol.Status) {++ t.Helper()home, err := os.MkdirTemp("", "ovn")if err != nil {t.Fatal(err)@@@ -76,8 +93,7 @@ func startWorkerNode(t *testing.T) *env {cfg := config.Default()cfg.Plugins = nilcfg.Video = config.VideoConfig{Source: "off"}-- cfg.Worker = config.WorkerConfig{Enabled: true, AllowSameUser: true, Caps: config.WorkerCaps{MaxMemBytes: 8 << 30},-- Functions: []config.FunctionConfig{{Name: "hello", Version: "1.0.0", Command: []string{exe, "-test.run=^TestWorkerHelperProcess$", "--", "job"}}}}++ cfg.Worker = workerCfgif err := cfg.Validate(); err != nil {t.Fatal(err)}@@@ -86,11 +102,15 @@ func startWorkerNode(t *testing.T) *env {if err != nil {t.Fatal(err)}-- if !slices.Contains(n.jobs.runtimeClasses(), protocol.ClassFunction) {-- if os.Getenv("OPENVIBE_WORKER_TESTS") == "require" {-- t.Fatal("job isolation unavailable: the worker probe failed")++ if probeShouldPass {++ if n.worker == nil {++ if os.Getenv("OPENVIBE_WORKER_TESTS") == "require" {++ t.Fatal("job isolation unavailable: the worker probe failed")++ }++ t.Skip("job isolation unavailable here (OPENVIBE_WORKER_TESTS=require fails instead)")}-- t.Skip("job isolation unavailable here (OPENVIBE_WORKER_TESTS=require fails instead)")++ } else if n.worker != nil {++ t.Fatal("worker probe passed without an identity policy")}ctx, cancel := context.WithCancel(context.Background())done := make(chan struct{})@@@ -98,23 +118,38 @@ 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()-- f, err := e.conn.Expect(protocol.TypeStatus, 10*time.Second, func(m protocol.Message) bool {-- _, ok := m.(protocol.Status).Capabilities[protocol.CapWorker]-- return ok-- })++ f, err := e.conn.Expect(protocol.TypeStatus, 10*time.Second, nil)if err != nil {t.Fatal(err)}-- caps := cfg.Worker.Caps.WithDefaults()-- 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)}-- got := f.Msg.(protocol.Status).Capabilities[protocol.CapWorker]-- if !reflect.DeepEqual(got, want) {-- t.Fatalf("worker capability = %#v, want %#v", got, want)++ return e, f.Msg.(protocol.Status)++}++++func TestWorkerWithNoFunctions(t *testing.T) {++ e, status := startWorkerNodeWith(t, config.WorkerConfig{Enabled: true, AllowSameUser: true}, true)++ if classes := e.node.jobs.runtimeClasses(); len(classes) != 0 {++ t.Fatalf("runtime classes with no functions = %v, want none", classes)++ }++ if _, ok := status.Capabilities[protocol.CapWorker]; ok {++ t.Fatalf("worker capability advertised with no functions: %+v", status.Capabilities)++ }++ if nk := e.jobNack(testJob(jobID(1)), protocol.JobNotAvailable); nk.FaultCode != protocol.FaultUnsupported {++ t.Fatalf("job with no functions: %+v", nk)++ }++}++++func TestWorkerFailedProbeDoesNotAdvertiseFunctions(t *testing.T) {++ e, status := startWorkerNodeWith(t, config.WorkerConfig{Enabled: true,++ Functions: []config.FunctionConfig{{Name: "hello", Version: "1.0.0", Command: []string{"/bin/true"}}}}, false)++ if classes := e.node.jobs.runtimeClasses(); len(classes) != 0 {++ t.Fatalf("runtime classes after failed probe = %v, want none", classes)++ }++ if _, ok := status.Capabilities[protocol.CapWorker]; ok {++ t.Fatalf("worker capability advertised after failed probe: %+v", status.Capabilities)++ }++ if nk := e.jobNack(testJob(jobID(1)), protocol.JobNotAvailable); nk.FaultCode != protocol.FaultUnsupported {++ t.Fatalf("job after failed probe: %+v", nk)}-- return e}// TestJobRunsOverTheLink: a job sent by the fake Bot is acked, then job_started, job_stdout, job_usage (second 0)
@@ -251,8 +251,9 @@ worker caps in `status.capabilities.worker`:
@@ -251,8 +251,9 @@ worker caps in `status.capabilities.worker`:
`capabilities` holds one `worker:<class>` name per entry in `runtime_classes` (the `platform.resource-offer@1`
`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
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
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.
+`code` class ships. The limits are `worker.caps` with defaults applied; the key is absent while the worker is off or
+declares no `functions` (a probe that passes with none advertises nothing, so Fabric places no job this Node would
+refuse), and `worker` is reserved as a plugin name.
### Fault codes
### Fault codes
@@ -41,8 +41,8 @@ the server. The worker is **off by default**; off, the Node refuses every job `c
@@ -41,8 +41,8 @@ the server. The worker is **off by default**; off, the Node refuses every job `c
The class `function` — advertised as the reserved Fabric capability `worker:function` in `status.capabilities.worker`
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
(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
+Linux, at least one `functions` entry is declared 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
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
24.04 and later, unprivileged user namespaces are restricted by AppArmor; a Node running as root with `run_as` is not
affected.
affected.
@@ -155,9 +155,11 @@ func New(opt Options) (*Node, error) {
@@ -155,9 +155,11 @@ func New(opt Options) (*Node, error) {
} else {
} else {
n.worker = w
n.worker = w
w.SetStopped(stopFault(n.latch.State()))
w.SetStopped(stopFault(n.latch.State()))
- // The only class advertised (status.capabilities.worker), and only with a worker to run it; with none,
- // every job is refused "class not available".
- n.jobs.admit, n.jobs.classes = w.Admit, []string{protocol.ClassFunction}
+ n.jobs.admit = w.Admit
+ // Advertise function jobs only when the worker has a function to run.
+ if len(opt.Config.Worker.Functions) > 0 {
+ n.jobs.classes = []string{protocol.ClassFunction}
+ }
opt.Log.Info("worker on", "functions", len(opt.Config.Worker.Functions))
opt.Log.Info("worker on", "functions", len(opt.Config.Worker.Functions))
}
}
}
}
@@ -58,6 +58,23 @@ func startWorkerNode(t *testing.T) *env {
@@ -58,6 +58,23 @@ func startWorkerNode(t *testing.T) *env {
if err != nil {
if err != nil {
t.Fatal(err)
t.Fatal(err)
}
}
+ cfg := config.WorkerConfig{Enabled: true, AllowSameUser: true, Caps: config.WorkerCaps{MaxMemBytes: 8 << 30},
+ Functions: []config.FunctionConfig{{Name: "hello", Version: "1.0.0", Command: []string{exe, "-test.run=^TestWorkerHelperProcess$", "--", "job"}}}}
+ e, status := startWorkerNodeWith(t, cfg, true)
+ caps := cfg.Caps.WithDefaults()
+ 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)}
+ got := status.Capabilities[protocol.CapWorker]
+ if !reflect.DeepEqual(got, want) {
+ t.Fatalf("worker capability = %#v, want %#v", got, want)
+ }
+ return e
+}
+
+func startWorkerNodeWith(t *testing.T, workerCfg config.WorkerConfig, probeShouldPass bool) (*env, protocol.Status) {
+ t.Helper()
home, err := os.MkdirTemp("", "ovn")
home, err := os.MkdirTemp("", "ovn")
if err != nil {
if err != nil {
t.Fatal(err)
t.Fatal(err)
@@ -76,8 +93,7 @@ func startWorkerNode(t *testing.T) *env {
@@ -76,8 +93,7 @@ func startWorkerNode(t *testing.T) *env {
cfg := config.Default()
cfg := config.Default()
cfg.Plugins = nil
cfg.Plugins = nil
cfg.Video = config.VideoConfig{Source: "off"}
cfg.Video = config.VideoConfig{Source: "off"}
- cfg.Worker = config.WorkerConfig{Enabled: true, AllowSameUser: true, Caps: config.WorkerCaps{MaxMemBytes: 8 << 30},
- Functions: []config.FunctionConfig{{Name: "hello", Version: "1.0.0", Command: []string{exe, "-test.run=^TestWorkerHelperProcess$", "--", "job"}}}}
+ cfg.Worker = workerCfg
if err := cfg.Validate(); err != nil {
if err := cfg.Validate(); err != nil {
t.Fatal(err)
t.Fatal(err)
}
}
@@ -86,11 +102,15 @@ func startWorkerNode(t *testing.T) *env {
@@ -86,11 +102,15 @@ func startWorkerNode(t *testing.T) *env {
if err != nil {
if err != nil {
t.Fatal(err)
t.Fatal(err)
}
}
- if !slices.Contains(n.jobs.runtimeClasses(), protocol.ClassFunction) {
- if os.Getenv("OPENVIBE_WORKER_TESTS") == "require" {
- t.Fatal("job isolation unavailable: the worker probe failed")
+ if probeShouldPass {
+ if n.worker == nil {
+ if os.Getenv("OPENVIBE_WORKER_TESTS") == "require" {
+ t.Fatal("job isolation unavailable: the worker probe failed")
+ }
+ t.Skip("job isolation unavailable here (OPENVIBE_WORKER_TESTS=require fails instead)")
}
}
- t.Skip("job isolation unavailable here (OPENVIBE_WORKER_TESTS=require fails instead)")
+ } else if n.worker != nil {
+ t.Fatal("worker probe passed without an identity policy")
}
}
ctx, cancel := context.WithCancel(context.Background())
ctx, cancel := context.WithCancel(context.Background())
done := make(chan struct{})
done := make(chan struct{})
@@ -98,23 +118,38 @@ func startWorkerNode(t *testing.T) *env {
@@ -98,23 +118,38 @@ 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()
- f, err := e.conn.Expect(protocol.TypeStatus, 10*time.Second, func(m protocol.Message) bool {
- _, ok := m.(protocol.Status).Capabilities[protocol.CapWorker]
- return ok
- })
+ f, err := e.conn.Expect(protocol.TypeStatus, 10*time.Second, nil)
if err != nil {
if err != nil {
t.Fatal(err)
t.Fatal(err)
}
}
- caps := cfg.Worker.Caps.WithDefaults()
- 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)}
- got := f.Msg.(protocol.Status).Capabilities[protocol.CapWorker]
- if !reflect.DeepEqual(got, want) {
- t.Fatalf("worker capability = %#v, want %#v", got, want)
+ return e, f.Msg.(protocol.Status)
+}
+
+func TestWorkerWithNoFunctions(t *testing.T) {
+ e, status := startWorkerNodeWith(t, config.WorkerConfig{Enabled: true, AllowSameUser: true}, true)
+ if classes := e.node.jobs.runtimeClasses(); len(classes) != 0 {
+ t.Fatalf("runtime classes with no functions = %v, want none", classes)
+ }
+ if _, ok := status.Capabilities[protocol.CapWorker]; ok {
+ t.Fatalf("worker capability advertised with no functions: %+v", status.Capabilities)
+ }
+ if nk := e.jobNack(testJob(jobID(1)), protocol.JobNotAvailable); nk.FaultCode != protocol.FaultUnsupported {
+ t.Fatalf("job with no functions: %+v", nk)
+ }
+}
+
+func TestWorkerFailedProbeDoesNotAdvertiseFunctions(t *testing.T) {
+ e, status := startWorkerNodeWith(t, config.WorkerConfig{Enabled: true,
+ Functions: []config.FunctionConfig{{Name: "hello", Version: "1.0.0", Command: []string{"/bin/true"}}}}, false)
+ if classes := e.node.jobs.runtimeClasses(); len(classes) != 0 {
+ t.Fatalf("runtime classes after failed probe = %v, want none", classes)
+ }
+ if _, ok := status.Capabilities[protocol.CapWorker]; ok {
+ t.Fatalf("worker capability advertised after failed probe: %+v", status.Capabilities)
+ }
+ if nk := e.jobNack(testJob(jobID(1)), protocol.JobNotAvailable); nk.FaultCode != protocol.FaultUnsupported {
+ t.Fatalf("job after failed probe: %+v", nk)
}
}
- return e
}
}
// TestJobRunsOverTheLink: a job sent by the fake Bot is acked, then job_started, job_stdout, job_usage (second 0)
// TestJobRunsOverTheLink: a job sent by the fake Bot is acked, then job_started, job_stdout, job_usage (second 0)