@cryptotaxi247 / netdata-1 / commits / 55f4e5a56

chore(go/agent/jobmgr): harden dyncfg lifecycle, locking, and wait-decision flow (#21840)

Ilya Mashchenko committed Feb 28, 2026 at 06:30 UTC 55f4e5a568b22b0d0143987602669bcd8978412e
19 files changed +1657 -421
src/go/plugin/agent/discovery/sd/sd.go
+18 -5
@@ -8,6 +8,7 @@ import (
8 "io"
9 "log/slog"
10 "sync"
11 + "time"
12
13 "github.com/netdata/netdata/go/plugins/plugin/agent/discovery/sd/pipeline"
14 "github.com/netdata/netdata/go/plugins/plugin/agent/policy"
@@ -20,6 +21,8 @@ import (
21 "github.com/netdata/netdata/go/plugins/pkg/netdataapi"
22 )
23
24 +const waitDecisionTimeout = 5 * time.Second
25 +
26 type Config struct {
27 ConfigDefaults confgroup.Registry
28 PluginName string
@@ -68,6 +71,7 @@ func NewServiceDiscovery(cfg Config) (*ServiceDiscovery, error) {
71 WaitKey: func(cfg sdConfig) string {
72 return cfg.PipelineKey()
73 },
74 + WaitTimeout: waitDecisionTimeout,
75
76 Path: fmt.Sprintf(dyncfgSDPath, cfg.PluginName),
77 EnableFailCode: 422,
@@ -167,12 +171,21 @@ func (d *ServiceDiscovery) Run(ctx context.Context, in chan<- []*confgroup.Group
171 func (d *ServiceDiscovery) run(ctx context.Context) {
172 for {
173 if d.handler.WaitingForDecision() {
170 - // Waiting for enable/disable command - only process dyncfg commands
171 - select {
172 - case <-ctx.Done():
174 + step, ok := d.handler.NextWaitDecisionStep(ctx, d.dyncfgCh)
175 + if !ok {
176 return
174 - case fn := <-d.dyncfgCh:
175 - d.dyncfgSeqExec(fn)
177 + }
178 + if step.HasCommand {
179 + d.dyncfgSeqExec(step.Command)
180 + continue
181 + }
182 + if step.TimedOut {
183 + d.Errorf(
184 + "dyncfg: timed out waiting for enable/disable decision for '%s' (elapsed=%s threshold=%s); keeping status 'accepted' and continuing",
185 + step.Timeout.Key,
186 + step.Timeout.Elapsed,
187 + step.Timeout.Threshold,
188 + )
189 }
190 } else {
191 select {
src/go/plugin/agent/discovery/sd/wait_decision_test.go new
+220
@@ -0,0 +1,220 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package sd
4 +
5 +import (
6 + "bytes"
7 + "context"
8 + "fmt"
9 + "testing"
10 + "time"
11 +
12 + "github.com/netdata/netdata/go/plugins/logger"
13 + "github.com/netdata/netdata/go/plugins/pkg/netdataapi"
14 + "github.com/netdata/netdata/go/plugins/pkg/safewriter"
15 + "github.com/netdata/netdata/go/plugins/plugin/agent/discovery/sd/pipeline"
16 + "github.com/netdata/netdata/go/plugins/plugin/agent/policy"
17 + "github.com/netdata/netdata/go/plugins/plugin/framework/confgroup"
18 + "github.com/netdata/netdata/go/plugins/plugin/framework/dyncfg"
19 + "github.com/netdata/netdata/go/plugins/plugin/framework/functions"
20 +
21 + "github.com/stretchr/testify/assert"
22 + "github.com/stretchr/testify/require"
23 +)
24 +
25 +func TestServiceDiscovery_Run_WaitDecision(t *testing.T) {
26 + tests := map[string]struct {
27 + waitTimeout time.Duration
28 + run func(t *testing.T, sd *ServiceDiscovery, confCh chan confFile, stop func())
29 + }{
30 + "timeout clears wait gate and keeps accepted state": {
31 + waitTimeout: 40 * time.Millisecond,
32 + run: func(t *testing.T, sd *ServiceDiscovery, confCh chan confFile, stop func()) {
33 + cfg := prepareConfigFile("/etc/netdata/sd.d/job1.conf", "job1")
34 + confCh <- cfg
35 +
36 + require.Eventually(t, sd.handler.WaitingForDecision, time.Second, 10*time.Millisecond)
37 + require.Eventually(t, func() bool { return !sd.handler.WaitingForDecision() }, time.Second, 10*time.Millisecond)
38 +
39 + stop()
40 +
41 + entry, ok := sd.exposed.LookupByKey(testDiscovererTypeNetListeners + ":job1")
42 + require.True(t, ok, "expected discovered config to stay exposed after wait timeout")
43 + assert.Equal(t, dyncfg.StatusAccepted, entry.Status)
44 + assert.False(t, sd.mgr.IsRunning(pipelineKeyFromSource(cfg.source)))
45 + },
46 + },
47 + "enable command before timeout clears wait and starts pipeline": {
48 + waitTimeout: 750 * time.Millisecond,
49 + run: func(t *testing.T, sd *ServiceDiscovery, confCh chan confFile, stop func()) {
50 + cfg := prepareConfigFile("/etc/netdata/sd.d/job1.conf", "job1")
51 + confCh <- cfg
52 +
53 + require.Eventually(t, sd.handler.WaitingForDecision, time.Second, 10*time.Millisecond)
54 +
55 + sd.dyncfgCh <- dyncfg.NewFunction(functions.Function{
56 + UID: "enable-job1",
57 + Args: []string{sd.dyncfgJobID(testDiscovererTypeNetListeners, "job1"), "enable"},
58 + })
59 +
60 + require.Eventually(t, func() bool {
61 + return !sd.handler.WaitingForDecision() &&
62 + exposedExistsByKey(sd.exposed, testDiscovererTypeNetListeners+":job1") &&
63 + sd.mgr.IsRunning(pipelineKeyFromSource(cfg.source))
64 + }, time.Second, 10*time.Millisecond)
65 +
66 + stop()
67 +
68 + entry, ok := sd.exposed.LookupByKey(testDiscovererTypeNetListeners + ":job1")
69 + require.True(t, ok)
70 + assert.Equal(t, dyncfg.StatusRunning, entry.Status)
71 + },
72 + },
73 + "timeout unblocks and next config is processed": {
74 + waitTimeout: 40 * time.Millisecond,
75 + run: func(t *testing.T, sd *ServiceDiscovery, confCh chan confFile, stop func()) {
76 + cfg1 := prepareConfigFile("/etc/netdata/sd.d/job1.conf", "job1")
77 + cfg2 := prepareConfigFile("/etc/netdata/sd.d/job2.conf", "job2")
78 +
79 + confCh <- cfg1
80 + require.Eventually(t, sd.handler.WaitingForDecision, time.Second, 10*time.Millisecond)
81 +
82 + secondSent := make(chan struct{})
83 + go func() {
84 + confCh <- cfg2
85 + close(secondSent)
86 + }()
87 +
88 + select {
89 + case <-secondSent:
90 + t.Fatalf("second config should block while wait gate is active")
91 + case <-time.After(20 * time.Millisecond):
92 + }
93 +
94 + require.Eventually(t, func() bool { return !sd.handler.WaitingForDecision() }, time.Second, 10*time.Millisecond)
95 + require.Eventually(t, func() bool {
96 + select {
97 + case <-secondSent:
98 + return true
99 + default:
100 + return false
101 + }
102 + }, time.Second, 10*time.Millisecond)
103 +
104 + require.Eventually(t, func() bool {
105 + ok1 := exposedExistsByKey(sd.exposed, testDiscovererTypeNetListeners+":job1")
106 + ok2 := exposedExistsByKey(sd.exposed, testDiscovererTypeNetListeners+":job2")
107 + return ok1 && ok2
108 + }, time.Second, 10*time.Millisecond)
109 +
110 + stop()
111 +
112 + entry1, ok := sd.exposed.LookupByKey(testDiscovererTypeNetListeners + ":job1")
113 + require.True(t, ok)
114 + assert.Equal(t, dyncfg.StatusAccepted, entry1.Status)
115 + entry2, ok := sd.exposed.LookupByKey(testDiscovererTypeNetListeners + ":job2")
116 + require.True(t, ok)
117 + assert.Equal(t, dyncfg.StatusAccepted, entry2.Status)
118 + },
119 + },
120 + }
121 +
122 + for name, tc := range tests {
123 + t.Run(name, func(t *testing.T) {
124 + sd, confCh, cancel, done := newWaitTestServiceDiscovery(t, tc.waitTimeout)
125 + stopped := false
126 + stop := func() {
127 + if stopped {
128 + return
129 + }
130 + stopped = true
131 + stopWaitTestServiceDiscovery(t, sd, cancel, done)
132 + }
133 + defer stop()
134 + tc.run(t, sd, confCh, stop)
135 + })
136 + }
137 +}
138 +
139 +func newWaitTestServiceDiscovery(t *testing.T, waitTimeout time.Duration) (*ServiceDiscovery, chan confFile, context.CancelFunc, <-chan struct{}) {
140 + t.Helper()
141 +
142 + var out bytes.Buffer
143 + confProv := &mockConfigProvider{ch: make(chan confFile)}
144 +
145 + sd := &ServiceDiscovery{
146 + Logger: logger.New(),
147 + confProv: confProv,
148 + pluginName: testPluginName,
149 + fnReg: functions.NewManager(),
150 + discoverers: testDiscovererRegistry(),
151 + dyncfgApi: dyncfg.NewResponder(netdataapi.New(safewriter.New(&out))),
152 + seen: dyncfg.NewSeenCache[sdConfig](),
153 + exposed: dyncfg.NewExposedCache[sdConfig](),
154 + dyncfgCh: make(chan dyncfg.Function, 1),
155 + newPipeline: newWaitTestPipeline,
156 + runModePolicy: policy.RunModePolicy{},
157 + configDefaults: nil,
158 + }
159 + sd.sdCb = &sdCallbacks{sd: sd}
160 + sd.handler = dyncfg.NewHandler(dyncfg.HandlerOpts[sdConfig]{
161 + Logger: sd.Logger,
162 + API: sd.dyncfgApi,
163 + Seen: sd.seen,
164 + Exposed: sd.exposed,
165 + Callbacks: sd.sdCb,
166 + WaitKey: func(cfg sdConfig) string {
167 + return cfg.PipelineKey()
168 + },
169 + WaitTimeout: waitTimeout,
170 +
171 + Path: fmt.Sprintf(dyncfgSDPath, testPluginName),
172 + EnableFailCode: 422,
173 + JobCommands: []dyncfg.Command{
174 + dyncfg.CommandSchema,
175 + dyncfg.CommandGet,
176 + dyncfg.CommandEnable,
177 + dyncfg.CommandDisable,
178 + dyncfg.CommandUpdate,
179 + dyncfg.CommandTest,
180 + dyncfg.CommandUserconfig,
181 + },
182 + })
183 +
184 + send := func(context.Context, []*confgroup.Group) {}
185 + sd.mgr = NewPipelineManager(sd.Logger, sd.newPipeline, send)
186 +
187 + ctx, cancel := context.WithCancel(context.Background())
188 + sd.ctx = ctx
189 +
190 + done := make(chan struct{})
191 + go func() {
192 + defer close(done)
193 + sd.run(ctx)
194 + }()
195 +
196 + return sd, confProv.ch, cancel, done
197 +}
198 +
199 +func stopWaitTestServiceDiscovery(t *testing.T, sd *ServiceDiscovery, cancel context.CancelFunc, done <-chan struct{}) {
200 + t.Helper()
201 +
202 + cancel()
203 +
204 + select {
205 + case <-done:
206 + case <-time.After(5 * time.Second):
207 + t.Fatal("service discovery run loop did not stop")
208 + }
209 +
210 + sd.mgr.StopAll()
211 +}
212 +
213 +func newWaitTestPipeline(cfg pipeline.Config) (sdPipeline, error) {
214 + return newTestPipeline(cfg.Name), nil
215 +}
216 +
217 +func exposedExistsByKey(cache *dyncfg.ExposedCache[sdConfig], key string) bool {
218 + _, ok := cache.LookupByKey(key)
219 + return ok
220 +}
src/go/plugin/agent/jobmgr/cache.go
+8 -3
@@ -103,10 +103,15 @@ func (c *runningJobs) lookup(fullName string) (runtimeJob, bool) {
103 j, ok := c.items[fullName]
104 return j, ok
105 }
106 -func (c *runningJobs) forEach(fn func(fullName string, job runtimeJob)) {
107 - for k, j := range c.items {
108 - fn(k, j)
106 +func (c *runningJobs) snapshot() []runtimeJob {
107 + c.mux.Lock()
108 + defer c.mux.Unlock()
109 +
110 + jobs := make([]runtimeJob, 0, len(c.items))
111 + for _, job := range c.items {
112 + jobs = append(jobs, job)
113 }
114 + return jobs
115 }
116
117 func (c *retryingTasks) add(cfg confgroup.Config, retry *retryTask) {
src/go/plugin/agent/jobmgr/doc.go new
+24
@@ -0,0 +1,24 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +// Package jobmgr owns runtime orchestration for discovered and dyncfg-managed jobs.
4 +//
5 +// Concurrency contract (must remain true unless this document is updated):
6 +//
7 +// - Manager.run() is the serialized owner of dyncfg state transitions:
8 +// exposed configs, vnode map updates, and add/remove command application.
9 +//
10 +// - Discovery ingestion (runProcessConfGroups) does not mutate manager state directly.
11 +// It publishes add/remove intents through channels consumed by Manager.run().
12 +//
13 +// - runningJobs.items is protected by runningJobs.mux.
14 +// All traversals must use runningJobs.snapshot() and execute callbacks outside the lock.
15 +// No external API/job callbacks may execute while holding runningJobs.mux.
16 +//
17 +// - Function handlers run on framework/functions manager goroutine(s) and must avoid
18 +// cross-goroutine access to manager-owned mutable maps; async work uses manager-rooted
19 +// contexts and bounded worker limits where required.
20 +//
21 +// - Wait-for-decision gating for discovered configs is handler-owned. Manager.run()
22 +// delegates wait-step orchestration to dyncfg handler and keeps dyncfg command
23 +// processing progress even while waiting.
24 +package jobmgr
src/go/plugin/agent/jobmgr/dyncfg_collector.go
+63 -18
@@ -9,11 +9,13 @@ import (
9 "log/slog"
10 "slices"
11 "strings"
12 + "time"
13
14 "gopkg.in/yaml.v2"
15
16 "github.com/netdata/netdata/go/plugins/logger"
17 "github.com/netdata/netdata/go/plugins/pkg/netdataapi"
18 + "github.com/netdata/netdata/go/plugins/plugin/framework/collectorapi"
19 "github.com/netdata/netdata/go/plugins/plugin/framework/confgroup"
20 "github.com/netdata/netdata/go/plugins/plugin/framework/dyncfg"
21 )
@@ -66,14 +68,19 @@ func (m *Manager) exposedLookupByName(module, job string) (*dyncfg.Entry[confgro
68 return m.exposed.LookupByKey(key)
69 }
70
71 +type dyncfgCmdTestTask struct {
72 + fn dyncfg.Function
73 + moduleName string
74 + creator collectorapi.Creator
75 + cfg confgroup.Config
76 + timeout time.Duration
77 +}
78 +
79 func (m *Manager) dyncfgCollectorExec(fn dyncfg.Function) {
80 switch fn.Command() {
81 case dyncfg.CommandUserconfig:
82 m.dyncfgCmdUserconfig(fn)
83 return
74 - case dyncfg.CommandTest:
75 - m.dyncfgCmdTest(fn)
76 - return
84 case dyncfg.CommandSchema:
85 m.dyncfgCmdSchema(fn)
86 return
@@ -192,7 +199,7 @@ func (m *Manager) dyncfgCmdTest(fn dyncfg.Function) {
199 }
200
201 if cfg.Vnode() != "" {
195 - if _, ok := m.vnodes[cfg.Vnode()]; !ok {
202 + if _, ok := m.vnodes.Lookup(cfg.Vnode()); !ok {
203 m.Warningf("dyncfg: %s: module %s: vnode %s not found", cmd, mn, cfg.Vnode())
204 m.dyncfgApi.SendCodef(fn, 400, "The specified vnode '%s' is not registered.", cfg.Vnode())
205 return
@@ -202,36 +209,74 @@ func (m *Manager) dyncfgCmdTest(fn dyncfg.Function) {
209 cfg.SetModule(mn)
210 cfg.SetName(jn)
211
205 - job, err := newConfigModule(creator)
212 + if err := m.baseContext().Err(); err != nil {
213 + m.dyncfgApi.SendCodef(fn, 503, "Job manager is shutting down.")
214 + return
215 + }
216 +
217 + select {
218 + case m.cmdTestSem <- struct{}{}:
219 + task := dyncfgCmdTestTask{
220 + fn: fn,
221 + moduleName: mn,
222 + creator: creator,
223 + cfg: cfg,
224 + timeout: m.dyncfgCmdTestTimeout(fn),
225 + }
226 + m.cmdTestWG.Go(func() {
227 + m.runDyncfgCmdTest(task)
228 + })
229 + default:
230 + m.Warningf("dyncfg: %s: module %s: too many concurrent test requests", cmd, mn)
231 + m.dyncfgApi.SendCodef(fn, 503, "Too many concurrent test requests, try again later.")
232 + }
233 +}
234 +
235 +func (m *Manager) runDyncfgCmdTest(task dyncfgCmdTestTask) {
236 + defer func() { <-m.cmdTestSem }()
237 +
238 + job, err := newConfigModule(task.creator)
239 if err != nil {
207 - m.Warningf("dyncfg: %s: module %s: failed to create module: %v", cmd, mn, err)
208 - m.dyncfgApi.SendCodef(fn, 500, "Module %s instantiation failed: %v.", mn, err)
240 + m.Warningf("dyncfg: test: module %s: failed to create module: %v", task.moduleName, err)
241 + m.dyncfgApi.SendCodef(task.fn, 500, "Module %s instantiation failed: %v.", task.moduleName, err)
242 return
243 }
244
212 - if err := applyConfig(cfg, job); err != nil {
213 - m.Warningf("dyncfg: %s: module %s: failed to apply config: %v", cmd, mn, err)
214 - m.dyncfgApi.SendCodef(fn, 400, "Invalid configuration. Failed to apply configuration: %v.", err)
245 + cleanupCtx, cleanupCancel := context.WithTimeout(m.baseContext(), cmdTestWorkerDrainWait)
246 + defer cleanupCancel()
247 + defer job.Cleanup(cleanupCtx)
248 +
249 + if err := applyConfig(task.cfg, job); err != nil {
250 + m.Warningf("dyncfg: test: module %s: failed to apply config: %v", task.moduleName, err)
251 + m.dyncfgApi.SendCodef(task.fn, 400, "Invalid configuration. Failed to apply configuration: %v.", err)
252 return
253 }
254
255 job.GetBase().Logger = logger.New().With(
219 - slog.String("collector", cfg.Module()),
220 - slog.String("job", cfg.Name()),
256 + slog.String("collector", task.cfg.Module()),
257 + slog.String("job", task.cfg.Name()),
258 )
259
223 - defer job.Cleanup(context.Background())
260 + ctx, cancel := context.WithTimeout(m.baseContext(), task.timeout)
261 + defer cancel()
262
225 - if err := job.Init(context.Background()); err != nil {
226 - m.dyncfgApi.SendCodef(fn, 422, "Job initialization failed: %v", err)
263 + if err := job.Init(ctx); err != nil {
264 + m.dyncfgApi.SendCodef(task.fn, 422, "Job initialization failed: %v", err)
265 return
266 }
229 - if err := job.Check(context.Background()); err != nil {
230 - m.dyncfgApi.SendCodef(fn, 422, "Job check failed: %v", err)
267 + if err := job.Check(ctx); err != nil {
268 + m.dyncfgApi.SendCodef(task.fn, 422, "Job check failed: %v", err)
269 return
270 }
271
234 - m.dyncfgApi.SendCodef(fn, 200, "")
272 + m.dyncfgApi.SendCodef(task.fn, 200, "")
273 +}
274 +
275 +func (m *Manager) dyncfgCmdTestTimeout(fn dyncfg.Function) time.Duration {
276 + if timeout := fn.Fn().Timeout; timeout > 0 {
277 + return timeout
278 + }
279 + return cmdTestDefaultTimeout
280 }
281
282 func (m *Manager) dyncfgCmdSchema(fn dyncfg.Function) {
src/go/plugin/agent/jobmgr/dyncfg_collector_test.go
+69
@@ -4,10 +4,14 @@ package jobmgr
4
5 import (
6 "bytes"
7 + "context"
8 + "encoding/json"
9 "strings"
10 "testing"
11 + "time"
12
13 "github.com/stretchr/testify/assert"
14 + "github.com/stretchr/testify/require"
15
16 "github.com/netdata/netdata/go/plugins/pkg/netdataapi"
17 "github.com/netdata/netdata/go/plugins/pkg/safewriter"
@@ -54,3 +58,68 @@ func TestDyncfgConfigUserconfig_InvalidPayload_Returns400Only(t *testing.T) {
58 })
59 }
60 }
61 +
62 +func TestDyncfgCollectorExec_TestCommandQueued(t *testing.T) {
63 + mgr := New(Config{PluginName: testPluginName})
64 + mgr.ctx = context.Background()
65 + mgr.dyncfgCh = make(chan dyncfg.Function, 1)
66 +
67 + fn := dyncfg.NewFunction(functions.Function{
68 + UID: "queued-test",
69 + Args: []string{mgr.dyncfgModID("success"), "test", "job"},
70 + })
71 +
72 + mgr.dyncfgCollectorExec(fn)
73 +
74 + select {
75 + case queued := <-mgr.dyncfgCh:
76 + assert.Equal(t, dyncfg.CommandTest, queued.Command())
77 + assert.Equal(t, fn.UID(), queued.UID())
78 + case <-time.After(time.Second):
79 + t.Fatal("test command was not queued")
80 + }
81 +}
82 +
83 +func TestDyncfgCmdTest_WhenWorkerPoolFull_Returns503(t *testing.T) {
84 + var buf bytes.Buffer
85 +
86 + mgr := New(Config{PluginName: testPluginName})
87 + mgr.modules = prepareMockRegistry()
88 + mgr.ctx = context.Background()
89 + mgr.SetDyncfgResponder(dyncfg.NewResponder(netdataapi.New(safewriter.New(&buf))))
90 +
91 + for i := 0; i < cap(mgr.cmdTestSem); i++ {
92 + mgr.cmdTestSem <- struct{}{}
93 + }
94 +
95 + cfg := prepareDyncfgCfg("success", "job")
96 + payload, err := json.Marshal(cfg)
97 + require.NoError(t, err)
98 +
99 + fn := dyncfg.NewFunction(functions.Function{
100 + UID: "pool-full",
101 + ContentType: "application/json",
102 + Payload: payload,
103 + Args: []string{
104 + mgr.dyncfgModID("success"),
105 + string(dyncfg.CommandTest),
106 + "job",
107 + },
108 + })
109 +
110 + mgr.dyncfgCmdTest(fn)
111 +
112 + out := buf.String()
113 + assert.Equal(t, 1, strings.Count(out, "FUNCTION_RESULT_BEGIN pool-full"))
114 + assert.Contains(t, out, "\"status\":503")
115 +}
116 +
117 +func TestDyncfgCmdTestTimeout_RequestTimeoutOverridesDefault(t *testing.T) {
118 + mgr := New(Config{PluginName: testPluginName})
119 +
120 + withRequestTimeout := dyncfg.NewFunction(functions.Function{Timeout: 7 * time.Second})
121 + assert.Equal(t, 7*time.Second, mgr.dyncfgCmdTestTimeout(withRequestTimeout))
122 +
123 + withoutTimeout := dyncfg.NewFunction(functions.Function{})
124 + assert.Equal(t, cmdTestDefaultTimeout, mgr.dyncfgCmdTestTimeout(withoutTimeout))
125 +}
src/go/plugin/agent/jobmgr/dyncfg_vnode.go
+31 -17
@@ -116,7 +116,7 @@ func (m *Manager) dyncfgVnodeGet(fn dyncfg.Function) {
116 id := fn.ID()
117 name := strings.TrimPrefix(id, m.dyncfgVnodePrefixValue()+":")
118
119 - cfg, ok := m.vnodes[name]
119 + cfg, ok := m.vnodes.Lookup(name)
120 if !ok {
121 m.Warningf("dyncfg: %s: vnode %s not found", cmd, name)
122 m.dyncfgApi.SendCodef(fn, 404, "The specified vnode '%s' is not registered.", name)
@@ -171,19 +171,24 @@ func (m *Manager) dyncfgVnodeAdd(fn dyncfg.Function) {
171 return
172 }
173
174 - if orig, ok := m.vnodes[name]; ok && orig.Equal(cfg) {
174 + if orig, ok := m.vnodes.Lookup(name); ok && orig.Equal(cfg) {
175 m.dyncfgApi.SendCodef(fn, 202, "")
176 m.dyncfgVnodeJobCreate(cfg, dyncfg.StatusRunning)
177 return
178 }
179
180 - m.vnodes[name] = cfg
180 + _, _, err = m.vnodes.Upsert(cfg)
181 + if err != nil {
182 + m.Warningf("dyncfg: %s: vnode job %s: %v", cmd, name, err)
183 + m.dyncfgApi.SendCodef(fn, 400, "Failed to update vnode configuration: %v.", err)
184 + return
185 + }
186
182 - m.runningJobs.forEach(func(_ string, job runtimeJob) {
187 + for _, job := range m.runningJobs.snapshot() {
188 if job.Vnode().Name == name {
189 job.UpdateVnode(cfg)
190 }
186 - })
191 + }
192
193 m.dyncfgApi.SendCodef(fn, 202, "")
194 m.dyncfgVnodeJobCreate(cfg, dyncfg.StatusRunning)
@@ -195,7 +200,7 @@ func (m *Manager) dyncfgVnodeRemove(fn dyncfg.Function) {
200 id := fn.ID()
201 name := strings.TrimPrefix(id, m.dyncfgVnodePrefixValue()+":")
202
198 - vnode, ok := m.vnodes[name]
203 + vnode, ok := m.vnodes.Lookup(name)
204 if !ok {
205 m.Warningf("dyncfg: %s: vnode %s not found", cmd, name)
206 m.dyncfgApi.SendCodef(fn, 404, "The specified vnode '%s' is not registered.", name)
@@ -213,7 +218,7 @@ func (m *Manager) dyncfgVnodeRemove(fn dyncfg.Function) {
218 return
219 }
220
216 - delete(m.vnodes, name)
221 + _, _ = m.vnodes.Remove(name)
222
223 m.dyncfgApi.ConfigDelete(id)
224 m.dyncfgApi.SendCodef(fn, 200, "")
@@ -264,7 +269,7 @@ func (m *Manager) dyncfgVnodeUpdate(fn dyncfg.Function) {
269 id := fn.ID()
270 name := strings.TrimPrefix(id, m.dyncfgVnodePrefixValue()+":")
271
267 - orig, ok := m.vnodes[name]
272 + orig, ok := m.vnodes.Lookup(name)
273 if !ok {
274 m.Warningf("dyncfg: %s: vnode %s not found", cmd, name)
275 m.dyncfgApi.SendCodef(fn, 404, "The specified vnode '%s' is not registered.", name)
@@ -291,13 +296,18 @@ func (m *Manager) dyncfgVnodeUpdate(fn dyncfg.Function) {
296 return
297 }
298
294 - m.vnodes[name] = cfg
299 + _, _, err = m.vnodes.Upsert(cfg)
300 + if err != nil {
301 + m.Warningf("dyncfg: %s: vnode job %s: %v", cmd, name, err)
302 + m.dyncfgApi.SendCodef(fn, 400, "Failed to update vnode configuration: %v.", err)
303 + return
304 + }
305
296 - m.runningJobs.forEach(func(_ string, job runtimeJob) {
306 + for _, job := range m.runningJobs.snapshot() {
307 if job.Vnode().Name == name {
308 job.UpdateVnode(cfg)
309 }
300 - })
310 + }
311
312 m.dyncfgApi.SendCodef(fn, 202, "")
313 m.dyncfgVnodeJobCreate(cfg, dyncfg.StatusRunning)
@@ -331,18 +341,22 @@ func (m *Manager) dyncfgVnodeAffectedJobs(vnode string) string {
341 }
342
343 func (m *Manager) verifyVnodeUnique(newCfg *vnodes.VirtualNode) error {
334 - for _, cfg := range m.vnodes {
344 + var err error
345 + m.vnodes.ForEach(func(cfg *vnodes.VirtualNode) bool {
346 if cfg.Name == newCfg.Name {
336 - continue
347 + return true
348 }
349 if cfg.Hostname == newCfg.Hostname {
339 - return fmt.Errorf("duplicate virtual node name detected (job '%s')", cfg.Name)
350 + err = fmt.Errorf("duplicate virtual node name detected (job '%s')", cfg.Name)
351 + return false
352 }
353 if cfg.GUID == newCfg.GUID {
342 - return fmt.Errorf("duplicate virtual node guid detected (job '%s')", cfg.Name)
354 + err = fmt.Errorf("duplicate virtual node guid detected (job '%s')", cfg.Name)
355 + return false
356 }
344 - }
345 - return nil
357 + return true
358 + })
359 + return err
360 }
361
362 func dyncfgUpdateVnodeConfig(cfg *vnodes.VirtualNode, name string, fn dyncfg.Function) {
src/go/plugin/agent/jobmgr/filestatus.go
-5
@@ -119,11 +119,6 @@ func (s *fileStatus) add(cfg confgroup.Config, status string) {
119 }
120
121 s.items[cfg.Module()][s.jobKey(cfg)] = status
122 -
123 - select {
124 - case s.ch <- struct{}{}:
125 - default:
126 - }
122 }
123
124 func (s *fileStatus) remove(cfg confgroup.Config) {
src/go/plugin/agent/jobmgr/funcshandler.go
+152 -185
@@ -19,6 +19,86 @@ const (
19 paramJob = "__job"
20 )
21
22 +type methodParamResolver func(ctx context.Context, methodCfg *funcapi.MethodConfig, handler funcapi.MethodHandler, methodID string) ([]funcapi.ParamConfig, bool, error)
23 +
24 +type methodResponseWriter func(dataResp *funcapi.FunctionResponse, methodParams []funcapi.ParamConfig, updateEvery int)
25 +
26 +type methodExecutionInput struct {
27 + fn functions.Function
28 + moduleName string
29 + jobName string
30 + jobLabel string
31 + methodID string
32 + methodCfg *funcapi.MethodConfig
33 + job collectorapi.RuntimeJob
34 + jobGen uint64
35 + payload map[string]any
36 + argValues map[string][]string
37 + resolveParams methodParamResolver
38 + augmentParams func(funcapi.ResolvedParams)
39 + respond methodResponseWriter
40 +}
41 +
42 +// executeMethodRequest runs the common method execution pipeline for both
43 +// module-level and job-bound method handlers.
44 +func (m *Manager) executeMethodRequest(in methodExecutionInput) {
45 + ctx, cancel := context.WithTimeout(m.baseContext(), in.fn.Timeout)
46 + defer cancel()
47 +
48 + if !in.job.IsRunning() {
49 + m.respondError(in.fn, 503, "job '%s' is no longer running", in.jobLabel)
50 + return
51 + }
52 +
53 + creator, ok := m.moduleFuncs.getCreator(in.moduleName)
54 + if !ok || creator.MethodHandler == nil {
55 + m.respondError(in.fn, 500, "module '%s' does not implement MethodHandler", in.moduleName)
56 + return
57 + }
58 +
59 + handler := creator.MethodHandler(in.job)
60 + if handler == nil {
61 + m.respondError(in.fn, 500, "module '%s' returned nil handler for job '%s'", in.moduleName, in.jobName)
62 + return
63 + }
64 +
65 + methodParams, paramsFromJob, err := in.resolveParams(ctx, in.methodCfg, handler, in.methodID)
66 + if err != nil {
67 + m.respondError(in.fn, 503, "job '%s' cannot provide parameters: %v", in.jobLabel, err)
68 + return
69 + }
70 +
71 + if paramsFromJob {
72 + if err := validateParamValues(methodParams, in.argValues, in.payload, in.jobName); err != nil {
73 + m.respondError(in.fn, 400, "%v", err)
74 + return
75 + }
76 + }
77 +
78 + methodParamValues := make(map[string][]string, len(methodParams))
79 + for _, paramCfg := range methodParams {
80 + methodParamValues[paramCfg.ID] = paramValues(in.argValues, in.payload, paramCfg.ID)
81 + }
82 + resolvedParams := funcapi.ResolveParams(methodParams, methodParamValues)
83 + if in.augmentParams != nil {
84 + in.augmentParams(resolvedParams)
85 + }
86 +
87 + dataResp := handler.Handle(ctx, in.methodID, resolvedParams)
88 +
89 + if !m.moduleFuncs.verifyJobGeneration(in.moduleName, in.jobName, in.jobGen) {
90 + m.respondError(in.fn, 503, "job '%s' was replaced during request, please retry", in.jobLabel)
91 + return
92 + }
93 +
94 + updateEvery := 1
95 + if in.methodCfg.UpdateEvery > 1 {
96 + updateEvery = in.methodCfg.UpdateEvery
97 + }
98 +
99 + in.respond(dataResp, methodParams, updateEvery)
100 +}
101 +
102 // makeMethodFuncHandler creates a function handler for a module+method function (module:method).
103 func (m *Manager) makeMethodFuncHandler(moduleName, methodID string) func(functions.Function) {
104 return func(fn functions.Function) {
@@ -59,75 +139,27 @@ func (m *Manager) makeMethodFuncHandler(moduleName, methodID string) func(functi
139 return
140 }
141
62 - // Create context with timeout from function request
63 - // This ensures DB queries are cancelled if the function times out
64 - // NOTE: fn.Timeout is already a time.Duration (set by parser as seconds)
65 - // Do NOT multiply by time.Second again - that would create huge timeouts
66 - ctx, cancel := context.WithTimeout(context.Background(), fn.Timeout)
67 - defer cancel()
68 -
69 - // RACE CONDITION MITIGATION: Verify job is still running before handler
70 - // The job could be stopped between lookup and handler call
71 - if !job.IsRunning() {
72 - m.respondError(fn, 503, "job '%s' is no longer running", jobName)
73 - return
74 - }
75 -
76 - // Get the creator for this module to call MethodHandler
77 - creator, ok := m.moduleFuncs.getCreator(moduleName)
78 - if !ok || creator.MethodHandler == nil {
79 - m.respondError(fn, 500, "module '%s' does not implement MethodHandler", moduleName)
80 - return
81 - }
82 -
83 - // Get the handler for this job
84 - handler := creator.MethodHandler(job)
85 - if handler == nil {
86 - m.respondError(fn, 500, "module '%s' returned nil handler for job '%s'", moduleName, jobName)
87 - return
88 - }
89 -
90 - // Resolve method-specific required params (job-aware)
91 - methodParams, paramsFromJob, err := m.resolveMethodParamsForJob(ctx, moduleName, methodID, methodCfg, job, handler)
92 - if err != nil {
93 - m.respondError(fn, 503, "job '%s' cannot provide parameters: %v", jobName, err)
94 - return
95 - }
96 -
97 - // Validate provided param values when job-specific options are available
98 - if paramsFromJob {
99 - if err := validateParamValues(methodParams, argValues, payload, jobName); err != nil {
100 - m.respondError(fn, 400, "%v", err)
101 - return
102 - }
103 - }
104 -
105 - methodParamValues := make(map[string][]string, len(methodParams))
106 - for _, paramCfg := range methodParams {
107 - methodParamValues[paramCfg.ID] = paramValues(argValues, payload, paramCfg.ID)
108 - }
109 - resolvedParams := funcapi.ResolveParams(methodParams, methodParamValues)
110 - resolvedParams[paramJob] = resolvedJob
111 -
112 - // Route to the module's handler - get DATA ONLY response
113 - dataResp := handler.Handle(ctx, methodID, resolvedParams)
114 -
115 - // RACE CONDITION MITIGATION: Verify job was not replaced during handler execution
116 - // If a config reload replaced this job while we were querying, the response
117 - // might contain stale data or the connection might have been closed
118 - if !m.moduleFuncs.verifyJobGeneration(moduleName, jobName, jobGen) {
119 - // Job was replaced during our request - the response may be unreliable
120 - // Return error to prompt client to retry with new job instance
121 - m.respondError(fn, 503, "job '%s' was replaced during request, please retry", jobName)
122 - return
123 - }
124 -
125 - // Core injects required_params into the response before sending
126 - updateEvery := 1
127 - if methodCfg.UpdateEvery > 1 {
128 - updateEvery = methodCfg.UpdateEvery
129 - }
130 - m.respondWithParams(fn, moduleName, dataResp, methodParams, updateEvery)
142 + m.executeMethodRequest(methodExecutionInput{
143 + fn: fn,
144 + moduleName: moduleName,
145 + jobName: jobName,
146 + jobLabel: jobName,
147 + methodID: methodID,
148 + methodCfg: methodCfg,
149 + job: job,
150 + jobGen: jobGen,
151 + payload: payload,
152 + argValues: argValues,
153 + resolveParams: func(ctx context.Context, methodCfg *funcapi.MethodConfig, handler funcapi.MethodHandler, methodID string) ([]funcapi.ParamConfig, bool, error) {
154 + return m.resolveMethodParamsForJob(ctx, moduleName, methodID, methodCfg, job, handler)
155 + },
156 + augmentParams: func(resolvedParams funcapi.ResolvedParams) {
157 + resolvedParams[paramJob] = resolvedJob
158 + },
159 + respond: func(dataResp *funcapi.FunctionResponse, methodParams []funcapi.ParamConfig, updateEvery int) {
160 + m.respondWithParams(fn, moduleName, dataResp, methodParams, updateEvery)
161 + },
162 + })
163 }
164 }
165
@@ -167,7 +199,26 @@ func (m *Manager) handleMethodFuncInfo(moduleName, methodID string, fn functions
199
200 // respondWithParams wraps the module's data response with current required_params
201 func (m *Manager) respondWithParams(fn functions.Function, moduleName string, dataResp *funcapi.FunctionResponse, methodParams []funcapi.ParamConfig, updateEvery int) {
170 - // Nil guard: if module returns nil, treat as internal error
202 + m.respondMethodDataWithParams(
203 + fn,
204 + dataResp,
205 + methodParams,
206 + updateEvery,
207 + buildAcceptedParams,
208 + func(params []funcapi.ParamConfig) []map[string]any {
209 + return m.buildRequiredParams(moduleName, params)
210 + },
211 + )
212 +}
213 +
214 +func (m *Manager) respondMethodDataWithParams(
215 + fn functions.Function,
216 + dataResp *funcapi.FunctionResponse,
217 + methodParams []funcapi.ParamConfig,
218 + updateEvery int,
219 + buildAccepted func([]funcapi.ParamConfig) []string,
220 + buildRequired func([]funcapi.ParamConfig) []map[string]any,
221 +) {
222 if dataResp == nil {
223 m.respondError(fn, 500, "internal error: module returned nil response")
224 return
@@ -183,8 +234,6 @@ func (m *Manager) respondWithParams(fn functions.Function, moduleName string, da
234 paramsForResponse = funcapi.MergeParamConfigs(paramsForResponse, dataResp.RequiredParams)
235 }
236
186 - // Build the full response with injected required_params
187 - // Use dynamic sort options from response if provided (reflects actual DB capabilities)
237 resp := map[string]any{
238 "v": 3,
239 "update_every": updateEvery,
@@ -192,11 +241,10 @@ func (m *Manager) respondWithParams(fn functions.Function, moduleName string, da
241 "type": "table",
242 "has_history": false,
243 "help": dataResp.Help,
195 - "accepted_params": buildAcceptedParams(paramsForResponse),
196 - "required_params": m.buildRequiredParams(moduleName, paramsForResponse),
244 + "accepted_params": buildAccepted(paramsForResponse),
245 + "required_params": buildRequired(paramsForResponse),
246 }
247
199 - // Only include data fields when present (avoid null values on errors)
248 if dataResp.Columns != nil {
249 resp["columns"] = dataResp.Columns
250 }
@@ -206,8 +254,6 @@ func (m *Manager) respondWithParams(fn functions.Function, moduleName string, da
254 if dataResp.DefaultSortColumn != "" {
255 resp["default_sort_column"] = dataResp.DefaultSortColumn
256 }
209 -
210 - // Add chart configuration if provided
257 if len(dataResp.Charts) > 0 {
258 resp["charts"] = dataResp.Charts
259 }
@@ -477,6 +523,9 @@ func (m *Manager) makeJobMethodFuncHandler(moduleName, jobName, methodID string)
523 return
524 }
525
526 + payload := parsePayload(fn.Payload)
527 + argValues := parseArgsParams(fn.Args)
528 +
529 // Get job WITH generation for race condition detection
530 job, jobGen := m.moduleFuncs.getJobWithGeneration(moduleName, jobName)
531 if job == nil {
@@ -484,68 +533,24 @@ func (m *Manager) makeJobMethodFuncHandler(moduleName, jobName, methodID string)
533 return
534 }
535
487 - // Create context with timeout from function request
488 - ctx, cancel := context.WithTimeout(context.Background(), fn.Timeout)
489 - defer cancel()
490 -
491 - // Verify job is still running before calling handler
492 - if !job.IsRunning() {
493 - m.respondError(fn, 503, "job '%s:%s' is no longer running", moduleName, jobName)
494 - return
495 - }
496 -
497 - // Get the creator for this module to call MethodHandler
498 - creator, ok := m.moduleFuncs.getCreator(moduleName)
499 - if !ok || creator.MethodHandler == nil {
500 - m.respondError(fn, 500, "module '%s' does not implement MethodHandler", moduleName)
501 - return
502 - }
503 -
504 - // Get the handler for this job
505 - handler := creator.MethodHandler(job)
506 - if handler == nil {
507 - m.respondError(fn, 500, "module '%s' returned nil handler for job '%s'", moduleName, jobName)
508 - return
509 - }
510 -
511 - payload := parsePayload(fn.Payload)
512 - argValues := parseArgsParams(fn.Args)
513 -
514 - // Resolve method-specific required params
515 - methodParams, paramsFromJob, err := m.resolveJobMethodParams(ctx, methodCfg, handler, methodID)
516 - if err != nil {
517 - m.respondError(fn, 503, "job '%s:%s' cannot provide parameters: %v", moduleName, jobName, err)
518 - return
519 - }
520 -
521 - // Validate provided param values
522 - if paramsFromJob {
523 - if err := validateParamValues(methodParams, argValues, payload, jobName); err != nil {
524 - m.respondError(fn, 400, "%v", err)
525 - return
526 - }
527 - }
528 -
529 - methodParamValues := make(map[string][]string, len(methodParams))
530 - for _, paramCfg := range methodParams {
531 - methodParamValues[paramCfg.ID] = paramValues(argValues, payload, paramCfg.ID)
532 - }
533 - resolvedParams := funcapi.ResolveParams(methodParams, methodParamValues)
534 -
535 - // Route to the module's handler
536 - dataResp := handler.Handle(ctx, methodID, resolvedParams)
537 -
538 - // Verify job was not replaced during handler execution
539 - if !m.moduleFuncs.verifyJobGeneration(moduleName, jobName, jobGen) {
540 - m.respondError(fn, 503, "job '%s:%s' was replaced during request, please retry", moduleName, jobName)
541 - return
542 - }
543 -
544 - updateEvery := 1
545 - if methodCfg.UpdateEvery > 1 {
546 - updateEvery = methodCfg.UpdateEvery
547 - }
548 - m.respondJobMethodWithParams(fn, dataResp, methodParams, updateEvery)
536 + m.executeMethodRequest(methodExecutionInput{
537 + fn: fn,
538 + moduleName: moduleName,
539 + jobName: jobName,
540 + jobLabel: fmt.Sprintf("%s:%s", moduleName, jobName),
541 + methodID: methodID,
542 + methodCfg: methodCfg,
543 + job: job,
544 + jobGen: jobGen,
545 + payload: payload,
546 + argValues: argValues,
547 + resolveParams: func(ctx context.Context, methodCfg *funcapi.MethodConfig, handler funcapi.MethodHandler, methodID string) ([]funcapi.ParamConfig, bool, error) {
548 + return m.resolveJobMethodParams(ctx, methodCfg, handler, methodID)
549 + },
550 + respond: func(dataResp *funcapi.FunctionResponse, methodParams []funcapi.ParamConfig, updateEvery int) {
551 + m.respondJobMethodWithParams(fn, dataResp, methodParams, updateEvery)
552 + },
553 + })
554 }
555 }
556
@@ -599,52 +604,14 @@ func (m *Manager) resolveJobMethodParams(ctx context.Context, methodCfg *funcapi
604
605 // respondJobMethodWithParams wraps the module's data response for job-specific methods
606 func (m *Manager) respondJobMethodWithParams(fn functions.Function, dataResp *funcapi.FunctionResponse, methodParams []funcapi.ParamConfig, updateEvery int) {
602 - if dataResp == nil {
603 - m.respondError(fn, 500, "internal error: module returned nil response")
604 - return
605 - }
606 -
607 - if dataResp.Status >= 400 {
608 - m.respondError(fn, dataResp.Status, "%s", dataResp.Message)
609 - return
610 - }
611 -
612 - paramsForResponse := methodParams
613 - if len(dataResp.RequiredParams) > 0 {
614 - paramsForResponse = funcapi.MergeParamConfigs(paramsForResponse, dataResp.RequiredParams)
615 - }
616 -
617 - resp := map[string]any{
618 - "v": 3,
619 - "update_every": updateEvery,
620 - "status": dataResp.Status,
621 - "type": "table",
622 - "has_history": false,
623 - "help": dataResp.Help,
624 - "accepted_params": buildJobMethodAcceptedParams(paramsForResponse),
625 - "required_params": buildJobMethodRequiredParams(paramsForResponse),
626 - }
627 -
628 - if dataResp.Columns != nil {
629 - resp["columns"] = dataResp.Columns
630 - }
631 - if dataResp.Data != nil {
632 - resp["data"] = dataResp.Data
633 - }
634 - if dataResp.DefaultSortColumn != "" {
635 - resp["default_sort_column"] = dataResp.DefaultSortColumn
636 - }
637 - if len(dataResp.Charts) > 0 {
638 - resp["charts"] = dataResp.Charts
639 - }
640 - if len(dataResp.DefaultCharts) > 0 {
641 - resp["default_charts"] = dataResp.DefaultCharts.Build()
642 - }
643 - if len(dataResp.GroupBy) > 0 {
644 - resp["group_by"] = dataResp.GroupBy
645 - }
646 -
647 - m.respondJSON(fn, resp)
607 + m.respondMethodDataWithParams(
608 + fn,
609 + dataResp,
610 + methodParams,
611 + updateEvery,
612 + buildJobMethodAcceptedParams,
613 + buildJobMethodRequiredParams,
614 + )
615 }
616
617 // buildJobMethodAcceptedParams creates accepted_params for job-specific methods (no __job)
src/go/plugin/agent/jobmgr/job_factory.go new
+167
@@ -0,0 +1,167 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package jobmgr
4 +
5 +import (
6 + "fmt"
7 + "io"
8 + "os"
9 + "path/filepath"
10 +
11 + "github.com/netdata/netdata/go/plugins/logger"
12 + "github.com/netdata/netdata/go/plugins/plugin/agent/internal/naming"
13 + "github.com/netdata/netdata/go/plugins/plugin/framework/collectorapi"
14 + "github.com/netdata/netdata/go/plugins/plugin/framework/confgroup"
15 + "github.com/netdata/netdata/go/plugins/plugin/framework/jobruntime"
16 + "github.com/netdata/netdata/go/plugins/plugin/framework/metricsaudit"
17 + "github.com/netdata/netdata/go/plugins/plugin/framework/runtimecomp"
18 + "github.com/netdata/netdata/go/plugins/plugin/framework/vnodes"
19 +)
20 +
21 +// jobFactory builds runtime jobs from configs without mutating manager-owned runtime maps.
22 +type jobFactory struct {
23 + logger *logger.Logger
24 +
25 + pluginName string
26 + modules collectorapi.Registry
27 + vnodes *vnodeStore
28 + out io.Writer
29 +
30 + auditMode bool
31 + auditAnalyzer metricsaudit.Analyzer
32 + auditDataDir string
33 +
34 + runtimeService runtimecomp.Service
35 +}
36 +
37 +func newJobFactory(m *Manager) *jobFactory {
38 + return &jobFactory{
39 + logger: m.Logger,
40 +
41 + pluginName: m.pluginName,
42 + modules: m.modules,
43 + vnodes: m.vnodes,
44 + out: m.out,
45 +
46 + auditMode: m.auditMode,
47 + auditAnalyzer: m.auditAnalyzer,
48 + auditDataDir: m.auditDataDir,
49 +
50 + runtimeService: m.runtimeService,
51 + }
52 +}
53 +
54 +func (f *jobFactory) create(cfg confgroup.Config) (runtimeJob, error) {
55 + creator, ok := f.modules[cfg.Module()]
56 + if !ok {
57 + return nil, fmt.Errorf("can not find %s module", cfg.Module())
58 + }
59 +
60 + functionOnly := creator.FunctionOnly || cfg.FunctionOnly()
61 + if cfg.FunctionOnly() && creator.Methods == nil && creator.JobMethods == nil {
62 + return nil, fmt.Errorf("function_only is set but %s module has no methods defined", cfg.Module())
63 + }
64 +
65 + var vnode *vnodes.VirtualNode
66 + if cfg.Vnode() != "" {
67 + n, ok := f.vnodes.Lookup(cfg.Vnode())
68 + if !ok || n == nil {
69 + return nil, fmt.Errorf("vnode '%s' is not found", cfg.Vnode())
70 + }
71 + vnode = n
72 + }
73 +
74 + f.logger.Debugf("creating %s[%s] job, config: %v", cfg.Module(), cfg.Name(), cfg)
75 +
76 + if creator.CreateV2 != nil {
77 + return f.createV2(cfg, creator, functionOnly, vnode)
78 + }
79 + return f.createV1(cfg, creator, functionOnly, vnode)
80 +}
81 +
82 +func (f *jobFactory) createV2(cfg confgroup.Config, creator collectorapi.Creator, functionOnly bool, vnode *vnodes.VirtualNode) (runtimeJob, error) {
83 + mod := creator.CreateV2()
84 + if mod == nil {
85 + return nil, fmt.Errorf("module %s CreateV2 returned nil", cfg.Module())
86 + }
87 + if err := applyConfig(cfg, mod); err != nil {
88 + return nil, err
89 + }
90 +
91 + jobCfg := jobruntime.JobV2Config{
92 + PluginName: f.pluginName,
93 + Name: cfg.Name(),
94 + ModuleName: cfg.Module(),
95 + FullName: cfg.FullName(),
96 + UpdateEvery: cfg.UpdateEvery(),
97 + AutoDetectEvery: cfg.AutoDetectionRetry(),
98 + IsStock: cfg.SourceType() == "stock",
99 + Labels: makeLabels(cfg),
100 + Out: f.out,
101 + Module: mod,
102 + FunctionOnly: functionOnly,
103 + RuntimeService: f.runtimeService,
104 + }
105 + if vnode != nil {
106 + jobCfg.Vnode = *vnode.Copy()
107 + }
108 + return jobruntime.NewJobV2(jobCfg), nil
109 +}
110 +
111 +func (f *jobFactory) createV1(cfg confgroup.Config, creator collectorapi.Creator, functionOnly bool, vnode *vnodes.VirtualNode) (runtimeJob, error) {
112 + if creator.Create == nil {
113 + return nil, fmt.Errorf("module %s has no compatible creator", cfg.Module())
114 + }
115 +
116 + jobCaptureDir, err := f.createV1CaptureDir(cfg)
117 + if err != nil {
118 + return nil, err
119 + }
120 +
121 + mod := creator.Create()
122 + if err := applyConfig(cfg, mod); err != nil {
123 + return nil, err
124 + }
125 +
126 + if f.auditAnalyzer != nil && jobCaptureDir != "" {
127 + f.auditAnalyzer.RegisterJob(cfg.Name(), cfg.Module(), jobCaptureDir)
128 + }
129 + if jobCaptureDir != "" {
130 + if captureAware, ok := mod.(metricsaudit.Capturable); ok {
131 + captureAware.EnableCaptureArtifacts(jobCaptureDir)
132 + }
133 + }
134 +
135 + jobCfg := jobruntime.JobConfig{
136 + PluginName: f.pluginName,
137 + Name: cfg.Name(),
138 + ModuleName: cfg.Module(),
139 + FullName: cfg.FullName(),
140 + UpdateEvery: cfg.UpdateEvery(),
141 + AutoDetectEvery: cfg.AutoDetectionRetry(),
142 + Priority: cfg.Priority(),
143 + Labels: makeLabels(cfg),
144 + IsStock: cfg.SourceType() == "stock",
145 + Module: mod,
146 + Out: f.out,
147 + AuditMode: f.auditMode,
148 + AuditAnalyzer: f.auditAnalyzer,
149 + FunctionOnly: functionOnly,
150 + }
151 + if vnode != nil {
152 + jobCfg.Vnode = *vnode.Copy()
153 + }
154 +
155 + return jobruntime.NewJob(jobCfg), nil
156 +}
157 +
158 +func (f *jobFactory) createV1CaptureDir(cfg confgroup.Config) (string, error) {
159 + if f.auditDataDir == "" {
160 + return "", nil
161 + }
162 + jobCaptureDir := filepath.Join(f.auditDataDir, naming.Sanitize(cfg.Module()), naming.Sanitize(cfg.Name()))
163 + if err := os.MkdirAll(jobCaptureDir, 0o755); err != nil {
164 + return "", fmt.Errorf("creating audit directory: %w", err)
165 + }
166 + return jobCaptureDir, nil
167 +}
src/go/plugin/agent/jobmgr/manager.go
+92 -142
@@ -7,8 +7,6 @@ import (
7 "fmt"
8 "io"
9 "log/slog"
10 - "os"
11 - "path/filepath"
10 "slices"
11 "sync"
12 "time"
@@ -17,13 +15,11 @@ import (
15 "github.com/netdata/netdata/go/plugins/pkg/funcapi"
16 "github.com/netdata/netdata/go/plugins/pkg/netdataapi"
17 "github.com/netdata/netdata/go/plugins/pkg/ticker"
20 - "github.com/netdata/netdata/go/plugins/plugin/agent/internal/naming"
18 "github.com/netdata/netdata/go/plugins/plugin/agent/policy"
19 "github.com/netdata/netdata/go/plugins/plugin/framework/collectorapi"
20 "github.com/netdata/netdata/go/plugins/plugin/framework/confgroup"
21 "github.com/netdata/netdata/go/plugins/plugin/framework/dyncfg"
22 "github.com/netdata/netdata/go/plugins/plugin/framework/functions"
26 - "github.com/netdata/netdata/go/plugins/plugin/framework/jobruntime"
23 "github.com/netdata/netdata/go/plugins/plugin/framework/metricsaudit"
24 "github.com/netdata/netdata/go/plugins/plugin/framework/runtimecomp"
25 "github.com/netdata/netdata/go/plugins/plugin/framework/vnodes"
@@ -47,6 +43,13 @@ type Config struct {
43 RuntimeService runtimecomp.Service
44 }
45
46 +const (
47 + waitDecisionTimeout = 5 * time.Second
48 + cmdTestWorkerCap = 4
49 + cmdTestDefaultTimeout = 60 * time.Second
50 + cmdTestWorkerDrainWait = 5 * time.Second
51 +)
52 +
53 func New(cfg Config) *Manager {
54 out := cfg.Out
55 if out == nil {
@@ -77,7 +80,7 @@ func New(cfg Config) *Manager {
80 configDefaults: cfg.ConfigDefaults,
81 varLibDir: cfg.VarLibDir,
82 fnReg: fnReg,
80 - vnodes: vnodesReg,
83 + vnodes: newVnodeStore(vnodesReg),
84
85 auditMode: cfg.AuditMode,
86 auditAnalyzer: cfg.AuditAnalyzer,
@@ -92,11 +95,12 @@ func New(cfg Config) *Manager {
95 runningJobs: newRunningJobsCache(),
96 retryingTasks: newRetryingTasksCache(),
97
95 - started: make(chan struct{}),
96 - addCh: make(chan confgroup.Config),
97 - rmCh: make(chan confgroup.Config),
98 - dyncfgCh: make(chan dyncfg.Function),
99 - dyncfgApi: api,
98 + started: make(chan struct{}),
99 + addCh: make(chan confgroup.Config),
100 + rmCh: make(chan confgroup.Config),
101 + dyncfgCh: make(chan dyncfg.Function),
102 + cmdTestSem: make(chan struct{}, cmdTestWorkerCap),
103 + dyncfgApi: api,
104 }
105
106 mgr.collectorCb = &collectorCallbacks{mgr: mgr}
@@ -109,6 +113,7 @@ func New(cfg Config) *Manager {
113 WaitKey: func(cfg confgroup.Config) string {
114 return cfg.FullName()
115 },
116 + WaitTimeout: waitDecisionTimeout,
117
118 Path: fmt.Sprintf(dyncfgCollectorPath, cfg.PluginName),
119 EnableFailCode: 200,
@@ -144,7 +149,7 @@ type Manager struct {
149 configDefaults confgroup.Registry
150 varLibDir string
151 fnReg FunctionRegistry
147 - vnodes map[string]*vnodes.VirtualNode
152 + vnodes *vnodeStore
153
154 // Metrics-audit mode.
155 auditMode bool
@@ -163,11 +168,13 @@ type Manager struct {
168 handler *dyncfg.Handler[confgroup.Config]
169 collectorCb *collectorCallbacks
170
166 - ctx context.Context
167 - started chan struct{}
168 - addCh chan confgroup.Config
169 - rmCh chan confgroup.Config
170 - dyncfgCh chan dyncfg.Function
171 + ctx context.Context
172 + started chan struct{}
173 + addCh chan confgroup.Config
174 + rmCh chan confgroup.Config
175 + dyncfgCh chan dyncfg.Function
176 + cmdTestSem chan struct{}
177 + cmdTestWG sync.WaitGroup
178
179 dyncfgApi *dyncfg.Responder
180
@@ -189,9 +196,10 @@ func (m *Manager) Run(ctx context.Context, in chan []*confgroup.Group) {
196
197 m.dyncfgVnodeModuleCreate()
198
192 - for _, cfg := range m.vnodes {
199 + m.vnodes.ForEach(func(cfg *vnodes.VirtualNode) bool {
200 m.dyncfgVnodeJobCreate(cfg, dyncfg.StatusRunning)
194 - }
201 + return true
202 + })
203
204 for name, creator := range m.modules {
205 m.dyncfgCollectorModuleCreate(name)
@@ -313,11 +321,21 @@ func (m *Manager) runProcessConfGroups(in chan []*confgroup.Group) {
321 func (m *Manager) run() {
322 for {
323 if m.handler.WaitingForDecision() {
316 - select {
317 - case <-m.ctx.Done():
324 + step, ok := m.handler.NextWaitDecisionStep(m.ctx, m.dyncfgCh)
325 + if !ok {
326 return
319 - case fn := <-m.dyncfgCh:
320 - m.dyncfgSeqExec(fn)
327 + }
328 + if step.HasCommand {
329 + m.dyncfgSeqExec(step.Command)
330 + continue
331 + }
332 + if step.TimedOut {
333 + m.Errorf(
334 + "dyncfg: timed out waiting for enable/disable decision for '%s' (elapsed=%s threshold=%s); keeping status 'accepted' and continuing",
335 + step.Timeout.Key,
336 + step.Timeout.Elapsed,
337 + step.Timeout.Threshold,
338 + )
339 }
340 } else {
341 select {
@@ -391,9 +409,9 @@ func (m *Manager) runNotifyRunningJobs() {
409 case <-m.ctx.Done():
410 return
411 case clock := <-tk.C:
394 - m.runningJobs.lock()
395 - m.runningJobs.forEach(func(_ string, job runtimeJob) { job.Tick(clock) })
396 - m.runningJobs.unlock()
412 + for _, job := range m.runningJobs.snapshot() {
413 + job.Tick(clock)
414 + }
415 }
416 }
417 }
@@ -401,11 +419,11 @@ func (m *Manager) runNotifyRunningJobs() {
419 func (m *Manager) startRunningJob(job runtimeJob) {
420 m.stopRunningJob(job.FullName())
421
404 - m.runningJobs.lock()
405 - defer m.runningJobs.unlock()
406 -
422 go job.Start()
423 +
424 + m.runningJobs.lock()
425 m.runningJobs.add(job.FullName(), job)
426 + m.runningJobs.unlock()
427
428 // Track job for module function routing.
429 m.moduleFuncs.addJob(job.ModuleName(), job.Name(), job)
@@ -455,16 +473,31 @@ func (m *Manager) cleanup() {
473 }
474 }
475
458 - m.runningJobs.lock()
459 - defer m.runningJobs.unlock()
476 + for _, job := range m.runningJobs.snapshot() {
477 + m.stopRunningJob(job.FullName())
478 + }
479
461 - m.runningJobs.forEach(func(_ string, job runtimeJob) {
462 - job.Stop()
463 - })
480 + m.waitCmdTestWorkers()
481 +}
482 +
483 +func (m *Manager) waitCmdTestWorkers() {
484 + done := make(chan struct{})
485 + go func() {
486 + m.cmdTestWG.Wait()
487 + close(done)
488 + }()
489 +
490 + select {
491 + case <-done:
492 + case <-time.After(cmdTestWorkerDrainWait):
493 + m.Warningf("dyncfg: timeout waiting %s for command test workers to drain", cmdTestWorkerDrainWait)
494 + }
495 }
496
497 // registerJobMethods registers methods for a specific job with Netdata
498 func (m *Manager) registerJobMethods(job collectorapi.RuntimeJob, methods []funcapi.MethodConfig) {
499 + planned := make(map[string]struct{}, len(methods))
500 +
501 for _, method := range methods {
502 if method.ID == "" {
503 m.Warningf("skipping job method registration for %s[%s]: empty method ID", job.ModuleName(), job.Name())
@@ -473,6 +506,25 @@ func (m *Manager) registerJobMethods(job collectorapi.RuntimeJob, methods []func
506
507 funcName := fmt.Sprintf("%s:%s", job.ModuleName(), method.ID)
508
509 + if _, exists := planned[method.ID]; exists {
510 + m.Errorf("job method registration aborted for %s[%s]: duplicate method ID in batch ('%s')", job.ModuleName(), job.Name(), funcName)
511 + return
512 + }
513 + planned[method.ID] = struct{}{}
514 +
515 + if collision, exists := m.moduleFuncs.findMethodCollision(job.ModuleName(), job.Name(), method.ID); exists {
516 + m.Errorf("job method registration aborted for %s[%s]: collision on '%s' (%s)", job.ModuleName(), job.Name(), funcName, collision)
517 + return
518 + }
519 + }
520 +
521 + for _, method := range methods {
522 + if method.ID == "" {
523 + continue
524 + }
525 +
526 + funcName := fmt.Sprintf("%s:%s", job.ModuleName(), method.ID)
527 +
528 // Register Go handler for this function
529 m.fnReg.Register(funcName, m.makeJobMethodFuncHandler(job.ModuleName(), job.Name(), method.ID))
530
@@ -532,117 +584,15 @@ func (m *Manager) unregisterJobMethods(job collectorapi.RuntimeJob) {
584 m.moduleFuncs.unregisterJobMethods(job.ModuleName(), job.Name())
585 }
586
535 -func (m *Manager) createCollectorJob(cfg confgroup.Config) (runtimeJob, error) {
536 - creator, ok := m.modules[cfg.Module()]
537 - if !ok {
538 - return nil, fmt.Errorf("can not find %s module", cfg.Module())
539 - }
540 -
541 - // Determine if job is function-only (module-level OR config-level)
542 - functionOnly := creator.FunctionOnly || cfg.FunctionOnly()
543 -
544 - // Reject if config sets function_only but module has no methods
545 - // Note: module-level FunctionOnly without Methods is caught at registration time
546 - if cfg.FunctionOnly() && creator.Methods == nil && creator.JobMethods == nil {
547 - return nil, fmt.Errorf("function_only is set but %s module has no methods defined", cfg.Module())
548 - }
549 -
550 - var vnode *vnodes.VirtualNode
551 -
552 - if cfg.Vnode() != "" {
553 - n, ok := m.vnodes[cfg.Vnode()]
554 - if !ok || n == nil {
555 - return nil, fmt.Errorf("vnode '%s' is not found", cfg.Vnode())
556 - }
557 - vnode = n
558 - }
559 -
560 - m.Debugf("creating %s[%s] job, config: %v", cfg.Module(), cfg.Name(), cfg)
561 -
562 - useV2 := creator.CreateV2 != nil
563 -
564 - var jobCaptureDir string
565 - if m.auditDataDir != "" && !useV2 {
566 - jobCaptureDir = filepath.Join(m.auditDataDir, naming.Sanitize(cfg.Module()), naming.Sanitize(cfg.Name()))
567 - if err := os.MkdirAll(jobCaptureDir, 0o755); err != nil {
568 - return nil, fmt.Errorf("creating audit directory: %w", err)
569 - }
570 - }
571 -
572 - if useV2 {
573 - mod := creator.CreateV2()
574 - if mod == nil {
575 - return nil, fmt.Errorf("module %s CreateV2 returned nil", cfg.Module())
576 - }
577 - if err := applyConfig(cfg, mod); err != nil {
578 - return nil, err
579 - }
580 -
581 - jobCfg := jobruntime.JobV2Config{
582 - PluginName: m.pluginName,
583 - Name: cfg.Name(),
584 - ModuleName: cfg.Module(),
585 - FullName: cfg.FullName(),
586 - UpdateEvery: cfg.UpdateEvery(),
587 - AutoDetectEvery: cfg.AutoDetectionRetry(),
588 - IsStock: cfg.SourceType() == "stock",
589 - Labels: makeLabels(cfg),
590 - Out: m.out,
591 - Module: mod,
592 - FunctionOnly: functionOnly,
593 - RuntimeService: m.runtimeService,
594 - }
595 - if vnode != nil {
596 - jobCfg.Vnode = *vnode.Copy()
597 - }
598 - return jobruntime.NewJobV2(jobCfg), nil
599 - }
600 -
601 - if creator.Create == nil {
602 - return nil, fmt.Errorf("module %s has no compatible creator", cfg.Module())
603 - }
604 -
605 - mod := creator.Create()
606 -
607 - if err := applyConfig(cfg, mod); err != nil {
608 - return nil, err
609 - }
610 -
611 - if m.auditAnalyzer != nil && jobCaptureDir != "" {
612 - // Auditing hooks are V1-only; V2 jobs are intentionally excluded.
613 - m.auditAnalyzer.RegisterJob(cfg.Name(), cfg.Module(), jobCaptureDir)
587 +func (m *Manager) baseContext() context.Context {
588 + if m.ctx != nil {
589 + return m.ctx
590 }
591 + return context.Background()
592 +}
593
616 - if jobCaptureDir != "" {
617 - if captureAware, ok := mod.(metricsaudit.Capturable); ok {
618 - captureAware.EnableCaptureArtifacts(jobCaptureDir)
619 - }
620 - }
621 -
622 - jobCfg := jobruntime.JobConfig{
623 - PluginName: m.pluginName,
624 - Name: cfg.Name(),
625 - ModuleName: cfg.Module(),
626 - FullName: cfg.FullName(),
627 - UpdateEvery: cfg.UpdateEvery(),
628 - AutoDetectEvery: cfg.AutoDetectionRetry(),
629 - Priority: cfg.Priority(),
630 - Labels: makeLabels(cfg),
631 - IsStock: cfg.SourceType() == "stock",
632 - Module: mod,
633 - Out: m.out,
634 - AuditMode: m.auditMode,
635 - AuditAnalyzer: m.auditAnalyzer,
636 - FunctionOnly: functionOnly,
637 - }
638 -
639 - if vnode != nil {
640 - jobCfg.Vnode = *vnode.Copy()
641 - }
642 -
643 - job := jobruntime.NewJob(jobCfg)
644 -
645 - return job, nil
594 +func (m *Manager) createCollectorJob(cfg confgroup.Config) (runtimeJob, error) {
595 + return newJobFactory(m).create(cfg)
596 }
597
598 func runRetryTask(ctx context.Context, out chan<- confgroup.Config, cfg confgroup.Config) {
src/go/plugin/agent/jobmgr/manager_process_test.go
+225
@@ -4,12 +4,19 @@ package jobmgr
4
5 import (
6 "context"
7 + "sync"
8 "testing"
9 "time"
10
11 "github.com/stretchr/testify/assert"
12 + "github.com/stretchr/testify/require"
13
14 + "github.com/netdata/netdata/go/plugins/pkg/funcapi"
15 + "github.com/netdata/netdata/go/plugins/plugin/framework/collectorapi"
16 "github.com/netdata/netdata/go/plugins/plugin/framework/confgroup"
17 + "github.com/netdata/netdata/go/plugins/plugin/framework/dyncfg"
18 + "github.com/netdata/netdata/go/plugins/plugin/framework/functions"
19 + "github.com/netdata/netdata/go/plugins/plugin/framework/vnodes"
20 )
21
22 func TestRunProcessConfGroups_ChannelCloseDoesNotSpin(t *testing.T) {
@@ -58,3 +65,221 @@ func TestRunProcessConfGroups_ChannelCloseDoesNotSpin(t *testing.T) {
65 })
66 }
67 }
68 +
69 +func TestRun_WaitTimeoutClearsGateAndKeepsAccepted(t *testing.T) {
70 + mgr := New(Config{PluginName: testPluginName})
71 + mgr.modules = prepareMockRegistry()
72 +
73 + ctx, cancel := context.WithCancel(context.Background())
74 + defer cancel()
75 + mgr.ctx = ctx
76 +
77 + done := make(chan struct{})
78 + go func() {
79 + mgr.run()
80 + close(done)
81 + }()
82 + defer func() {
83 + cancel()
84 + select {
85 + case <-done:
86 + case <-time.After(2 * time.Second):
87 + t.Fatal("run did not stop after cancel")
88 + }
89 + }()
90 +
91 + cfg1 := prepareStockCfg("success", "wait1")
92 + cfg2 := prepareStockCfg("success", "wait2")
93 +
94 + mgr.addCh <- cfg1
95 + require.Eventually(t, mgr.handler.WaitingForDecision, time.Second, 10*time.Millisecond)
96 +
97 + secondSent := make(chan struct{})
98 + go func() {
99 + mgr.addCh <- cfg2
100 + close(secondSent)
101 + }()
102 +
103 + select {
104 + case <-secondSent:
105 + t.Fatal("second add was processed before wait timeout")
106 + case <-time.After(500 * time.Millisecond):
107 + }
108 +
109 + select {
110 + case <-secondSent:
111 + case <-time.After(7 * time.Second):
112 + t.Fatal("second add did not progress after wait timeout")
113 + }
114 +
115 + entry1, ok := mgr.exposed.LookupByKey(cfg1.ExposedKey())
116 + require.True(t, ok, "first config must stay exposed after timeout")
117 + assert.Equal(t, dyncfg.StatusAccepted, entry1.Status)
118 +}
119 +
120 +func TestRunNotifyRunningJobs_TickOutsideLock(t *testing.T) {
121 + mgr := New(Config{PluginName: testPluginName})
122 +
123 + ctx, cancel := context.WithCancel(context.Background())
124 + defer cancel()
125 + mgr.ctx = ctx
126 +
127 + job := &lockProbeJob{
128 + fullName: "success_lockprobe",
129 + moduleName: "success",
130 + name: "lockprobe",
131 + tickStarted: make(chan struct{}),
132 + tickRelease: make(chan struct{}),
133 + }
134 +
135 + mgr.runningJobs.lock()
136 + mgr.runningJobs.add(job.FullName(), job)
137 + mgr.runningJobs.unlock()
138 +
139 + done := make(chan struct{})
140 + go func() {
141 + mgr.runNotifyRunningJobs()
142 + close(done)
143 + }()
144 +
145 + select {
146 + case <-job.tickStarted:
147 + case <-time.After(2 * time.Second):
148 + t.Fatal("tick did not start")
149 + }
150 +
151 + stopDone := make(chan struct{})
152 + go func() {
153 + mgr.stopRunningJob(job.FullName())
154 + close(stopDone)
155 + }()
156 +
157 + select {
158 + case <-stopDone:
159 + case <-time.After(300 * time.Millisecond):
160 + t.Fatal("stopRunningJob blocked while Tick was in progress")
161 + }
162 +
163 + close(job.tickRelease)
164 + cancel()
165 +
166 + select {
167 + case <-done:
168 + case <-time.After(2 * time.Second):
169 + t.Fatal("runNotifyRunningJobs did not stop")
170 + }
171 +}
172 +
173 +func TestRegisterJobMethods_FailFastOnCollisionWithStaticMethod(t *testing.T) {
174 + fnReg := &recordingFunctionRegistry{}
175 + mgr := New(Config{PluginName: testPluginName, FnReg: fnReg})
176 + mgr.moduleFuncs.registerModule("mod", collectorapi.Creator{
177 + Methods: func() []funcapi.MethodConfig {
178 + return []funcapi.MethodConfig{{ID: "dup"}}
179 + },
180 + })
181 +
182 + job := &lockProbeJob{fullName: "mod_job1", moduleName: "mod", name: "job1"}
183 + mgr.registerJobMethods(job, []funcapi.MethodConfig{{ID: "dup"}})
184 +
185 + assert.Empty(t, fnReg.registeredNames())
186 + assert.Empty(t, mgr.moduleFuncs.getJobMethods("mod", "job1"))
187 +}
188 +
189 +func TestRegisterJobMethods_FailFastOnCollisionWithOtherJob(t *testing.T) {
190 + fnReg := &recordingFunctionRegistry{}
191 + mgr := New(Config{PluginName: testPluginName, FnReg: fnReg})
192 + mgr.moduleFuncs.registerModule("mod", collectorapi.Creator{})
193 + mgr.moduleFuncs.registerJobMethods("mod", "jobA", []funcapi.MethodConfig{{ID: "dup"}})
194 +
195 + job := &lockProbeJob{fullName: "mod_jobB", moduleName: "mod", name: "jobB"}
196 + mgr.registerJobMethods(job, []funcapi.MethodConfig{{ID: "dup"}})
197 +
198 + assert.Empty(t, fnReg.registeredNames())
199 + assert.Empty(t, mgr.moduleFuncs.getJobMethods("mod", "jobB"))
200 +}
201 +
202 +func TestRegisterJobMethods_FailFastOnDuplicateWithinBatch(t *testing.T) {
203 + fnReg := &recordingFunctionRegistry{}
204 + mgr := New(Config{PluginName: testPluginName, FnReg: fnReg})
205 + mgr.moduleFuncs.registerModule("mod", collectorapi.Creator{})
206 +
207 + job := &lockProbeJob{fullName: "mod_job1", moduleName: "mod", name: "job1"}
208 + mgr.registerJobMethods(job, []funcapi.MethodConfig{
209 + {ID: "dup"},
210 + {ID: "dup"},
211 + })
212 +
213 + assert.Empty(t, fnReg.registeredNames())
214 + assert.Empty(t, mgr.moduleFuncs.getJobMethods("mod", "job1"))
215 +}
216 +
217 +func TestRegisterJobMethods_SuccessCommitsAllMethods(t *testing.T) {
218 + fnReg := &recordingFunctionRegistry{}
219 + mgr := New(Config{PluginName: testPluginName, FnReg: fnReg})
220 + mgr.moduleFuncs.registerModule("mod", collectorapi.Creator{})
221 +
222 + job := &lockProbeJob{fullName: "mod_job1", moduleName: "mod", name: "job1"}
223 + mgr.registerJobMethods(job, []funcapi.MethodConfig{
224 + {ID: "a"},
225 + {ID: "b"},
226 + })
227 +
228 + assert.ElementsMatch(t, []string{"mod:a", "mod:b"}, fnReg.registeredNames())
229 + assert.Len(t, mgr.moduleFuncs.getJobMethods("mod", "job1"), 2)
230 +}
231 +
232 +type lockProbeJob struct {
233 + fullName string
234 + moduleName string
235 + name string
236 +
237 + tickOnce sync.Once
238 + stopOnce sync.Once
239 + tickStarted chan struct{}
240 + tickRelease chan struct{}
241 +}
242 +
243 +func (j *lockProbeJob) FullName() string { return j.fullName }
244 +func (j *lockProbeJob) ModuleName() string { return j.moduleName }
245 +func (j *lockProbeJob) Name() string { return j.name }
246 +func (j *lockProbeJob) Collector() any { return nil }
247 +func (j *lockProbeJob) Start() {}
248 +func (j *lockProbeJob) Stop() { j.stopOnce.Do(func() {}) }
249 +func (j *lockProbeJob) Tick(_ int) {
250 + j.tickOnce.Do(func() {
251 + close(j.tickStarted)
252 + <-j.tickRelease
253 + })
254 +}
255 +func (j *lockProbeJob) AutoDetection() error { return nil }
256 +func (j *lockProbeJob) AutoDetectionEvery() int { return 0 }
257 +func (j *lockProbeJob) RetryAutoDetection() bool { return false }
258 +func (j *lockProbeJob) Cleanup() {}
259 +func (j *lockProbeJob) IsRunning() bool { return true }
260 +func (j *lockProbeJob) Panicked() bool { return false }
261 +func (j *lockProbeJob) Vnode() vnodes.VirtualNode { return vnodes.VirtualNode{} }
262 +func (j *lockProbeJob) UpdateVnode(_ *vnodes.VirtualNode) {}
263 +
264 +type recordingFunctionRegistry struct {
265 + mu sync.Mutex
266 + registered []string
267 +}
268 +
269 +func (r *recordingFunctionRegistry) Register(name string, _ func(functions.Function)) {
270 + r.mu.Lock()
271 + r.registered = append(r.registered, name)
272 + r.mu.Unlock()
273 +}
274 +
275 +func (r *recordingFunctionRegistry) Unregister(string) {}
276 +func (r *recordingFunctionRegistry) RegisterPrefix(string, string, func(functions.Function)) {}
277 +func (r *recordingFunctionRegistry) UnregisterPrefix(string, string) {}
278 +
279 +func (r *recordingFunctionRegistry) registeredNames() []string {
280 + r.mu.Lock()
281 + defer r.mu.Unlock()
282 + out := make([]string, len(r.registered))
283 + copy(out, r.registered)
284 + return out
285 +}
src/go/plugin/agent/jobmgr/modulefuncs.go
+31
@@ -290,3 +290,34 @@ func (r *moduleFuncRegistry) getJobMethod(moduleName, jobName, methodID string)
290 }
291 return nil, false
292 }
293 +
294 +// findMethodCollision checks whether module:method key would collide with already-registered methods.
295 +// It checks static module methods and job methods from other jobs within the same module.
296 +func (r *moduleFuncRegistry) findMethodCollision(moduleName, jobName, methodID string) (collision string, ok bool) {
297 + r.mu.RLock()
298 + defer r.mu.RUnlock()
299 +
300 + mf, ok := r.modules[moduleName]
301 + if !ok {
302 + return "", false
303 + }
304 +
305 + if mf.methodsByID != nil {
306 + if _, exists := mf.methodsByID[methodID]; exists {
307 + return "static method", true
308 + }
309 + }
310 +
311 + for ownerJob, methods := range mf.jobMethods {
312 + if ownerJob == jobName {
313 + continue
314 + }
315 + for _, method := range methods {
316 + if method.ID == methodID {
317 + return "job method on " + ownerJob, true
318 + }
319 + }
320 + }
321 +
322 + return "", false
323 +}
src/go/plugin/agent/jobmgr/sim_test.go
+90 -5
@@ -7,6 +7,7 @@ import (
7 "context"
8 "errors"
9 "strings"
10 + "sync"
11 "testing"
12 "time"
13
@@ -38,14 +39,56 @@ type runSim struct {
39 wantDyncfg string
40 }
41
42 +const funcResultEndMarker = "FUNCTION_RESULT_END\n\n"
43 +
44 +type simOutput struct {
45 + mu sync.Mutex
46 +
47 + buf bytes.Buffer
48 + funcResultCount int
49 + tail string
50 +}
51 +
52 +func (o *simOutput) Write(p []byte) (int, error) {
53 + o.mu.Lock()
54 + defer o.mu.Unlock()
55 +
56 + n, err := o.buf.Write(p)
57 + if n > 0 {
58 + data := o.tail + string(p[:n])
59 + o.funcResultCount += strings.Count(data, funcResultEndMarker)
60 +
61 + tailLen := len(funcResultEndMarker) - 1
62 + if len(data) > tailLen {
63 + o.tail = data[len(data)-tailLen:]
64 + } else {
65 + o.tail = data
66 + }
67 + }
68 +
69 + return n, err
70 +}
71 +
72 +func (o *simOutput) String() string {
73 + o.mu.Lock()
74 + defer o.mu.Unlock()
75 + return o.buf.String()
76 +}
77 +
78 +func (o *simOutput) FuncResultCount() int {
79 + o.mu.Lock()
80 + defer o.mu.Unlock()
81 + return o.funcResultCount
82 +}
83 +
84 func (s *runSim) run(t *testing.T) {
85 t.Helper()
86
87 require.NotNil(t, s.do, "s.do is nil")
88
46 - var buf bytes.Buffer
89 + var out simOutput
90 mgr := New(Config{PluginName: testPluginName})
48 - mgr.SetDyncfgResponder(dyncfg.NewResponder(netdataapi.New(safewriter.New(&buf))))
91 + mgr.SetDyncfgResponder(dyncfg.NewResponder(netdataapi.New(safewriter.New(&out))))
92 mgr.modules = prepareMockRegistry()
93
94 done := make(chan struct{})
@@ -63,6 +106,21 @@ func (s *runSim) run(t *testing.T) {
106 }
107
108 s.do(mgr, grpCh)
109 +
110 + expectedResults := strings.Count(s.wantDyncfg, "FUNCTION_RESULT_END")
111 + require.Eventually(t, func() bool {
112 + return countDiscovered(mgr) == len(s.wantDiscovered) &&
113 + mgr.seen.Count() == len(s.wantSeen) &&
114 + mgr.exposed.Count() == len(s.wantExposed) &&
115 + runningSetMatches(mgr.runningJobs.snapshot(), s.wantRunning) &&
116 + out.FuncResultCount() >= expectedResults
117 + }, timeout, 10*time.Millisecond, "manager state did not settle before shutdown")
118 +
119 + runningBeforeShutdown := make(map[string]struct{})
120 + for _, job := range mgr.runningJobs.snapshot() {
121 + runningBeforeShutdown[job.FullName()] = struct{}{}
122 + }
123 +
124 cancel()
125
126 select {
@@ -72,7 +130,7 @@ func (s *runSim) run(t *testing.T) {
130 }
131
132 var lines []string
75 - for _, s := range strings.Split(buf.String(), "\n") {
133 + for _, s := range strings.Split(out.String(), "\n") {
134 if strings.HasPrefix(s, "CONFIG") && strings.Contains(s, " template ") {
135 continue
136 }
@@ -123,10 +181,10 @@ func (s *runSim) run(t *testing.T) {
181 require.Truef(t, we.status == entry.Status, "exposed: wrong status for '%s', want %s got %s", we.cfg.UID(), we.status, entry.Status)
182 }
183
126 - wantLen, gotLen = len(s.wantRunning), len(mgr.runningJobs.items)
184 + wantLen, gotLen = len(s.wantRunning), len(runningBeforeShutdown)
185 require.Equalf(t, wantLen, gotLen, "runningJobs: different len (want %d got %d)", wantLen, gotLen)
186 for _, name := range s.wantRunning {
129 - _, ok := mgr.runningJobs.lookup(name)
187 + _, ok := runningBeforeShutdown[name]
188 require.Truef(t, ok, "runningJobs: job '%s' is not found", name)
189 }
190 }
@@ -203,3 +261,30 @@ func prepareMockRegistry() collectorapi.Registry {
261
262 return reg
263 }
264 +
265 +func countDiscovered(mgr *Manager) int {
266 + var n int
267 + for _, cfgs := range mgr.discoveredConfigs.items {
268 + n += len(cfgs)
269 + }
270 + return n
271 +}
272 +
273 +func runningSetMatches(jobs []runtimeJob, want []string) bool {
274 + if len(jobs) != len(want) {
275 + return false
276 + }
277 +
278 + wantSet := make(map[string]struct{}, len(want))
279 + for _, name := range want {
280 + wantSet[name] = struct{}{}
281 + }
282 +
283 + for _, job := range jobs {
284 + if _, ok := wantSet[job.FullName()]; !ok {
285 + return false
286 + }
287 + }
288 +
289 + return true
290 +}
src/go/plugin/agent/jobmgr/vnode_store.go new
+54
@@ -0,0 +1,54 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package jobmgr
4 +
5 +import (
6 + "fmt"
7 +
8 + "github.com/netdata/netdata/go/plugins/plugin/framework/vnodes"
9 +)
10 +
11 +// vnodeStore owns manager vnode state. It is intentionally lock-free because
12 +// mutations are serialized by Manager.run and startup initialization.
13 +type vnodeStore struct {
14 + items map[string]*vnodes.VirtualNode
15 +}
16 +
17 +func newVnodeStore(items map[string]*vnodes.VirtualNode) *vnodeStore {
18 + if items == nil {
19 + items = make(map[string]*vnodes.VirtualNode)
20 + }
21 + return &vnodeStore{items: items}
22 +}
23 +
24 +func (s *vnodeStore) Lookup(name string) (*vnodes.VirtualNode, bool) {
25 + cfg, ok := s.items[name]
26 + return cfg, ok
27 +}
28 +
29 +func (s *vnodeStore) Upsert(cfg *vnodes.VirtualNode) (changed bool, affectedJobNames []string, err error) {
30 + if cfg == nil {
31 + return false, nil, fmt.Errorf("nil vnode config")
32 + }
33 + if orig, ok := s.items[cfg.Name]; ok && orig.Equal(cfg) {
34 + return false, nil, nil
35 + }
36 + s.items[cfg.Name] = cfg
37 + return true, nil, nil
38 +}
39 +
40 +func (s *vnodeStore) Remove(name string) (removed bool, err error) {
41 + if _, ok := s.items[name]; !ok {
42 + return false, nil
43 + }
44 + delete(s.items, name)
45 + return true, nil
46 +}
47 +
48 +func (s *vnodeStore) ForEach(fn func(cfg *vnodes.VirtualNode) bool) {
49 + for _, cfg := range s.items {
50 + if !fn(cfg) {
51 + return
52 + }
53 + }
54 +}
src/go/plugin/framework/dyncfg/handler.go
+212 -35
@@ -3,8 +3,10 @@
3 package dyncfg
4
5 import (
6 + "context"
7 "errors"
8 "sync"
9 + "time"
10
11 "github.com/netdata/netdata/go/plugins/logger"
12 "github.com/netdata/netdata/go/plugins/pkg/netdataapi"
@@ -50,12 +52,13 @@ type CodedError interface {
52
53 // HandlerOpts configures the handler with component-specific settings.
54 type HandlerOpts[C Config] struct {
53 - Logger *logger.Logger
54 - API *Responder
55 - Seen *SeenCache[C]
56 - Exposed *ExposedCache[C]
57 - Callbacks Callbacks[C]
58 - WaitKey func(cfg C) string // optional key used to gate config processing until enable/disable
55 + Logger *logger.Logger
56 + API *Responder
57 + Seen *SeenCache[C]
58 + Exposed *ExposedCache[C]
59 + Callbacks Callbacks[C]
60 + WaitKey func(cfg C) string // optional key used to gate config processing until enable/disable
61 + WaitTimeout time.Duration // optional timeout for decision wait; zero keeps wait open until matching command
62
63 Path string // dyncfg path (e.g. "/collectors/go.d/Jobs")
64 EnableFailCode int // response code for enable failure (jobmgr: 200, SD: 422)
@@ -76,9 +79,187 @@ type Handler[C Config] struct {
79 enableFailCode int
80 removeStockOnEnableFail bool
81 jobCommands []Command
79 - waitKeyFn func(cfg C) string
80 - waitKey string
81 - waitMu sync.RWMutex
82 + waitGate *waitGate[C]
83 +}
84 +
85 +// WaitTimeoutEvent describes a wait gate timeout transition.
86 +type WaitTimeoutEvent struct {
87 + Key string
88 + Elapsed time.Duration
89 + Threshold time.Duration
90 +}
91 +
92 +// WaitDecisionStep is one serialized wait-loop transition.
93 +type WaitDecisionStep struct {
94 + Command Function
95 + HasCommand bool
96 + Timeout WaitTimeoutEvent
97 + TimedOut bool
98 +}
99 +
100 +// waitGate encapsulates wait-for-decision state and timing orchestration.
101 +type waitGate[C Config] struct {
102 + keyFn func(cfg C) string
103 + timeout time.Duration
104 + key string
105 + since time.Time
106 + deadline time.Time
107 + mu sync.RWMutex
108 + now func() time.Time
109 +}
110 +
111 +func newWaitGate[C Config](keyFn func(cfg C) string, timeout time.Duration) *waitGate[C] {
112 + return &waitGate[C]{
113 + keyFn: keyFn,
114 + timeout: timeout,
115 + now: time.Now,
116 + }
117 +}
118 +
119 +func (wg *waitGate[C]) waitForDecision(cfg C) {
120 + if wg.keyFn == nil {
121 + return
122 + }
123 + key := wg.keyFn(cfg)
124 + if key == "" {
125 + return
126 + }
127 +
128 + wg.mu.Lock()
129 + wg.key = key
130 + wg.since = time.Time{}
131 + wg.deadline = time.Time{}
132 + if wg.timeout > 0 {
133 + now := wg.nowTime()
134 + wg.since = now
135 + wg.deadline = now.Add(wg.timeout)
136 + }
137 + wg.mu.Unlock()
138 +}
139 +
140 +func (wg *waitGate[C]) waitingForDecision() bool {
141 + wg.mu.RLock()
142 + defer wg.mu.RUnlock()
143 + return wg.key != ""
144 +}
145 +
146 +func (wg *waitGate[C]) decisionRemaining() (time.Duration, bool) {
147 + wg.mu.RLock()
148 + defer wg.mu.RUnlock()
149 +
150 + if wg.timeout <= 0 || wg.key == "" || wg.deadline.IsZero() {
151 + return 0, false
152 + }
153 + now := wg.nowTime()
154 + if now.After(wg.deadline) || now.Equal(wg.deadline) {
155 + return 0, true
156 + }
157 + return wg.deadline.Sub(now), true
158 +}
159 +
160 +func (wg *waitGate[C]) nextStep(ctx context.Context, dyncfgCh <-chan Function) (WaitDecisionStep, bool) {
161 + var step WaitDecisionStep
162 +
163 + waitFor, hasTimeout := wg.decisionRemaining()
164 + if !hasTimeout {
165 + select {
166 + case <-ctx.Done():
167 + return step, false
168 + case fn := <-dyncfgCh:
169 + step.Command = fn
170 + step.HasCommand = true
171 + return step, true
172 + }
173 + }
174 +
175 + timer := time.NewTimer(waitFor)
176 + defer func() {
177 + if !timer.Stop() {
178 + select {
179 + case <-timer.C:
180 + default:
181 + }
182 + }
183 + }()
184 +
185 + select {
186 + case <-ctx.Done():
187 + return step, false
188 + case fn := <-dyncfgCh:
189 + step.Command = fn
190 + step.HasCommand = true
191 + return step, true
192 + case <-timer.C:
193 + step.Timeout, step.TimedOut = wg.expireDecision()
194 + return step, true
195 + }
196 +}
197 +
198 +func (wg *waitGate[C]) expireDecision() (WaitTimeoutEvent, bool) {
199 + var event WaitTimeoutEvent
200 +
201 + if wg.timeout <= 0 {
202 + return event, false
203 + }
204 + now := wg.nowTime()
205 +
206 + wg.mu.Lock()
207 + defer wg.mu.Unlock()
208 +
209 + if wg.key == "" || wg.deadline.IsZero() || now.Before(wg.deadline) {
210 + return event, false
211 + }
212 +
213 + event.Key = wg.key
214 + event.Threshold = wg.timeout
215 + if !wg.since.IsZero() && now.After(wg.since) {
216 + event.Elapsed = now.Sub(wg.since)
217 + } else {
218 + event.Elapsed = wg.timeout
219 + }
220 +
221 + wg.clearLocked()
222 + return event, true
223 +}
224 +
225 +func (wg *waitGate[C]) currentKey() string {
226 + wg.mu.RLock()
227 + defer wg.mu.RUnlock()
228 + return wg.key
229 +}
230 +
231 +func (wg *waitGate[C]) keyFor(cfg C) string {
232 + if wg.keyFn == nil {
233 + return ""
234 + }
235 + return wg.keyFn(cfg)
236 +}
237 +
238 +func (wg *waitGate[C]) clearIfMatch(key string) {
239 + wg.mu.Lock()
240 + defer wg.mu.Unlock()
241 + if wg.key == key {
242 + wg.clearLocked()
243 + }
244 +}
245 +
246 +func (wg *waitGate[C]) clearLocked() {
247 + wg.key = ""
248 + wg.since = time.Time{}
249 + wg.deadline = time.Time{}
250 +}
251 +
252 +func (wg *waitGate[C]) nowTime() time.Time {
253 + if wg.now != nil {
254 + return wg.now()
255 + }
256 + return time.Now()
257 +}
258 +
259 +func (wg *waitGate[C]) setNow(now func() time.Time) {
260 + wg.mu.Lock()
261 + wg.now = now
262 + wg.mu.Unlock()
263 }
264
265 func NewHandler[C Config](opts HandlerOpts[C]) *Handler[C] {
@@ -92,7 +273,7 @@ func NewHandler[C Config](opts HandlerOpts[C]) *Handler[C] {
273 enableFailCode: opts.EnableFailCode,
274 removeStockOnEnableFail: opts.RemoveStockOnEnableFail,
275 jobCommands: opts.JobCommands,
95 - waitKeyFn: opts.WaitKey,
276 + waitGate: newWaitGate(opts.WaitKey, opts.WaitTimeout),
277 }
278 }
279
@@ -138,40 +319,40 @@ func (h *Handler[C]) RemoveDiscoveredConfig(cfg C) (*Entry[C], bool) {
319 // WaitForDecision blocks non-dyncfg config processing until a matching
320 // enable/disable command is observed for the provided config.
321 func (h *Handler[C]) WaitForDecision(cfg C) {
141 - if h.waitKeyFn == nil {
142 - return
143 - }
144 - key := h.waitKeyFn(cfg)
145 - if key == "" {
146 - return
147 - }
148 - h.waitMu.Lock()
149 - h.waitKey = key
150 - h.waitMu.Unlock()
322 + h.waitGate.waitForDecision(cfg)
323 }
324
325 // WaitingForDecision reports whether config processing should currently wait
326 // for a matching enable/disable command.
327 func (h *Handler[C]) WaitingForDecision() bool {
156 - h.waitMu.RLock()
157 - defer h.waitMu.RUnlock()
158 - return h.waitKey != ""
328 + return h.waitGate.waitingForDecision()
329 +}
330 +
331 +// WaitDecisionRemaining returns time until wait gate timeout.
332 +func (h *Handler[C]) WaitDecisionRemaining() (time.Duration, bool) {
333 + return h.waitGate.decisionRemaining()
334 +}
335 +
336 +// NextWaitDecisionStep blocks until either a dyncfg command arrives, wait timeout fires, or context is canceled.
337 +// It centralizes wait-loop orchestration so caller logic stays minimal.
338 +func (h *Handler[C]) NextWaitDecisionStep(ctx context.Context, dyncfgCh <-chan Function) (WaitDecisionStep, bool) {
339 + return h.waitGate.nextStep(ctx, dyncfgCh)
340 +}
341 +
342 +// ExpireWaitDecision clears the current wait gate when it exceeds configured timeout.
343 +func (h *Handler[C]) ExpireWaitDecision() (WaitTimeoutEvent, bool) {
344 + return h.waitGate.expireDecision()
345 }
346
347 // SyncDecision updates wait-state based on the incoming command.
348 // Only a matching enable/disable command clears the current wait key.
349 func (h *Handler[C]) SyncDecision(fn Function) {
164 - if h.waitKeyFn == nil {
165 - return
166 - }
350 cmd := fn.Command()
351 if cmd != CommandEnable && cmd != CommandDisable {
352 return
353 }
354
172 - h.waitMu.RLock()
173 - waitKey := h.waitKey
174 - h.waitMu.RUnlock()
355 + waitKey := h.waitGate.currentKey()
356 if waitKey == "" {
357 return
358 }
@@ -184,15 +365,11 @@ func (h *Handler[C]) SyncDecision(fn Function) {
365 if !ok {
366 return
367 }
187 - if h.waitKeyFn(entry.Cfg) != waitKey {
368 + if h.waitGate.keyFor(entry.Cfg) != waitKey {
369 return
370 }
371
191 - h.waitMu.Lock()
192 - if h.waitKey == waitKey {
193 - h.waitKey = ""
194 - }
195 - h.waitMu.Unlock()
372 + h.waitGate.clearIfMatch(waitKey)
373 }
374
375 // NotifyJobCreate registers/updates a config in the dyncfg API (upsert).
src/go/plugin/framework/dyncfg/handler_test.go
+135
@@ -4,10 +4,12 @@ package dyncfg
4
5 import (
6 "bytes"
7 + "context"
8 "errors"
9 "fmt"
10 "strings"
11 "testing"
12 + "time"
13
14 "github.com/netdata/netdata/go/plugins/logger"
15 "github.com/netdata/netdata/go/plugins/pkg/netdataapi"
@@ -109,6 +111,10 @@ func (m *mockCallbacks) ConfigID(cfg testConfig) string {
111 }
112
113 func newTestHandler(cb *mockCallbacks) *Handler[testConfig] {
114 + return newTestHandlerWithWaitTimeout(cb, 5*time.Second)
115 +}
116 +
117 +func newTestHandlerWithWaitTimeout(cb *mockCallbacks, waitTimeout time.Duration) *Handler[testConfig] {
118 var buf bytes.Buffer
119 api := NewResponder(netdataapi.New(safewriter.New(&buf)))
120 return NewHandler(HandlerOpts[testConfig]{
@@ -120,6 +126,7 @@ func newTestHandler(cb *mockCallbacks) *Handler[testConfig] {
126 WaitKey: func(cfg testConfig) string {
127 return cfg.Source()
128 },
129 + WaitTimeout: waitTimeout,
130
131 Path: "/test/path",
132 EnableFailCode: 200,
@@ -203,6 +210,134 @@ func TestHandler_WaitForDecision_MismatchedCommandKeepsWait(t *testing.T) {
210 assert.False(t, h.WaitingForDecision())
211 }
212
213 +func TestHandler_WaitForDecision_TimeoutClearsWait(t *testing.T) {
214 + cb := &mockCallbacks{}
215 + h := newTestHandlerWithWaitTimeout(cb, 5*time.Second)
216 +
217 + cfg := testConfig{
218 + uid: "uid-job1",
219 + key: "job1",
220 + sourceType: "stock",
221 + source: "mod/job1",
222 + }
223 + h.exposed.Add(&Entry[testConfig]{Cfg: cfg, Status: StatusAccepted})
224 +
225 + base := time.Unix(1000, 0)
226 + h.waitGate.setNow(func() time.Time { return base })
227 +
228 + h.WaitForDecision(cfg)
229 + assert.True(t, h.WaitingForDecision())
230 +
231 + h.waitGate.setNow(func() time.Time { return base.Add(4 * time.Second) })
232 + remaining, ok := h.WaitDecisionRemaining()
233 + assert.True(t, ok)
234 + assert.Equal(t, time.Second, remaining)
235 +
236 + _, timedOut := h.ExpireWaitDecision()
237 + assert.False(t, timedOut)
238 + assert.True(t, h.WaitingForDecision())
239 +
240 + h.waitGate.setNow(func() time.Time { return base.Add(5 * time.Second) })
241 + event, timedOut := h.ExpireWaitDecision()
242 + assert.True(t, timedOut)
243 + assert.Equal(t, "mod/job1", event.Key)
244 + assert.Equal(t, 5*time.Second, event.Threshold)
245 + assert.Equal(t, 5*time.Second, event.Elapsed)
246 + assert.False(t, h.WaitingForDecision())
247 +
248 + _, ok = h.WaitDecisionRemaining()
249 + assert.False(t, ok)
250 +}
251 +
252 +func TestHandler_WaitForDecision_TimeoutDisabledKeepsWait(t *testing.T) {
253 + cb := &mockCallbacks{}
254 + h := newTestHandlerWithWaitTimeout(cb, 0)
255 +
256 + cfg := testConfig{
257 + uid: "uid-job1",
258 + key: "job1",
259 + sourceType: "stock",
260 + source: "mod/job1",
261 + }
262 + h.exposed.Add(&Entry[testConfig]{Cfg: cfg, Status: StatusAccepted})
263 +
264 + base := time.Unix(1000, 0)
265 + h.waitGate.setNow(func() time.Time { return base })
266 + h.WaitForDecision(cfg)
267 +
268 + h.waitGate.setNow(func() time.Time { return base.Add(24 * time.Hour) })
269 + _, timedOut := h.ExpireWaitDecision()
270 + assert.False(t, timedOut)
271 + assert.True(t, h.WaitingForDecision())
272 +}
273 +
274 +func TestHandler_NextWaitDecisionStep_Command(t *testing.T) {
275 + cb := &mockCallbacks{}
276 + h := newTestHandlerWithWaitTimeout(cb, 5*time.Second)
277 +
278 + cfg := testConfig{
279 + uid: "uid-job1",
280 + key: "job1",
281 + sourceType: "stock",
282 + source: "mod/job1",
283 + }
284 + h.exposed.Add(&Entry[testConfig]{Cfg: cfg, Status: StatusAccepted})
285 + h.WaitForDecision(cfg)
286 +
287 + ch := make(chan Function, 1)
288 + fn := newTestFn("test:job1", "enable", "", nil)
289 + ch <- fn
290 +
291 + step, ok := h.NextWaitDecisionStep(context.Background(), ch)
292 + require.True(t, ok)
293 + require.True(t, step.HasCommand)
294 + assert.Equal(t, fn.UID(), step.Command.UID())
295 + assert.False(t, step.TimedOut)
296 +}
297 +
298 +func TestHandler_NextWaitDecisionStep_Timeout(t *testing.T) {
299 + cb := &mockCallbacks{}
300 + h := newTestHandlerWithWaitTimeout(cb, 20*time.Millisecond)
301 +
302 + cfg := testConfig{
303 + uid: "uid-job1",
304 + key: "job1",
305 + sourceType: "stock",
306 + source: "mod/job1",
307 + }
308 + h.exposed.Add(&Entry[testConfig]{Cfg: cfg, Status: StatusAccepted})
309 + h.WaitForDecision(cfg)
310 +
311 + ch := make(chan Function)
312 + step, ok := h.NextWaitDecisionStep(context.Background(), ch)
313 + require.True(t, ok)
314 + require.True(t, step.TimedOut)
315 + assert.Equal(t, "mod/job1", step.Timeout.Key)
316 + assert.False(t, h.WaitingForDecision())
317 +}
318 +
319 +func TestHandler_NextWaitDecisionStep_ContextCancel(t *testing.T) {
320 + cb := &mockCallbacks{}
321 + h := newTestHandlerWithWaitTimeout(cb, 5*time.Second)
322 +
323 + cfg := testConfig{
324 + uid: "uid-job1",
325 + key: "job1",
326 + sourceType: "stock",
327 + source: "mod/job1",
328 + }
329 + h.exposed.Add(&Entry[testConfig]{Cfg: cfg, Status: StatusAccepted})
330 + h.WaitForDecision(cfg)
331 +
332 + ctx, cancel := context.WithCancel(context.Background())
333 + cancel()
334 +
335 + ch := make(chan Function)
336 + _, ok := h.NextWaitDecisionStep(ctx, ch)
337 + assert.False(t, ok)
338 + assert.True(t, h.WaitingForDecision())
339 +}
340 +
341 func TestHandler_AddDiscoveredConfig_TracksSeenAndExposed(t *testing.T) {
342 cb := &mockCallbacks{}
343 h := newTestHandler(cb)
src/go/plugin/framework/functions/manager.go
+19 -6
@@ -107,6 +107,19 @@ func (m *Manager) run(ctx context.Context, quitCh chan struct{}) {
107 func (m *Manager) lookupFunction(name string) (func(Function), bool) {
108 m.mux.Lock()
109 fs, ok := m.FunctionRegistry[name]
110 + var (
111 + direct func(Function)
112 + prefixes map[string]func(Function)
113 + )
114 + if ok && fs != nil {
115 + direct = fs.direct
116 + if len(fs.prefixes) > 0 {
117 + prefixes = make(map[string]func(Function), len(fs.prefixes))
118 + for prefix, handler := range fs.prefixes {
119 + prefixes[prefix] = handler
120 + }
121 + }
122 + }
123 m.mux.Unlock()
124
125 if !ok || fs == nil {
@@ -114,13 +127,13 @@ func (m *Manager) lookupFunction(name string) (func(Function), bool) {
127 }
128
129 return func(f Function) {
117 - if len(fs.prefixes) > 0 {
118 - m.handlePrefixRouting(f, fs)
130 + if len(prefixes) > 0 {
131 + m.handlePrefixRouting(f, prefixes)
132 return
133 }
134
122 - if fs.direct != nil {
123 - fs.direct(f)
135 + if direct != nil {
136 + direct(f)
137 return
138 }
139
@@ -128,14 +141,14 @@ func (m *Manager) lookupFunction(name string) (func(Function), bool) {
141 }, true
142 }
143
131 -func (m *Manager) handlePrefixRouting(f Function, fs *functionSet) {
144 +func (m *Manager) handlePrefixRouting(f Function, prefixes map[string]func(Function)) {
145 if len(f.Args) == 0 {
146 m.respf(&f, 503, "unknown function '%s' (%v)", f.Name, f.Args)
147 return
148 }
149
150 id := f.Args[0]
138 - for prefix, handler := range fs.prefixes {
151 + for prefix, handler := range prefixes {
152 if strings.HasPrefix(id, prefix) {
153 handler(f)
154 return
src/go/plugin/framework/functions/manager_snapshot_test.go new
+47
@@ -0,0 +1,47 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package functions
4 +
5 +import (
6 + "testing"
7 +
8 + "github.com/stretchr/testify/require"
9 +)
10 +
11 +func TestLookupFunction_UsesDirectSnapshot(t *testing.T) {
12 + mgr := NewManager()
13 +
14 + called := make(chan struct{}, 1)
15 + mgr.Register("fn", func(Function) { called <- struct{}{} })
16 +
17 + handler, ok := mgr.lookupFunction("fn")
18 + require.True(t, ok)
19 +
20 + mgr.Unregister("fn")
21 + handler(Function{Name: "fn"})
22 +
23 + select {
24 + case <-called:
25 + default:
26 + t.Fatal("snapshot handler should still invoke the originally resolved direct function")
27 + }
28 +}
29 +
30 +func TestLookupFunction_UsesPrefixSnapshot(t *testing.T) {
31 + mgr := NewManager()
32 +
33 + called := make(chan struct{}, 1)
34 + mgr.RegisterPrefix("config", "collector:", func(Function) { called <- struct{}{} })
35 +
36 + handler, ok := mgr.lookupFunction("config")
37 + require.True(t, ok)
38 +
39 + mgr.UnregisterPrefix("config", "collector:")
40 + handler(Function{Name: "config", Args: []string{"collector:job"}})
41 +
42 + select {
43 + case <-called:
44 + default:
45 + t.Fatal("snapshot handler should still route using the prefix set captured at lookup time")
46 + }
47 +}