chore(go.d): add prefix-based func registration (#21245)
Ilya Mashchenko committed
Oct 30, 2025 at 18:37 UTC
f5734ae13baec0970ad8f22b0c9e6b18fb13db6a
6 files changed
+276
-23
src/go/plugin/go.d/agent/functions/ext.go
+60
-5
@@ -11,20 +11,75 @@ func (m *Manager) Register(name string, fn func(Function)) {
11
m.mux.Lock()
12
defer m.mux.Unlock()
13
14
- if _, ok := m.FunctionRegistry[name]; !ok {
15
- m.Debugf("registering function '%s'", name)
14
+ fs, ok := m.FunctionRegistry[name]
15
+ if !ok {
16
+ m.Debugf("registering function '%s' (direct)", name)
17
+ fs = &functionSet{prefixes: make(map[string]func(Function))}
18
+ m.FunctionRegistry[name] = fs
19
} else {
17
- m.Warningf("re-registering function '%s'", name)
20
+ if fs.direct != nil {
21
+ m.Warningf("re-registering direct function '%s'", name)
22
+ } else {
23
+ m.Debugf("registering function '%s' (direct)", name)
24
+ }
25
}
19
- m.FunctionRegistry[name] = fn
26
+
27
+ fs.direct = fn
28
}
29
30
func (m *Manager) Unregister(name string) {
31
m.mux.Lock()
32
defer m.mux.Unlock()
33
26
- if _, ok := m.FunctionRegistry[name]; !ok {
34
+ if _, ok := m.FunctionRegistry[name]; ok {
35
delete(m.FunctionRegistry, name)
36
m.Debugf("unregistering function '%s'", name)
37
}
38
}
39
+
40
+func (m *Manager) RegisterPrefix(name, prefix string, fn func(Function)) {
41
+ if fn == nil {
42
+ m.Warningf("not registering '%s' with prefix '%s': nil function", name, prefix)
43
+ return
44
+ }
45
+ if prefix == "" {
46
+ m.Warningf("not registering '%s': empty prefix", name)
47
+ return
48
+ }
49
+
50
+ m.mux.Lock()
51
+ defer m.mux.Unlock()
52
+
53
+ fs := m.FunctionRegistry[name]
54
+ if fs == nil {
55
+ fs = &functionSet{prefixes: make(map[string]func(Function))}
56
+ m.FunctionRegistry[name] = fs
57
+ }
58
+
59
+ if _, exists := fs.prefixes[prefix]; exists {
60
+ m.Warningf("re-registering function '%s' with prefix '%s'", name, prefix)
61
+ } else {
62
+ m.Debugf("registering function '%s' with prefix '%s'", name, prefix)
63
+ }
64
+
65
+ fs.prefixes[prefix] = fn
66
+}
67
+
68
+func (m *Manager) UnregisterPrefix(name, prefix string) {
69
+ m.mux.Lock()
70
+ defer m.mux.Unlock()
71
+
72
+ fs, ok := m.FunctionRegistry[name]
73
+ if !ok || fs.prefixes == nil {
74
+ return
75
+ }
76
+
77
+ if _, exists := fs.prefixes[prefix]; exists {
78
+ m.Debugf("unregistering function '%s' with prefix '%s'", name, prefix)
79
+ delete(fs.prefixes, prefix)
80
+ }
81
+
82
+ if fs.direct == nil && len(fs.prefixes) == 0 {
83
+ delete(m.FunctionRegistry, name)
84
+ }
85
+}
src/go/plugin/go.d/agent/functions/manager.go
+47
-8
@@ -8,6 +8,7 @@ import (
8
"fmt"
9
"log/slog"
10
"strconv"
11
+ "strings"
12
"sync"
13
"time"
14
@@ -16,6 +17,11 @@ import (
17
"github.com/netdata/netdata/go/plugins/pkg/safewriter"
18
)
19
20
+type functionSet struct {
21
+ direct func(Function) // for globally-unique names
22
+ prefixes map[string]func(Function) // for prefix-multiplexed names
23
+}
24
+
25
func NewManager() *Manager {
26
return &Manager{
27
Logger: logger.New().With(
@@ -24,7 +30,7 @@ func NewManager() *Manager {
30
api: netdataapi.New(safewriter.Stdout),
31
input: stdinInput,
32
mux: &sync.Mutex{},
27
- FunctionRegistry: make(map[string]func(Function)),
33
+ FunctionRegistry: make(map[string]*functionSet),
34
}
35
}
36
@@ -36,7 +42,7 @@ type Manager struct {
42
input input
43
44
mux *sync.Mutex
39
- FunctionRegistry map[string]func(Function)
45
+ FunctionRegistry map[string]*functionSet
46
}
47
48
func (m *Manager) Run(ctx context.Context, quitCh chan struct{}) {
@@ -81,29 +87,62 @@ func (m *Manager) run(ctx context.Context, quitCh chan struct{}) {
87
continue
88
}
89
84
- function, ok := m.lookupFunction(fn.Name)
90
+ handler, ok := m.lookupFunction(fn.Name)
91
if !ok {
92
m.Infof("skipping execution of '%s': unregistered function", fn.Name)
93
m.respf(fn, 501, "unregistered function: %s", fn.Name)
94
continue
95
}
90
- if function == nil {
96
+ if handler == nil {
97
m.Warningf("skipping execution of '%s': nil function registered", fn.Name)
98
m.respf(fn, 501, "nil function: %s", fn.Name)
99
continue
100
}
101
96
- function(*fn)
102
+ handler(*fn)
103
}
104
}
105
}
106
107
func (m *Manager) lookupFunction(name string) (func(Function), bool) {
108
m.mux.Lock()
103
- defer m.mux.Unlock()
109
+ fs, ok := m.FunctionRegistry[name]
110
+ m.mux.Unlock()
111
+
112
+ if !ok || fs == nil {
113
+ return nil, false
114
+ }
115
+
116
+ return func(f Function) {
117
+ if len(fs.prefixes) > 0 {
118
+ m.handlePrefixRouting(f, fs)
119
+ return
120
+ }
121
+
122
+ if fs.direct != nil {
123
+ fs.direct(f)
124
+ return
125
+ }
126
+
127
+ m.respf(&f, 503, "unknown function '%s' (%v)", f.Name, f.Args)
128
+ }, true
129
+}
130
+
131
+func (m *Manager) handlePrefixRouting(f Function, fs *functionSet) {
132
+ if len(f.Args) == 0 {
133
+ m.respf(&f, 503, "unknown function '%s' (%v)", f.Name, f.Args)
134
+ return
135
+ }
136
+
137
+ id := f.Args[0]
138
+ for prefix, handler := range fs.prefixes {
139
+ if strings.HasPrefix(id, prefix) {
140
+ handler(f)
141
+ return
142
+ }
143
+ }
144
105
- f, ok := m.FunctionRegistry[name]
106
- return f, ok
145
+ m.respf(&f, 503, "unknown function '%s' (%v)", f.Name, f.Args)
146
}
147
148
func (m *Manager) respf(fn *Function, code int, msgf string, a ...any) {
src/go/plugin/go.d/agent/functions/manager_test.go
+157
@@ -5,6 +5,7 @@ package functions
5
import (
6
"bufio"
7
"context"
8
+ "fmt"
9
"sort"
10
"strings"
11
"testing"
@@ -77,6 +78,162 @@ func TestManager_Register(t *testing.T) {
78
}
79
}
80
81
+func TestManager_RegisterPrefix(t *testing.T) {
82
+ type inputFn struct {
83
+ name string
84
+ prefix string
85
+ invalid bool
86
+ }
87
+
88
+ tests := map[string]struct {
89
+ input []inputFn
90
+ expected []string // flattened as "name:prefix"
91
+ }{
92
+ "valid registration (two prefixes under same name)": {
93
+ input: []inputFn{
94
+ {name: "config", prefix: "collector:"},
95
+ {name: "config", prefix: "vnode:"},
96
+ },
97
+ expected: []string{"config:collector:", "config:vnode:"},
98
+ },
99
+ "registration with duplicates (same name+prefix)": {
100
+ input: []inputFn{
101
+ {name: "config", prefix: "collector:"},
102
+ {name: "config", prefix: "collector:"}, // duplicate should overwrite, not duplicate
103
+ {name: "config", prefix: "vnode:"},
104
+ },
105
+ expected: []string{"config:collector:", "config:vnode:"},
106
+ },
107
+ "registration across multiple names": {
108
+ input: []inputFn{
109
+ {name: "config", prefix: "collector:"},
110
+ {name: "status", prefix: "node:"},
111
+ },
112
+ expected: []string{"config:collector:", "status:node:"},
113
+ },
114
+ "registration with nil functions is ignored": {
115
+ input: []inputFn{
116
+ {name: "config", prefix: "collector:"},
117
+ {name: "config", prefix: "vnode:", invalid: true}, // nil fn -> ignored
118
+ },
119
+ expected: []string{"config:collector:"},
120
+ },
121
+ }
122
+
123
+ for name, test := range tests {
124
+ t.Run(name, func(t *testing.T) {
125
+ mgr := NewManager()
126
+
127
+ for _, v := range test.input {
128
+ if v.invalid {
129
+ mgr.RegisterPrefix(v.name, v.prefix, nil)
130
+ } else {
131
+ mgr.RegisterPrefix(v.name, v.prefix, func(Function) {})
132
+ }
133
+ }
134
+
135
+ var got []string
136
+ for fname, fs := range mgr.FunctionRegistry {
137
+ if fs == nil || len(fs.prefixes) == 0 {
138
+ continue
139
+ }
140
+ for p := range fs.prefixes {
141
+ got = append(got, fmt.Sprintf("%s:%s", fname, p))
142
+ }
143
+ }
144
+ sort.Strings(got)
145
+ sort.Strings(test.expected)
146
+
147
+ assert.Equal(t, test.expected, got)
148
+ })
149
+ }
150
+}
151
+
152
+func TestManager_UnregisterPrefix(t *testing.T) {
153
+ type regFn struct {
154
+ name string
155
+ prefix string
156
+ }
157
+ type unregFn struct {
158
+ name string
159
+ prefix string
160
+ }
161
+
162
+ tests := map[string]struct {
163
+ register []regFn
164
+ unreg []unregFn
165
+ expected []string // flattened as "name:prefix"
166
+ }{
167
+ "remove one of multiple prefixes keeps the other": {
168
+ register: []regFn{
169
+ {name: "config", prefix: "collector:"},
170
+ {name: "config", prefix: "vnode:"},
171
+ },
172
+ unreg: []unregFn{
173
+ {name: "config", prefix: "collector:"},
174
+ },
175
+ expected: []string{"config:vnode:"},
176
+ },
177
+ "remove last prefix deletes the name entry": {
178
+ register: []regFn{
179
+ {name: "config", prefix: "collector:"},
180
+ },
181
+ unreg: []unregFn{
182
+ {name: "config", prefix: "collector:"},
183
+ },
184
+ expected: nil,
185
+ },
186
+ "unregister non-existing prefix is a no-op": {
187
+ register: []regFn{
188
+ {name: "config", prefix: "collector:"},
189
+ },
190
+ unreg: []unregFn{
191
+ {name: "config", prefix: "vnode:"}, // doesn't exist
192
+ },
193
+ expected: []string{"config:collector:"},
194
+ },
195
+ "unregister on unknown name is a no-op": {
196
+ register: []regFn{
197
+ {name: "config", prefix: "collector:"},
198
+ },
199
+ unreg: []unregFn{
200
+ {name: "status", prefix: "node:"}, // name not present
201
+ },
202
+ expected: []string{"config:collector:"},
203
+ },
204
+ }
205
+
206
+ for name, test := range tests {
207
+ t.Run(name, func(t *testing.T) {
208
+ mgr := NewManager()
209
+
210
+ // initial registrations
211
+ for _, r := range test.register {
212
+ mgr.RegisterPrefix(r.name, r.prefix, func(Function) {})
213
+ }
214
+
215
+ // perform unregistrations
216
+ for _, u := range test.unreg {
217
+ mgr.UnregisterPrefix(u.name, u.prefix)
218
+ }
219
+
220
+ var got []string
221
+ for fname, fs := range mgr.FunctionRegistry {
222
+ if fs == nil || len(fs.prefixes) == 0 {
223
+ continue
224
+ }
225
+ for p := range fs.prefixes {
226
+ got = append(got, fmt.Sprintf("%s:%s", fname, p))
227
+ }
228
+ }
229
+ sort.Strings(got)
230
+ sort.Strings(test.expected)
231
+
232
+ assert.Equal(t, test.expected, got)
233
+ })
234
+ }
235
+}
236
+
237
func TestManager_Run(t *testing.T) {
238
tests := map[string]struct {
239
register []string
src/go/plugin/go.d/agent/jobmgr/di.go
+2
-2
@@ -13,8 +13,8 @@ type Vnodes interface {
13
}
14
15
type FunctionRegistry interface {
16
- Register(name string, reg func(functions.Function))
17
- Unregister(name string)
16
+ RegisterPrefix(name, prefix string, fn func(functions.Function))
17
+ UnregisterPrefix(name string, prefix string)
18
}
19
20
type dyncfgAPI interface {
src/go/plugin/go.d/agent/jobmgr/manager.go
+4
-2
@@ -95,7 +95,8 @@ func (m *Manager) Run(ctx context.Context, in chan []*confgroup.Group) {
95
defer func() { m.cleanup(); m.Info("instance is stopped") }()
96
m.ctx = ctx
97
98
- m.FnReg.Register("config", m.dyncfgConfig)
98
+ m.FnReg.RegisterPrefix("config", m.dyncfgCollectorPrefixValue(), m.dyncfgConfig)
99
+ m.FnReg.RegisterPrefix("config", m.dyncfgVnodePrefixValue(), m.dyncfgConfig)
100
101
m.dyncfgVnodeModuleCreate()
102
@@ -284,7 +285,8 @@ func (m *Manager) stopRunningJob(name string) {
285
}
286
287
func (m *Manager) cleanup() {
287
- m.FnReg.Unregister("config")
288
+ m.FnReg.UnregisterPrefix("config", m.dyncfgCollectorPrefixValue())
289
+ m.FnReg.UnregisterPrefix("config", m.dyncfgVnodePrefixValue())
290
291
m.runningJobs.lock()
292
defer m.runningJobs.unlock()
src/go/plugin/go.d/agent/jobmgr/noop.go
+6
-6
@@ -10,9 +10,9 @@ import (
10
11
type noop struct{}
12
13
-func (n noop) Lock(string) (bool, error) { return true, nil }
14
-func (n noop) Unlock(string) {}
15
-func (n noop) UnlockAll() {}
16
-func (n noop) Lookup(string) (*vnodes.VirtualNode, bool) { return nil, false }
17
-func (n noop) Register(name string, reg func(functions.Function)) {}
18
-func (n noop) Unregister(name string) {}
13
+func (n noop) Lock(string) (bool, error) { return true, nil }
14
+func (n noop) Unlock(string) {}
15
+func (n noop) UnlockAll() {}
16
+func (n noop) Lookup(string) (*vnodes.VirtualNode, bool) { return nil, false }
17
+func (n noop) RegisterPrefix(name, prefix string, reg func(functions.Function)) {}
18
+func (n noop) UnregisterPrefix(name, prefix string) {}