fix(jobmgr): register collector static functions on first job start (#21894)
Ilya Mashchenko committed
Mar 5, 2026 at 16:46 UTC
60b130ca4c7eb1776f439e642394d5c2b99dfbd9
3 files changed
+138
-35
src/go/plugin/agent/jobmgr/manager.go
+53
-35
@@ -94,6 +94,7 @@ func New(cfg Config) *Manager {
94
runtimeService: cfg.RuntimeService,
95
96
moduleFuncs: newModuleFuncRegistry(),
97
+ staticMethodsSeen: make(map[string]struct{}),
98
discoveredConfigs: newDiscoveredConfigsCache(),
99
seen: seen,
100
exposed: exposed,
@@ -166,6 +167,9 @@ type Manager struct {
167
168
fileStatus *fileStatus
169
moduleFuncs *moduleFuncRegistry
170
+ // staticMethodsSeen tracks modules whose static methods were already exposed.
171
+ // Registration is delayed until the first started job for a module.
172
+ staticMethodsSeen map[string]struct{}
173
174
discoveredConfigs *discoveredConfigs
175
seen *dyncfg.SeenCache[confgroup.Config]
@@ -217,41 +221,7 @@ func (m *Manager) Run(ctx context.Context, in chan []*confgroup.Group) {
221
m.moduleFuncs.registerModule(name, creator)
222
}
223
220
- // Register static module-level functions
221
- if creator.Methods != nil {
222
- methods := creator.Methods()
223
- for _, method := range methods {
224
- if method.ID == "" {
225
- m.Warningf("skipping function registration for module '%s': empty method ID", name)
226
- continue
227
- }
228
- funcName := fmt.Sprintf("%s:%s", name, method.ID)
229
- m.fnReg.Register(funcName, m.makeMethodFuncHandler(name, method.ID))
230
-
231
- // Notify Netdata about this function so it appears in the functions API
232
- help := method.Help
233
- if help == "" {
234
- help = fmt.Sprintf("%s %s data function", name, method.ID)
235
- }
236
-
237
- // https://github.com/netdata/netdata/blob/1bc1775a17590b3c0fe3a4fe547dc6146d07be89/src/libnetdata/user-auth/http-access.h#L21
238
- const cloudAccess = "0x0013" // SIGNED_ID | SAME_SPACE | SENSITIVE_DATA
239
- access := "0x0000"
240
- if method.RequireCloud {
241
- access = cloudAccess
242
- }
243
- m.dyncfgApi.FunctionGlobal(netdataapi.FunctionGlobalOpts{
244
- Name: funcName,
245
- Timeout: 60,
246
- Help: help,
247
- Tags: "top",
248
- Access: access,
249
- Priority: 100,
250
- Version: 3,
251
- })
252
- }
253
- }
254
- // Note: Per-job methods (JobMethods) are registered in startRunningJob
224
+ // Note: Static module methods and per-job methods are registered in startRunningJob.
225
}
226
227
m.loadFileStatus()
@@ -435,6 +405,7 @@ func (m *Manager) startRunningJob(job runtimeJob) {
405
406
// Track job for module function routing.
407
m.moduleFuncs.addJob(job.ModuleName(), job.Name(), job)
408
+ m.registerModuleMethodsOnFirstJobStart(job.ModuleName())
409
410
// Register job-specific methods if module provides JobMethods callback
411
creator, ok := m.modules.Lookup(job.ModuleName())
@@ -460,6 +431,9 @@ func (m *Manager) stopRunningJob(name string) {
431
m.unregisterJobMethods(job)
432
// Remove job from module function registry.
433
m.moduleFuncs.removeJob(job.ModuleName(), job.Name())
434
+ // Static module methods remain registered for now. Once Netdata supports
435
+ // function removal, this should unregister static methods when the last
436
+ // running job for the module is removed.
437
job.Stop()
438
}
439
}
@@ -502,6 +476,50 @@ func (m *Manager) waitCmdTestWorkers() {
476
}
477
}
478
479
+func (m *Manager) registerModuleMethodsOnFirstJobStart(moduleName string) {
480
+ if _, ok := m.staticMethodsSeen[moduleName]; ok {
481
+ return
482
+ }
483
+
484
+ creator, ok := m.modules.Lookup(moduleName)
485
+ if !ok || creator.Methods == nil {
486
+ return
487
+ }
488
+
489
+ methods := creator.Methods()
490
+ for _, method := range methods {
491
+ if method.ID == "" {
492
+ m.Warningf("skipping function registration for module '%s': empty method ID", moduleName)
493
+ continue
494
+ }
495
+ funcName := fmt.Sprintf("%s:%s", moduleName, method.ID)
496
+ m.fnReg.Register(funcName, m.makeMethodFuncHandler(moduleName, method.ID))
497
+
498
+ help := method.Help
499
+ if help == "" {
500
+ help = fmt.Sprintf("%s %s data function", moduleName, method.ID)
501
+ }
502
+
503
+ // https://github.com/netdata/netdata/blob/1bc1775a17590b3c0fe3a4fe547dc6146d07be89/src/libnetdata/user-auth/http-access.h#L21
504
+ const cloudAccess = "0x0013" // SIGNED_ID | SAME_SPACE | SENSITIVE_DATA
505
+ access := "0x0000"
506
+ if method.RequireCloud {
507
+ access = cloudAccess
508
+ }
509
+ m.dyncfgApi.FunctionGlobal(netdataapi.FunctionGlobalOpts{
510
+ Name: funcName,
511
+ Timeout: 60,
512
+ Help: help,
513
+ Tags: "top",
514
+ Access: access,
515
+ Priority: 100,
516
+ Version: 3,
517
+ })
518
+ }
519
+
520
+ m.staticMethodsSeen[moduleName] = struct{}{}
521
+}
522
+
523
// registerJobMethods registers methods for a specific job with Netdata
524
func (m *Manager) registerJobMethods(job collectorapi.RuntimeJob, methods []funcapi.MethodConfig) {
525
planned := make(map[string]struct{}, len(methods))
src/go/plugin/agent/jobmgr/manager_process_test.go
+77
@@ -170,6 +170,83 @@ func TestRunNotifyRunningJobs_TickOutsideLock(t *testing.T) {
170
}
171
}
172
173
+func TestRun_DoesNotRegisterModuleMethodsBeforeAnyJobStarts(t *testing.T) {
174
+ fnReg := &recordingFunctionRegistry{}
175
+ mgr := New(Config{PluginName: testPluginName, FnReg: fnReg})
176
+
177
+ mgr.modules = collectorapi.Registry{
178
+ "mod": collectorapi.Creator{
179
+ Methods: func() []funcapi.MethodConfig {
180
+ return []funcapi.MethodConfig{{ID: "a"}}
181
+ },
182
+ },
183
+ }
184
+
185
+ ctx, cancel := context.WithCancel(context.Background())
186
+ defer cancel()
187
+
188
+ in := make(chan []*confgroup.Group)
189
+ done := make(chan struct{})
190
+ go func() {
191
+ mgr.Run(ctx, in)
192
+ close(done)
193
+ }()
194
+
195
+ waitCtx, waitCancel := context.WithTimeout(context.Background(), time.Second)
196
+ defer waitCancel()
197
+ require.True(t, mgr.WaitStarted(waitCtx), "manager did not report started")
198
+
199
+ cancel()
200
+ close(in)
201
+
202
+ select {
203
+ case <-done:
204
+ case <-time.After(2 * time.Second):
205
+ t.Fatal("manager did not stop after cancel")
206
+ }
207
+
208
+ assert.Empty(t, fnReg.registeredNames(), "static methods must not be registered before first started job")
209
+}
210
+
211
+func TestStartRunningJob_RegistersModuleMethodsOnFirstStartedJob(t *testing.T) {
212
+ fnReg := &recordingFunctionRegistry{}
213
+ mgr := New(Config{PluginName: testPluginName, FnReg: fnReg})
214
+ creator := collectorapi.Creator{
215
+ Methods: func() []funcapi.MethodConfig {
216
+ return []funcapi.MethodConfig{{ID: "a"}, {ID: "b"}}
217
+ },
218
+ }
219
+ mgr.modules = collectorapi.Registry{"mod": creator}
220
+ mgr.moduleFuncs.registerModule("mod", creator)
221
+
222
+ job := &lockProbeJob{fullName: "mod_job1", moduleName: "mod", name: "job1"}
223
+ mgr.startRunningJob(job)
224
+
225
+ assert.ElementsMatch(t, []string{"mod:a", "mod:b"}, fnReg.registeredNames())
226
+}
227
+
228
+func TestStartRunningJob_DoesNotReregisterModuleMethods(t *testing.T) {
229
+ fnReg := &recordingFunctionRegistry{}
230
+ mgr := New(Config{PluginName: testPluginName, FnReg: fnReg})
231
+ creator := collectorapi.Creator{
232
+ Methods: func() []funcapi.MethodConfig {
233
+ return []funcapi.MethodConfig{{ID: "a"}, {ID: "b"}}
234
+ },
235
+ }
236
+ mgr.modules = collectorapi.Registry{"mod": creator}
237
+ mgr.moduleFuncs.registerModule("mod", creator)
238
+
239
+ job1 := &lockProbeJob{fullName: "mod_job1", moduleName: "mod", name: "job1"}
240
+ mgr.startRunningJob(job1)
241
+
242
+ job2 := &lockProbeJob{fullName: "mod_job2", moduleName: "mod", name: "job2"}
243
+ mgr.startRunningJob(job2)
244
+
245
+ registered := fnReg.registeredNames()
246
+ assert.Len(t, registered, 2)
247
+ assert.ElementsMatch(t, []string{"mod:a", "mod:b"}, registered)
248
+}
249
+
250
func TestRegisterJobMethods_FailFastOnCollisionWithStaticMethod(t *testing.T) {
251
fnReg := &recordingFunctionRegistry{}
252
mgr := New(Config{PluginName: testPluginName, FnReg: fnReg})
src/go/plugin/agent/jobmgr/sim_test.go
+8
@@ -130,13 +130,21 @@ func (s *runSim) run(t *testing.T) {
130
}
131
132
var lines []string
133
+ skipNextEmpty := false
134
for _, s := range strings.Split(out.String(), "\n") {
135
if strings.HasPrefix(s, "CONFIG") && strings.Contains(s, " template ") {
136
+ skipNextEmpty = false
137
continue
138
}
139
if strings.HasPrefix(s, "FUNCTION GLOBAL") {
140
+ skipNextEmpty = true
141
continue
142
}
143
+ if skipNextEmpty && s == "" {
144
+ skipNextEmpty = false
145
+ continue
146
+ }
147
+ skipNextEmpty = false
148
if strings.HasPrefix(s, "FUNCTION_RESULT_BEGIN") {
149
parts := strings.Fields(s)
150
s = strings.Join(parts[:len(parts)-1], " ") // remove timestamp