@cryptotaxi247 / netdata-1 / commits / 1fd075efd

feat(go.d.plugin): add dyncfg vnodes (#19205)

* add dyncfg vnodes * fix tests

Ilya Mashchenko committed Dec 14, 2024 at 16:33 UTC 1fd075efd93fcd8e4d2d9f830cf54b8e82eeae9a
17 files changed +1323 -962
src/go/cmd/godplugin/config.go
-14
@@ -65,7 +65,6 @@ type config struct {
65 collectorsDir multipath.MultiPath
66 collectorsWatchPath []string
67 serviceDiscoveryDir multipath.MultiPath
68 - vnodesDir multipath.MultiPath
68 stateFile string
69 lockDir string
70 }
@@ -79,7 +78,6 @@ func newConfig(opts *cli.Option, env *envConfig) *config {
78 cfg.collectorsDir = cfg.initCollectorsDir(opts)
79 cfg.collectorsWatchPath = cfg.initCollectorsWatchPaths(opts, env)
80 cfg.serviceDiscoveryDir = cfg.initServiceDiscoveryConfigDir()
82 - cfg.vnodesDir = cfg.initVnodesDir()
81 cfg.stateFile = cfg.initStateFile(env)
82 cfg.lockDir = env.lockDir
83
@@ -156,18 +154,6 @@ func (c *config) initServiceDiscoveryConfigDir() multipath.MultiPath {
154 return mpath
155 }
156
159 -func (c *config) initVnodesDir() multipath.MultiPath {
160 - c.mustPluginDir()
161 -
162 - var mpath multipath.MultiPath
163 -
164 - for _, v := range c.pluginDir {
165 - mpath = append(mpath, filepath.Join(v, "vnodes"))
166 - }
167 -
168 - return mpath
169 -}
170 -
157 func (c *config) initCollectorsWatchPaths(opts *cli.Option, env *envConfig) []string {
158 if env.watchPath == "" {
159 return opts.WatchPath
src/go/cmd/godplugin/main.go
-1
@@ -55,7 +55,6 @@ func main() {
55 CollectorsConfigDir: cfg.collectorsDir,
56 ServiceDiscoveryConfigDir: cfg.serviceDiscoveryDir,
57 CollectorsConfigWatchPath: cfg.collectorsWatchPath,
58 - VnodesConfigDir: cfg.vnodesDir,
58 StateFile: cfg.stateFile,
59 LockDir: cfg.lockDir,
60 RunModule: opts.Module,
src/go/plugin/go.d/agent/agent.go
+1 -4
@@ -36,7 +36,6 @@ type Config struct {
36 CollectorsConfigDir []string
37 CollectorsConfigWatchPath []string
38 ServiceDiscoveryConfigDir []string
39 - VnodesConfigDir []string
39 StateFile string
40 LockDir string
41 ModuleRegistry module.Registry
@@ -54,7 +53,6 @@ type Agent struct {
53 CollectorsConfDir multipath.MultiPath
54 CollectorsConfigWatchPath []string
55 ServiceDiscoveryConfigDir multipath.MultiPath
57 - VnodesConfigDir multipath.MultiPath
56
57 StateFile string
58 LockDir string
@@ -81,7 +79,6 @@ func New(cfg Config) *Agent {
79 CollectorsConfDir: cfg.CollectorsConfigDir,
80 ServiceDiscoveryConfigDir: cfg.ServiceDiscoveryConfigDir,
81 CollectorsConfigWatchPath: cfg.CollectorsConfigWatchPath,
84 - VnodesConfigDir: cfg.VnodesConfigDir,
82 StateFile: cfg.StateFile,
83 LockDir: cfg.LockDir,
84 RunModule: cfg.RunModule,
@@ -201,7 +198,7 @@ func (a *Agent) run(ctx context.Context) {
198 jobMgr.ConfigDefaults = discCfg.Registry
199 jobMgr.FnReg = fnMgr
200
204 - if reg := a.setupVnodeRegistry(); reg != nil && reg.Len() > 0 {
201 + if reg := a.setupVnodeRegistry(); len(reg) > 0 {
202 jobMgr.Vnodes = reg
203 }
204
src/go/plugin/go.d/agent/functions/input.go
+8 -5
@@ -12,24 +12,27 @@ type input interface {
12 }
13
14 var stdinInput = func() input {
15 - r := &stdinReader{chLines: make(chan string)}
15 + r := &stdinReader{
16 + linesCh: make(chan string),
17 + }
18 +
19 go r.run()
20 +
21 return r
22 }()
23
24 type stdinReader struct {
21 - chLines chan string
25 + linesCh chan string
26 }
27
28 func (in *stdinReader) run() {
29 sc := bufio.NewScanner(bufio.NewReader(os.Stdin))
30
31 for sc.Scan() {
28 - text := sc.Text()
29 - in.chLines <- text
32 + in.linesCh <- sc.Text()
33 }
34 }
35
36 func (in *stdinReader) lines() chan string {
34 - return in.chLines
37 + return in.linesCh
38 }
src/go/plugin/go.d/agent/functions/manager_test.go
+10 -10
@@ -90,7 +90,7 @@ FUNCTION UID 1 "fn1 arg1 arg2" 0xFFFF "method=api,role=test"
90 `,
91 expected: []Function{
92 {
93 - key: "FUNCTION",
93 + key: lineFunction,
94 UID: "UID",
95 Timeout: time.Second,
96 Name: "fn1",
@@ -110,7 +110,7 @@ FUNCTION UID 1 "fn2 arg1 arg2" 0xFFFF "method=api,role=test"
110 `,
111 expected: []Function{
112 {
113 - key: "FUNCTION",
113 + key: lineFunction,
114 UID: "UID",
115 Timeout: time.Second,
116 Name: "fn1",
@@ -121,7 +121,7 @@ FUNCTION UID 1 "fn2 arg1 arg2" 0xFFFF "method=api,role=test"
121 Payload: nil,
122 },
123 {
124 - key: "FUNCTION",
124 + key: lineFunction,
125 UID: "UID",
126 Timeout: time.Second,
127 Name: "fn2",
@@ -143,7 +143,7 @@ FUNCTION_PAYLOAD_END
143 `,
144 expected: []Function{
145 {
146 - key: "FUNCTION_PAYLOAD",
146 + key: lineFunctionPayload,
147 UID: "UID",
148 Timeout: time.Second,
149 Name: "fn1",
@@ -170,7 +170,7 @@ FUNCTION_PAYLOAD_END
170 `,
171 expected: []Function{
172 {
173 - key: "FUNCTION_PAYLOAD",
173 + key: lineFunctionPayload,
174 UID: "UID",
175 Timeout: time.Second,
176 Name: "fn1",
@@ -181,7 +181,7 @@ FUNCTION_PAYLOAD_END
181 Payload: []byte("payload line1\npayload line2"),
182 },
183 {
184 - key: "FUNCTION_PAYLOAD",
184 + key: lineFunctionPayload,
185 UID: "UID",
186 Timeout: time.Second,
187 Name: "fn2",
@@ -211,7 +211,7 @@ FUNCTION_PAYLOAD_END
211 `,
212 expected: []Function{
213 {
214 - key: "FUNCTION_PAYLOAD",
214 + key: lineFunctionPayload,
215 UID: "UID",
216 Timeout: time.Second,
217 Name: "fn1",
@@ -222,7 +222,7 @@ FUNCTION_PAYLOAD_END
222 Payload: []byte("payload line1\npayload line2"),
223 },
224 {
225 - key: "FUNCTION",
225 + key: lineFunction,
226 UID: "UID",
227 Timeout: time.Second,
228 Name: "fn2",
@@ -233,7 +233,7 @@ FUNCTION_PAYLOAD_END
233 Payload: nil,
234 },
235 {
236 - key: "FUNCTION",
236 + key: lineFunction,
237 UID: "UID",
238 Timeout: time.Second,
239 Name: "fn3",
@@ -244,7 +244,7 @@ FUNCTION_PAYLOAD_END
244 Payload: nil,
245 },
246 {
247 - key: "FUNCTION_PAYLOAD",
247 + key: lineFunctionPayload,
248 UID: "UID",
249 Timeout: time.Second,
250 Name: "fn4",
src/go/plugin/go.d/agent/functions/parser.go
+10 -4
@@ -12,6 +12,12 @@ import (
12 "time"
13 )
14
15 +const (
16 + lineFunction = "FUNCTION"
17 + lineFunctionPayload = "FUNCTION_PAYLOAD"
18 + lineFunctionPayloadEnd = "FUNCTION_PAYLOAD_END"
19 +)
20 +
21 type Function struct {
22 key string
23 UID string
@@ -49,9 +55,9 @@ func (p *inputParser) parse(line string) (*Function, error) {
55 }
56
57 switch {
52 - case strings.HasPrefix(line, "FUNCTION "):
58 + case strings.HasPrefix(line, lineFunction+" "):
59 return p.parseFunction(line)
54 - case strings.HasPrefix(line, "FUNCTION_PAYLOAD "):
60 + case strings.HasPrefix(line, lineFunctionPayload+" "):
61 fn, err := p.parseFunction(line)
62 if err != nil {
63 return nil, err
@@ -66,7 +72,7 @@ func (p *inputParser) parse(line string) (*Function, error) {
72 }
73
74 func (p *inputParser) handlePayloadLine(line string) (*Function, error) {
69 - if line == "FUNCTION_PAYLOAD_END" {
75 + if line == lineFunctionPayloadEnd {
76 p.readingPayload = false
77 p.currentFn.Payload = []byte(p.payloadBuf.String())
78 fn := p.currentFn
@@ -74,7 +80,7 @@ func (p *inputParser) handlePayloadLine(line string) (*Function, error) {
80 return fn, nil
81 }
82
77 - if strings.HasPrefix(line, "FUNCTION") {
83 + if strings.HasPrefix(line, lineFunction) {
84 p.readingPayload = false
85 p.currentFn = nil
86 p.payloadBuf.Reset()
src/go/plugin/go.d/agent/jobmgr/dyncfg.go
+13 -806
@@ -3,23 +3,14 @@
3 package jobmgr
4
5 import (
6 - "context"
6 "encoding/json"
8 - "errors"
7 "fmt"
10 - "log/slog"
8 "reflect"
12 - "strconv"
9 "strings"
14 - "time"
15 - "unicode"
16 -
17 - "github.com/netdata/netdata/go/plugins/logger"
18 - "github.com/netdata/netdata/go/plugins/pkg/netdataapi"
19 - "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/confgroup"
20 - "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/functions"
10
11 "gopkg.in/yaml.v2"
12 +
13 + "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/functions"
14 )
15
16 type dyncfgStatus int
@@ -50,61 +41,6 @@ func (s dyncfgStatus) String() string {
41 }
42 }
43
53 -const (
54 - dyncfgIDPrefix = "go.d:collector:"
55 - dyncfgPath = "/collectors/jobs"
56 -)
57 -
58 -func dyncfgModID(name string) string {
59 - return fmt.Sprintf("%s%s", dyncfgIDPrefix, name)
60 -}
61 -func dyncfgJobID(cfg confgroup.Config) string {
62 - return fmt.Sprintf("%s%s:%s", dyncfgIDPrefix, cfg.Module(), cfg.Name())
63 -}
64 -
65 -func dyncfgModCmds() string {
66 - return "add schema enable disable test userconfig"
67 -}
68 -func dyncfgJobCmds(cfg confgroup.Config) string {
69 - cmds := "schema get enable disable update restart test userconfig"
70 - if isDyncfg(cfg) {
71 - cmds += " remove"
72 - }
73 - return cmds
74 -}
75 -
76 -func (m *Manager) dyncfgModuleCreate(name string) {
77 - m.api.CONFIGCREATE(netdataapi.ConfigOpts{
78 - ID: dyncfgModID(name),
79 - Status: dyncfgAccepted.String(),
80 - ConfigType: "template",
81 - Path: dyncfgPath,
82 - SourceType: "internal",
83 - Source: "internal",
84 - SupportedCommands: dyncfgModCmds(),
85 - })
86 -}
87 -
88 -func (m *Manager) dyncfgJobCreate(cfg confgroup.Config, status dyncfgStatus) {
89 - m.api.CONFIGCREATE(netdataapi.ConfigOpts{
90 - ID: dyncfgJobID(cfg),
91 - Status: status.String(),
92 - ConfigType: "job",
93 - Path: dyncfgPath,
94 - SourceType: cfg.SourceType(),
95 - Source: cfg.Source(),
96 - SupportedCommands: dyncfgJobCmds(cfg),
97 - })
98 -}
99 -
100 -func (m *Manager) dyncfgJobRemove(cfg confgroup.Config) {
101 - m.api.CONFIGDELETE(dyncfgJobID(cfg))
102 -}
103 -
104 -func (m *Manager) dyncfgJobStatus(cfg confgroup.Config, status dyncfgStatus) {
105 - m.api.CONFIGSTATUS(dyncfgJobID(cfg), status.String())
106 -}
107 -
44 func (m *Manager) dyncfgConfig(fn functions.Function) {
45 if len(fn.Args) < 2 {
46 m.Warningf("dyncfg: %s: missing required arguments, want 3 got %d", fn.Name, len(fn.Args))
@@ -120,751 +56,22 @@ func (m *Manager) dyncfgConfig(fn functions.Function) {
56
57 //m.Infof("QQ FN: '%s'", fn)
58
123 - action := strings.ToLower(fn.Args[1])
124 -
125 - switch action {
126 - case "userconfig":
127 - m.dyncfgConfigUserconfig(fn)
128 - return
129 - case "test":
130 - m.dyncfgConfigTest(fn)
131 - return
132 - case "schema":
133 - m.dyncfgConfigSchema(fn)
134 - return
135 - }
136 -
137 - select {
138 - case <-m.ctx.Done():
139 - m.dyncfgRespf(fn, 503, "Job manager is shutting down.")
140 - case m.dyncfgCh <- fn:
141 - }
142 -}
143 -
144 -func (m *Manager) dyncfgConfigExec(fn functions.Function) {
145 - action := strings.ToLower(fn.Args[1])
146 -
147 - switch action {
148 - case "test":
149 - m.dyncfgConfigTest(fn)
150 - case "schema":
151 - m.dyncfgConfigSchema(fn)
152 - case "get":
153 - m.dyncfgConfigGet(fn)
154 - case "restart":
155 - m.dyncfgConfigRestart(fn)
156 - case "enable":
157 - m.dyncfgConfigEnable(fn)
158 - case "disable":
159 - m.dyncfgConfigDisable(fn)
160 - case "add":
161 - m.dyncfgConfigAdd(fn)
162 - case "remove":
163 - m.dyncfgConfigRemove(fn)
164 - case "update":
165 - m.dyncfgConfigUpdate(fn)
166 - default:
167 - m.Warningf("dyncfg: function '%s' not implemented", fn.String())
168 - m.dyncfgRespf(fn, 501, "Function '%s' is not implemented.", fn.Name)
169 - }
170 -}
171 -
172 -func (m *Manager) dyncfgConfigUserconfig(fn functions.Function) {
173 - id := fn.Args[0]
174 - jn := "test"
175 - if len(fn.Args) > 2 {
176 - jn = fn.Args[2]
177 - }
178 -
179 - mn, ok := extractModuleName(id)
180 - if !ok {
181 - m.Warningf("dyncfg: userconfig: could not extract module and job from id (%s)", id)
182 - m.dyncfgRespf(fn, 400,
183 - "Invalid ID format. Could not extract module and job name from ID. Provided ID: %s.", id)
184 - return
185 - }
186 -
187 - creator, ok := m.Modules.Lookup(mn)
188 - if !ok {
189 - m.Warningf("dyncfg: userconfig: module %s not found", mn)
190 - m.dyncfgRespf(fn, 404, "The specified module '%s' is not registered.", mn)
191 - return
192 - }
193 -
194 - if creator.Config == nil || creator.Config() == nil {
195 - m.Warningf("dyncfg: userconfig: module %s: configuration not found", mn)
196 - m.dyncfgRespf(fn, 500, "Module %s does not provide configuration.", mn)
197 - return
198 - }
199 -
200 - bs, err := userConfigFromPayload(creator.Config(), jn, fn)
201 - if err != nil {
202 - m.Warningf("dyncfg: userconfig: module %s: failed to create config from payload: %v", mn, err)
203 - m.dyncfgRespf(fn, 400, "Invalid configuration format. Failed to create configuration from payload: %v.", err)
204 - }
205 -
206 - m.dyncfgRespPayloadYAML(fn, string(bs))
207 -}
208 -
209 -func (m *Manager) dyncfgConfigTest(fn functions.Function) {
210 - id := fn.Args[0]
211 - mn, ok := extractModuleName(id)
212 - if !ok {
213 - m.Warningf("dyncfg: test: could not extract module and job from id (%s)", id)
214 - m.dyncfgRespf(fn, 400,
215 - "Invalid ID format. Could not extract module and job name from ID. Provided ID: %s.", id)
216 - return
217 - }
218 -
219 - jn := "test"
220 - if len(fn.Args) > 2 {
221 - jn = fn.Args[2]
222 - }
223 -
224 - if err := validateJobName(jn); err != nil {
225 - m.Warningf("dyncfg: test: module %s: unacceptable job name '%s': %v", mn, jn, err)
226 - m.dyncfgRespf(fn, 400, "Unacceptable job name '%s': %v.", jn, err)
227 - return
228 - }
229 -
230 - creator, ok := m.Modules.Lookup(mn)
231 - if !ok {
232 - m.Warningf("dyncfg: test: module %s not found", mn)
233 - m.dyncfgRespf(fn, 404, "The specified module '%s' is not registered.", mn)
234 - return
235 - }
236 -
237 - cfg, err := configFromPayload(fn)
238 - if err != nil {
239 - m.Warningf("dyncfg: test: module %s: failed to create config from payload: %v", mn, err)
240 - m.dyncfgRespf(fn, 400, "Invalid configuration format. Failed to create configuration from payload: %v.", err)
241 - return
242 - }
243 -
244 - if cfg.Vnode() != "" {
245 - if _, ok := m.Vnodes.Lookup(cfg.Vnode()); !ok {
246 - m.Warningf("dyncfg: test: module %s: vnode %s not found", mn, cfg.Vnode())
247 - m.dyncfgRespf(fn, 400, "The specified vnode '%s' is not registered.", cfg.Vnode())
248 - return
249 - }
250 - }
251 -
252 - cfg.SetModule(mn)
253 - cfg.SetName(jn)
254 -
255 - job := creator.Create()
256 -
257 - if err := applyConfig(cfg, job); err != nil {
258 - m.Warningf("dyncfg: test: module %s: failed to apply config: %v", mn, err)
259 - m.dyncfgRespf(fn, 400, "Invalid configuration. Failed to apply configuration: %v.", err)
260 - return
261 - }
262 -
263 - job.GetBase().Logger = logger.New().With(
264 - slog.String("collector", cfg.Module()),
265 - slog.String("job", cfg.Name()),
266 - )
267 -
268 - defer job.Cleanup(context.Background())
269 -
270 - if err := job.Init(context.Background()); err != nil {
271 - m.dyncfgRespf(fn, 422, "Job initialization failed: %v", err)
272 - return
273 - }
274 - if err := job.Check(context.Background()); err != nil {
275 - m.dyncfgRespf(fn, 422, "Job check failed: %v", err)
276 - return
277 - }
278 -
279 - m.dyncfgRespf(fn, 200, "")
280 -}
281 -
282 -func (m *Manager) dyncfgConfigSchema(fn functions.Function) {
283 - id := fn.Args[0]
284 - mn, ok := extractModuleName(id)
285 - if !ok {
286 - m.Warningf("dyncfg: schema: could not extract module from id (%s)", id)
287 - m.dyncfgRespf(fn, 400, "Invalid ID format. Could not extract module name from ID. Provided ID: %s.", id)
288 - return
289 - }
290 -
291 - mod, ok := m.Modules.Lookup(mn)
292 - if !ok {
293 - m.Warningf("dyncfg: schema: module %s not found", mn)
294 - m.dyncfgRespf(fn, 404, "The specified module '%s' is not registered.", mn)
295 - return
296 - }
297 -
298 - if mod.JobConfigSchema == "" {
299 - m.Warningf("dyncfg: schema: module %s: schema not found", mn)
300 - m.dyncfgRespf(fn, 500, "Module %s configuration schema not found.", mn)
301 - return
302 - }
303 -
304 - m.dyncfgRespPayloadJSON(fn, mod.JobConfigSchema)
305 -}
306 -
307 -func (m *Manager) dyncfgConfigGet(fn functions.Function) {
308 - id := fn.Args[0]
309 - mn, jn, ok := extractModuleJobName(id)
310 - if !ok {
311 - m.Warningf("dyncfg: get: could not extract module and job from id (%s)", id)
312 - m.dyncfgRespf(fn, 400,
313 - "Invalid ID format. Could not extract module and job name from ID. Provided ID: %s.", id)
314 - return
315 - }
316 -
317 - creator, ok := m.Modules.Lookup(mn)
318 - if !ok {
319 - m.Warningf("dyncfg: get: module %s not found", mn)
320 - m.dyncfgRespf(fn, 404, "The specified module '%s' is not registered.", mn)
321 - return
322 - }
323 -
324 - ecfg, ok := m.exposedConfigs.lookupByName(mn, jn)
325 - if !ok {
326 - m.Warningf("dyncfg: get: module %s job %s not found", mn, jn)
327 - m.dyncfgRespf(fn, 404, "The specified module '%s' job '%s' is not registered.", mn, jn)
328 - return
329 - }
330 -
331 - mod := creator.Create()
332 -
333 - if err := applyConfig(ecfg.cfg, mod); err != nil {
334 - m.Warningf("dyncfg: get: module %s job %s failed to apply config: %v", mn, jn, err)
335 - m.dyncfgRespf(fn, 400, "Invalid configuration. Failed to apply configuration: %v.", err)
336 - return
337 - }
338 -
339 - conf := mod.Configuration()
340 - if conf == nil {
341 - m.Warningf("dyncfg: get: module %s: configuration not found", mn)
342 - m.dyncfgRespf(fn, 500, "Module %s does not provide configuration.", mn)
343 - return
344 - }
345 -
346 - bs, err := json.Marshal(conf)
347 - if err != nil {
348 - m.Warningf("dyncfg: get: module %s job %s failed to json marshal config: %v", mn, jn, err)
349 - m.dyncfgRespf(fn, 500, "Failed to convert configuration into JSON: %v.", err)
350 - return
351 - }
352 -
353 - m.dyncfgRespPayloadJSON(fn, string(bs))
354 -}
355 -
356 -func (m *Manager) dyncfgConfigRestart(fn functions.Function) {
357 - id := fn.Args[0]
358 - mn, jn, ok := extractModuleJobName(id)
359 - if !ok {
360 - m.Warningf("dyncfg: restart: could not extract module from id (%s)", id)
361 - m.dyncfgRespf(fn, 400, "Invalid ID format. Could not extract module name from ID. Provided ID: %s.", id)
362 - return
363 - }
364 -
365 - ecfg, ok := m.exposedConfigs.lookupByName(mn, jn)
366 - if !ok {
367 - m.Warningf("dyncfg: restart: module %s job %s not found", mn, jn)
368 - m.dyncfgRespf(fn, 404, "The specified module '%s' job '%s' is not registered.", mn, jn)
369 - return
370 - }
371 -
372 - job, err := m.createCollectorJob(ecfg.cfg)
373 - if err != nil {
374 - m.Warningf("dyncfg: restart: module %s job %s: failed to apply config: %v", mn, jn, err)
375 - m.dyncfgRespf(fn, 400, "Invalid configuration. Failed to apply configuration: %v.", err)
376 - m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
377 - return
378 - }
379 -
380 - switch ecfg.status {
381 - case dyncfgAccepted, dyncfgDisabled:
382 - m.Warningf("dyncfg: restart: module %s job %s: restarting not allowed in '%s' state", mn, jn, ecfg.status)
383 - m.dyncfgRespf(fn, 405, "Restarting data collection job is not allowed in '%s' state.", ecfg.status)
384 - m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
385 - return
386 - case dyncfgRunning:
387 - m.FileStatus.Remove(ecfg.cfg)
388 - m.FileLock.Unlock(ecfg.cfg.FullName())
389 - m.stopRunningJob(ecfg.cfg.FullName())
390 - default:
391 - }
392 -
393 - if err := job.AutoDetection(); err != nil {
394 - job.Cleanup()
395 - ecfg.status = dyncfgFailed
396 - m.dyncfgRespf(fn, 422, "Job restart failed: %v", err)
397 - m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
398 - return
399 - }
400 -
401 - if ok, err := m.FileLock.Lock(ecfg.cfg.FullName()); !ok && err == nil {
402 - job.Cleanup()
403 - ecfg.status = dyncfgFailed
404 - m.dyncfgRespf(fn, 500, "Job restart failed: cannot filelock.")
405 - m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
406 - return
407 - }
408 -
409 - ecfg.status = dyncfgRunning
410 -
411 - if isDyncfg(ecfg.cfg) {
412 - m.FileStatus.Save(ecfg.cfg, ecfg.status.String())
413 - }
414 - m.startRunningJob(job)
415 - m.dyncfgRespf(fn, 200, "")
416 - m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
417 -}
418 -
419 -func (m *Manager) dyncfgConfigEnable(fn functions.Function) {
420 - id := fn.Args[0]
421 - mn, jn, ok := extractModuleJobName(id)
422 - if !ok {
423 - m.Warningf("dyncfg: enable: could not extract module and job from id (%s)", id)
424 - m.dyncfgRespf(fn, 400, "Invalid ID format. Could not extract module and job name from ID. Provided ID: %s.", id)
425 - return
426 - }
427 -
428 - ecfg, ok := m.exposedConfigs.lookupByName(mn, jn)
429 - if !ok {
430 - m.Warningf("dyncfg: enable: module %s job %s not found", mn, jn)
431 - m.dyncfgRespf(fn, 404, "The specified module '%s' job '%s' is not registered.", mn, jn)
432 - return
433 - }
434 -
435 - if ecfg.cfg.FullName() == m.waitCfgOnOff {
436 - m.waitCfgOnOff = ""
437 - }
438 -
439 - switch ecfg.status {
440 - case dyncfgAccepted, dyncfgDisabled, dyncfgFailed:
441 - case dyncfgRunning:
442 - // non-dyncfg update triggers enable/disable
443 - m.dyncfgRespf(fn, 200, "")
444 - m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
445 - return
446 - default:
447 - m.Warningf("dyncfg: enable: module %s job %s: enabling not allowed in %s state", mn, jn, ecfg.status)
448 - m.dyncfgRespf(fn, 405, "Enabling data collection job is not allowed in '%s' state.", ecfg.status)
449 - m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
450 - return
451 - }
452 -
453 - job, err := m.createCollectorJob(ecfg.cfg)
454 - if err != nil {
455 - ecfg.status = dyncfgFailed
456 - m.Warningf("dyncfg: enable: module %s job %s: failed to apply config: %v", mn, jn, err)
457 - m.dyncfgRespf(fn, 400, "Invalid configuration. Failed to apply configuration: %v.", err)
458 - m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
459 - return
460 - }
461 -
462 - if err := job.AutoDetection(); err != nil {
463 - job.Cleanup()
464 - ecfg.status = dyncfgFailed
465 - m.dyncfgRespf(fn, 200, "Job enable failed: %v.", err)
466 -
467 - if isStock(ecfg.cfg) {
468 - m.exposedConfigs.remove(ecfg.cfg)
469 - m.dyncfgJobRemove(ecfg.cfg)
470 - } else {
471 - m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
472 - }
473 -
474 - if job.RetryAutoDetection() && !isDyncfg(ecfg.cfg) {
475 - m.Infof("%s[%s] job detection failed, will retry in %d seconds",
476 - ecfg.cfg.Module(), ecfg.cfg.Name(), job.AutoDetectionEvery())
477 -
478 - ctx, cancel := context.WithCancel(m.ctx)
479 - m.retryingTasks.add(ecfg.cfg, &retryTask{cancel: cancel})
480 - go runRetryTask(ctx, m.addCh, ecfg.cfg)
481 - }
482 - return
483 - }
484 -
485 - if ok, err := m.FileLock.Lock(ecfg.cfg.FullName()); !ok && err == nil {
486 - job.Cleanup()
487 - ecfg.status = dyncfgFailed
488 - m.dyncfgRespf(fn, 500, "Job enable failed: can not filelock.")
489 - m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
490 - return
491 - }
492 -
493 - ecfg.status = dyncfgRunning
494 -
495 - if isDyncfg(ecfg.cfg) {
496 - m.FileStatus.Save(ecfg.cfg, ecfg.status.String())
497 - }
498 -
499 - m.startRunningJob(job)
500 - m.dyncfgRespf(fn, 200, "")
501 - m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
502 -
503 -}
504 -
505 -func (m *Manager) dyncfgConfigDisable(fn functions.Function) {
506 - id := fn.Args[0]
507 - mn, jn, ok := extractModuleJobName(id)
508 - if !ok {
509 - m.Warningf("dyncfg: disable: could not extract module from id (%s)", id)
510 - m.dyncfgRespf(fn, 400, "Invalid ID format. Could not extract module name from ID. Provided ID: %s.", id)
511 - return
512 - }
513 -
514 - ecfg, ok := m.exposedConfigs.lookupByName(mn, jn)
515 - if !ok {
516 - m.Warningf("dyncfg: disable: module %s job %s not found", mn, jn)
517 - m.dyncfgRespf(fn, 404, "The specified module '%s' job '%s' is not registered.", mn, jn)
518 - return
519 - }
520 -
521 - if ecfg.cfg.FullName() == m.waitCfgOnOff {
522 - m.waitCfgOnOff = ""
523 - }
524 -
525 - switch ecfg.status {
526 - case dyncfgDisabled:
527 - m.dyncfgRespf(fn, 200, "")
528 - m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
529 - return
530 - case dyncfgRunning:
531 - m.stopRunningJob(ecfg.cfg.FullName())
532 - if isDyncfg(ecfg.cfg) {
533 - m.FileStatus.Remove(ecfg.cfg)
534 - }
535 - m.FileLock.Unlock(ecfg.cfg.FullName())
59 + switch id := fn.Args[0]; true {
60 + case strings.HasPrefix(id, dyncfgCollectorIDPrefix):
61 + m.dyncfgCollectorExec(fn)
62 + case strings.HasPrefix(id, dyncfgVnodeIDPrefix):
63 + m.dyncfgVnodeExec(fn)
64 default:
65 + m.dyncfgRespf(fn, 503, "unknown function '%s' (%s).", fn.Name, id)
66 }
538 -
539 - ecfg.status = dyncfgDisabled
540 - m.dyncfgRespf(fn, 200, "")
541 - m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
542 -}
543 -
544 -func (m *Manager) dyncfgConfigAdd(fn functions.Function) {
545 - if len(fn.Args) < 3 {
546 - m.Warningf("dyncfg: add: missing required arguments, want 3 got %d", len(fn.Args))
547 - m.dyncfgRespf(fn, 400, "Missing required arguments. Need at least 3, but got %d.", len(fn.Args))
548 - return
549 - }
550 -
551 - id := fn.Args[0]
552 - jn := fn.Args[2]
553 - mn, ok := extractModuleName(id)
554 - if !ok {
555 - m.Warningf("dyncfg: add: could not extract module from id (%s)", id)
556 - m.dyncfgRespf(fn, 400, "Invalid ID format. Could not extract module name from ID. Provided ID: %s.", id)
557 - return
558 - }
559 -
560 - if len(fn.Payload) == 0 {
561 - m.Warningf("dyncfg: add: module %s job %s missing configuration payload.", mn, jn)
562 - m.dyncfgRespf(fn, 400, "Missing configuration payload.")
563 - return
564 - }
565 -
566 - if err := validateJobName(jn); err != nil {
567 - m.Warningf("dyncfg: add: module %s: unacceptable job name '%s': %v", mn, jn, err)
568 - m.dyncfgRespf(fn, 400, "Unacceptable job name '%s': %v.", jn, err)
569 - return
570 - }
571 -
572 - cfg, err := configFromPayload(fn)
573 - if err != nil {
574 - m.Warningf("dyncfg: add: module %s job %s: failed to create config from payload: %v", mn, jn, err)
575 - m.dyncfgRespf(fn, 400, "Invalid configuration format. Failed to create configuration from payload: %v.", err)
576 - return
577 - }
578 -
579 - m.dyncfgSetConfigMeta(cfg, mn, jn)
580 -
581 - if _, err := m.createCollectorJob(cfg); err != nil {
582 - m.Warningf("dyncfg: add: module %s job %s: failed to apply config: %v", mn, jn, err)
583 - m.dyncfgRespf(fn, 400, "Invalid configuration. Failed to apply configuration: %v.", err)
584 - return
585 - }
586 -
587 - if ecfg, ok := m.exposedConfigs.lookup(cfg); ok {
588 - if scfg, ok := m.seenConfigs.lookup(ecfg.cfg); ok && isDyncfg(scfg.cfg) {
589 - m.seenConfigs.remove(ecfg.cfg)
590 - }
591 - m.exposedConfigs.remove(ecfg.cfg)
592 - m.stopRunningJob(ecfg.cfg.FullName())
593 - }
594 -
595 - scfg := &seenConfig{cfg: cfg, status: dyncfgAccepted}
596 - ecfg := scfg
597 - m.seenConfigs.add(scfg)
598 - m.exposedConfigs.add(ecfg)
599 -
600 - m.dyncfgRespf(fn, 202, "")
601 - m.dyncfgJobCreate(ecfg.cfg, ecfg.status)
602 -}
603 -
604 -func (m *Manager) dyncfgConfigRemove(fn functions.Function) {
605 - id := fn.Args[0]
606 - mn, jn, ok := extractModuleJobName(id)
607 - if !ok {
608 - m.Warningf("dyncfg: remove: could not extract module and job from id (%s)", id)
609 - m.dyncfgRespf(fn, 400, "Invalid ID format. Could not extract module and job name from ID. Provided ID: %s.", id)
610 - return
611 - }
612 -
613 - ecfg, ok := m.exposedConfigs.lookupByName(mn, jn)
614 - if !ok {
615 - m.Warningf("dyncfg: remove: module %s job %s not found", mn, jn)
616 - m.dyncfgRespf(fn, 404, "The specified module '%s' job '%s' is not registered.", mn, jn)
617 - return
618 - }
619 -
620 - if !isDyncfg(ecfg.cfg) {
621 - m.Warningf("dyncfg: remove: module %s job %s: can not remove jobs of type %s", mn, jn, ecfg.cfg.SourceType())
622 - m.dyncfgRespf(fn, 405, "Removing jobs of type '%s' is not supported. Only 'dyncfg' jobs can be removed.", ecfg.cfg.SourceType())
623 - return
624 - }
625 -
626 - m.seenConfigs.remove(ecfg.cfg)
627 - m.exposedConfigs.remove(ecfg.cfg)
628 - m.stopRunningJob(ecfg.cfg.FullName())
629 - m.FileLock.Unlock(ecfg.cfg.FullName())
630 - m.FileStatus.Remove(ecfg.cfg)
631 -
632 - m.dyncfgRespf(fn, 200, "")
633 - m.dyncfgJobRemove(ecfg.cfg)
634 -}
635 -
636 -func (m *Manager) dyncfgConfigUpdate(fn functions.Function) {
637 - id := fn.Args[0]
638 - mn, jn, ok := extractModuleJobName(id)
639 - if !ok {
640 - m.Warningf("dyncfg: update: could not extract module from id (%s)", id)
641 - m.dyncfgRespf(fn, 400, "Invalid ID format. Could not extract module name from ID. Provided ID: %s.", id)
642 - return
643 - }
644 -
645 - ecfg, ok := m.exposedConfigs.lookupByName(mn, jn)
646 - if !ok {
647 - m.Warningf("dyncfg: update: module %s job %s not found", mn, jn)
648 - m.dyncfgRespf(fn, 404, "The specified module '%s' job '%s' is not registered.", mn, jn)
649 - return
650 - }
651 -
652 - cfg, err := configFromPayload(fn)
653 - if err != nil {
654 - m.Warningf("dyncfg: update: module %s: failed to create config from payload: %v", mn, err)
655 - m.dyncfgRespf(fn, 400, "Invalid configuration format. Failed to create configuration from payload: %v.", err)
656 - m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
657 - return
658 - }
659 -
660 - m.dyncfgSetConfigMeta(cfg, mn, jn)
661 -
662 - if ecfg.status == dyncfgRunning && ecfg.cfg.UID() == cfg.UID() {
663 - m.dyncfgRespf(fn, 200, "")
664 - m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
665 - return
666 - }
667 -
668 - job, err := m.createCollectorJob(cfg)
669 - if err != nil {
670 - m.Warningf("dyncfg: update: module %s job %s: failed to apply config: %v", mn, jn, err)
671 - m.dyncfgRespf(fn, 400, "Invalid configuration. Failed to apply configuration: %v.", err)
672 - m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
673 - return
674 - }
675 -
676 - if ecfg.status == dyncfgAccepted {
677 - m.Warningf("dyncfg: update: module %s job %s: updating not allowed in %s", mn, jn, ecfg.status)
678 - m.dyncfgRespf(fn, 403, "Updating data collection job is not allowed in '%s' state.", ecfg.status)
679 - m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
680 - return
681 - }
682 -
683 - m.exposedConfigs.remove(ecfg.cfg)
684 - m.stopRunningJob(ecfg.cfg.FullName())
685 -
686 - scfg := &seenConfig{cfg: cfg, status: dyncfgAccepted}
687 - m.seenConfigs.add(scfg)
688 - m.exposedConfigs.add(scfg)
689 -
690 - if isDyncfg(ecfg.cfg) {
691 - m.seenConfigs.remove(ecfg.cfg)
692 - } else {
693 - // needed to update meta. There is no other way, unfortunately, but to send "create"
694 - defer m.dyncfgJobCreate(scfg.cfg, scfg.status)
695 - }
696 -
697 - if ecfg.status == dyncfgDisabled {
698 - scfg.status = dyncfgDisabled
699 - m.dyncfgRespf(fn, 200, "")
700 - m.dyncfgJobStatus(cfg, scfg.status)
701 - return
702 - }
703 -
704 - if err := job.AutoDetection(); err != nil {
705 - job.Cleanup()
706 - scfg.status = dyncfgFailed
707 - m.dyncfgRespf(fn, 200, "Job update failed: %v", err)
708 - m.dyncfgJobStatus(scfg.cfg, scfg.status)
709 - return
710 - }
711 -
712 - if ok, err := m.FileLock.Lock(scfg.cfg.FullName()); !ok && err == nil {
713 - job.Cleanup()
714 - scfg.status = dyncfgFailed
715 - m.dyncfgRespf(fn, 500, "Job update failed: cannot create file lock.")
716 - m.dyncfgJobStatus(scfg.cfg, scfg.status)
717 - return
718 - }
719 -
720 - scfg.status = dyncfgRunning
721 - m.startRunningJob(job)
722 - m.dyncfgRespf(fn, 200, "")
723 - m.dyncfgJobStatus(scfg.cfg, scfg.status)
67 }
68
726 -func (m *Manager) dyncfgSetConfigMeta(cfg confgroup.Config, module, name string) {
727 - cfg.SetProvider("dyncfg")
728 - cfg.SetSource(fmt.Sprintf("type=dyncfg,module=%s,job=%s", module, name))
729 - cfg.SetSourceType("dyncfg")
730 - cfg.SetModule(module)
731 - cfg.SetName(name)
732 - if def, ok := m.ConfigDefaults.Lookup(module); ok {
733 - cfg.ApplyDefaults(def)
69 +func unmarshalPayload(dst any, fn functions.Function) error {
70 + if v := reflect.ValueOf(dst); v.Kind() != reflect.Ptr || v.IsNil() {
71 + return fmt.Errorf("invalid config: expected a pointer to a struct, got a %s", v.Type())
72 }
735 -}
736 -
737 -func (m *Manager) dyncfgRespPayloadJSON(fn functions.Function, payload string) {
738 - m.dyncfgRespPayload(fn, payload, "application/json")
739 -}
740 -
741 -func (m *Manager) dyncfgRespPayloadYAML(fn functions.Function, payload string) {
742 - m.dyncfgRespPayload(fn, payload, "application/yaml")
743 -}
744 -
745 -func (m *Manager) dyncfgRespPayload(fn functions.Function, payload string, contentType string) {
746 - m.api.FUNCRESULT(netdataapi.FunctionResult{
747 - UID: fn.UID,
748 - ContentType: contentType,
749 - Payload: payload,
750 - Code: "200",
751 - ExpireTimestamp: strconv.FormatInt(time.Now().Unix(), 10),
752 - })
753 -}
754 -
755 -func (m *Manager) dyncfgRespf(fn functions.Function, code int, msgf string, a ...any) {
756 - if fn.UID == "" {
757 - return
758 - }
759 - bs, _ := json.Marshal(struct {
760 - Status int `json:"status"`
761 - Message string `json:"message"`
762 - }{
763 - Status: code,
764 - Message: fmt.Sprintf(msgf, a...),
765 - })
766 - m.api.FUNCRESULT(netdataapi.FunctionResult{
767 - UID: fn.UID,
768 - ContentType: "application/json",
769 - Payload: string(bs),
770 - Code: strconv.Itoa(code),
771 - ExpireTimestamp: strconv.FormatInt(time.Now().Unix(), 10),
772 - })
773 -}
774 -
775 -func userConfigFromPayload(cfg any, jobName string, fn functions.Function) ([]byte, error) {
776 - if v := reflect.ValueOf(cfg); v.Kind() != reflect.Ptr || v.IsNil() {
777 - return nil, fmt.Errorf("invalid config: expected a pointer to a struct, got a %s", v.Type())
778 - }
779 -
73 if fn.ContentType == "application/json" {
781 - if err := json.Unmarshal(fn.Payload, cfg); err != nil {
782 - return nil, err
783 - }
784 - } else {
785 - if err := yaml.Unmarshal(fn.Payload, cfg); err != nil {
786 - return nil, err
787 - }
788 - }
789 -
790 - bs, err := yaml.Marshal(cfg)
791 - if err != nil {
792 - return nil, err
793 - }
794 -
795 - var yms yaml.MapSlice
796 - if err := yaml.Unmarshal(bs, &yms); err != nil {
797 - return nil, err
798 - }
799 -
800 - yms = append([]yaml.MapItem{{Key: "name", Value: jobName}}, yms...)
801 -
802 - v := map[string]any{
803 - "jobs": []any{yms},
804 - }
805 -
806 - bs, err = yaml.Marshal(v)
807 - if err != nil {
808 - return nil, err
809 - }
810 -
811 - return bs, nil
812 -}
813 -
814 -func configFromPayload(fn functions.Function) (confgroup.Config, error) {
815 - var cfg confgroup.Config
816 -
817 - if fn.ContentType == "application/json" {
818 - if err := json.Unmarshal(fn.Payload, &cfg); err != nil {
819 - return nil, err
820 - }
821 -
822 - return cfg.Clone()
823 - }
824 -
825 - if err := yaml.Unmarshal(fn.Payload, &cfg); err != nil {
826 - return nil, err
827 - }
828 -
829 - return cfg, nil
830 -}
831 -
832 -func extractModuleJobName(id string) (mn string, jn string, ok bool) {
833 - if mn, ok = extractModuleName(id); !ok {
834 - return "", "", false
835 - }
836 - if jn, ok = extractJobName(id); !ok {
837 - return "", "", false
838 - }
839 - return mn, jn, true
840 -}
841 -
842 -func extractModuleName(id string) (string, bool) {
843 - id = strings.TrimPrefix(id, dyncfgIDPrefix)
844 - i := strings.IndexByte(id, ':')
845 - if i == -1 {
846 - return id, id != ""
847 - }
848 - return id[:i], true
849 -}
850 -
851 -func extractJobName(id string) (string, bool) {
852 - i := strings.LastIndexByte(id, ':')
853 - if i == -1 {
854 - return "", false
855 - }
856 - return id[i+1:], true
857 -}
858 -
859 -func validateJobName(jobName string) error {
860 - for _, r := range jobName {
861 - if unicode.IsSpace(r) {
862 - return errors.New("contains spaces")
863 - }
864 - switch r {
865 - case '.', ':':
866 - return fmt.Errorf("contains '%c'", r)
867 - }
74 + return json.Unmarshal(fn.Payload, dst)
75 }
869 - return nil
76 + return yaml.Unmarshal(fn.Payload, dst)
77 }
src/go/plugin/go.d/agent/jobmgr/dyncfg_collector.go new
+817
@@ -0,0 +1,817 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package jobmgr
4 +
5 +import (
6 + "context"
7 + "encoding/json"
8 + "errors"
9 + "fmt"
10 + "log/slog"
11 + "strconv"
12 + "strings"
13 + "time"
14 + "unicode"
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/go.d/agent/confgroup"
19 + "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/functions"
20 +
21 + "gopkg.in/yaml.v2"
22 +)
23 +
24 +const (
25 + dyncfgCollectorIDPrefix = "go.d:collector:"
26 + dyncfgCollectorPath = "/collectors/jobs"
27 +)
28 +
29 +func dyncfgModID(name string) string {
30 + return fmt.Sprintf("%s%s", dyncfgCollectorIDPrefix, name)
31 +}
32 +func dyncfgJobID(cfg confgroup.Config) string {
33 + return fmt.Sprintf("%s%s:%s", dyncfgCollectorIDPrefix, cfg.Module(), cfg.Name())
34 +}
35 +
36 +func dyncfgModCmds() string {
37 + return "add schema enable disable test userconfig"
38 +}
39 +func dyncfgJobCmds(cfg confgroup.Config) string {
40 + cmds := "schema get enable disable update restart test userconfig"
41 + if isDyncfg(cfg) {
42 + cmds += " remove"
43 + }
44 + return cmds
45 +}
46 +
47 +func (m *Manager) dyncfgCollectorModuleCreate(name string) {
48 + m.api.CONFIGCREATE(netdataapi.ConfigOpts{
49 + ID: dyncfgModID(name),
50 + Status: dyncfgAccepted.String(),
51 + ConfigType: "template",
52 + Path: dyncfgCollectorPath,
53 + SourceType: "internal",
54 + Source: "internal",
55 + SupportedCommands: dyncfgModCmds(),
56 + })
57 +}
58 +
59 +func (m *Manager) dyncfgCollectorJobCreate(cfg confgroup.Config, status dyncfgStatus) {
60 + m.api.CONFIGCREATE(netdataapi.ConfigOpts{
61 + ID: dyncfgJobID(cfg),
62 + Status: status.String(),
63 + ConfigType: "job",
64 + Path: dyncfgCollectorPath,
65 + SourceType: cfg.SourceType(),
66 + Source: cfg.Source(),
67 + SupportedCommands: dyncfgJobCmds(cfg),
68 + })
69 +}
70 +
71 +func (m *Manager) dyncfgJobRemove(cfg confgroup.Config) {
72 + m.api.CONFIGDELETE(dyncfgJobID(cfg))
73 +}
74 +
75 +func (m *Manager) dyncfgJobStatus(cfg confgroup.Config, status dyncfgStatus) {
76 + m.api.CONFIGSTATUS(dyncfgJobID(cfg), status.String())
77 +}
78 +
79 +func (m *Manager) dyncfgCollectorExec(fn functions.Function) {
80 + action := strings.ToLower(fn.Args[1])
81 +
82 + switch action {
83 + case "userconfig":
84 + m.dyncfgConfigUserconfig(fn)
85 + return
86 + case "test":
87 + m.dyncfgConfigTest(fn)
88 + return
89 + case "schema":
90 + m.dyncfgConfigSchema(fn)
91 + return
92 + }
93 +
94 + select {
95 + case <-m.ctx.Done():
96 + m.dyncfgRespf(fn, 503, "Job manager is shutting down.")
97 + case m.dyncfgCh <- fn:
98 + }
99 +}
100 +
101 +func (m *Manager) dyncfgCollectorSeqExec(fn functions.Function) {
102 + action := strings.ToLower(fn.Args[1])
103 +
104 + switch action {
105 + case "test":
106 + m.dyncfgConfigTest(fn)
107 + case "schema":
108 + m.dyncfgConfigSchema(fn)
109 + case "get":
110 + m.dyncfgConfigGet(fn)
111 + case "restart":
112 + m.dyncfgConfigRestart(fn)
113 + case "enable":
114 + m.dyncfgConfigEnable(fn)
115 + case "disable":
116 + m.dyncfgConfigDisable(fn)
117 + case "add":
118 + m.dyncfgConfigAdd(fn)
119 + case "remove":
120 + m.dyncfgConfigRemove(fn)
121 + case "update":
122 + m.dyncfgConfigUpdate(fn)
123 + default:
124 + m.Warningf("dyncfg: function '%s' action '%s' not implemented", fn.Name, action)
125 + m.dyncfgRespf(fn, 501, "Function '%s' action '%s' is not implemented.", fn.Name, action)
126 + }
127 +}
128 +
129 +func (m *Manager) dyncfgConfigUserconfig(fn functions.Function) {
130 + id := fn.Args[0]
131 + jn := "test"
132 + if len(fn.Args) > 2 {
133 + jn = fn.Args[2]
134 + }
135 +
136 + mn, ok := extractModuleName(id)
137 + if !ok {
138 + m.Warningf("dyncfg: userconfig: could not extract module and job from id (%s)", id)
139 + m.dyncfgRespf(fn, 400,
140 + "Invalid ID format. Could not extract module and job name from ID. Provided ID: %s.", id)
141 + return
142 + }
143 +
144 + creator, ok := m.Modules.Lookup(mn)
145 + if !ok {
146 + m.Warningf("dyncfg: userconfig: module %s not found", mn)
147 + m.dyncfgRespf(fn, 404, "The specified module '%s' is not registered.", mn)
148 + return
149 + }
150 +
151 + if creator.Config == nil || creator.Config() == nil {
152 + m.Warningf("dyncfg: userconfig: module %s: configuration not found", mn)
153 + m.dyncfgRespf(fn, 500, "Module %s does not provide configuration.", mn)
154 + return
155 + }
156 +
157 + bs, err := userConfigFromPayload(creator.Config(), jn, fn)
158 + if err != nil {
159 + m.Warningf("dyncfg: userconfig: module %s: failed to create config from payload: %v", mn, err)
160 + m.dyncfgRespf(fn, 400, "Invalid configuration format. Failed to create configuration from payload: %v.", err)
161 + }
162 +
163 + m.dyncfgRespPayloadYAML(fn, string(bs))
164 +}
165 +
166 +func (m *Manager) dyncfgConfigTest(fn functions.Function) {
167 + id := fn.Args[0]
168 + mn, ok := extractModuleName(id)
169 + if !ok {
170 + m.Warningf("dyncfg: test: could not extract module and job from id (%s)", id)
171 + m.dyncfgRespf(fn, 400,
172 + "Invalid ID format. Could not extract module and job name from ID. Provided ID: %s.", id)
173 + return
174 + }
175 +
176 + jn := "test"
177 + if len(fn.Args) > 2 {
178 + jn = fn.Args[2]
179 + }
180 +
181 + if err := validateJobName(jn); err != nil {
182 + m.Warningf("dyncfg: test: module %s: unacceptable job name '%s': %v", mn, jn, err)
183 + m.dyncfgRespf(fn, 400, "Unacceptable job name '%s': %v.", jn, err)
184 + return
185 + }
186 +
187 + creator, ok := m.Modules.Lookup(mn)
188 + if !ok {
189 + m.Warningf("dyncfg: test: module %s not found", mn)
190 + m.dyncfgRespf(fn, 404, "The specified module '%s' is not registered.", mn)
191 + return
192 + }
193 +
194 + cfg, err := configFromPayload(fn)
195 + if err != nil {
196 + m.Warningf("dyncfg: test: module %s: failed to create config from payload: %v", mn, err)
197 + m.dyncfgRespf(fn, 400, "Invalid configuration format. Failed to create configuration from payload: %v.", err)
198 + return
199 + }
200 +
201 + if cfg.Vnode() != "" {
202 + if _, ok := m.Vnodes[cfg.Vnode()]; !ok {
203 + m.Warningf("dyncfg: test: module %s: vnode %s not found", mn, cfg.Vnode())
204 + m.dyncfgRespf(fn, 400, "The specified vnode '%s' is not registered.", cfg.Vnode())
205 + return
206 + }
207 + }
208 +
209 + cfg.SetModule(mn)
210 + cfg.SetName(jn)
211 +
212 + job := creator.Create()
213 +
214 + if err := applyConfig(cfg, job); err != nil {
215 + m.Warningf("dyncfg: test: module %s: failed to apply config: %v", mn, err)
216 + m.dyncfgRespf(fn, 400, "Invalid configuration. Failed to apply configuration: %v.", err)
217 + return
218 + }
219 +
220 + job.GetBase().Logger = logger.New().With(
221 + slog.String("collector", cfg.Module()),
222 + slog.String("job", cfg.Name()),
223 + )
224 +
225 + defer job.Cleanup(context.Background())
226 +
227 + if err := job.Init(context.Background()); err != nil {
228 + m.dyncfgRespf(fn, 422, "Job initialization failed: %v", err)
229 + return
230 + }
231 + if err := job.Check(context.Background()); err != nil {
232 + m.dyncfgRespf(fn, 422, "Job check failed: %v", err)
233 + return
234 + }
235 +
236 + m.dyncfgRespf(fn, 200, "")
237 +}
238 +
239 +func (m *Manager) dyncfgConfigSchema(fn functions.Function) {
240 + id := fn.Args[0]
241 + mn, ok := extractModuleName(id)
242 + if !ok {
243 + m.Warningf("dyncfg: schema: could not extract module from id (%s)", id)
244 + m.dyncfgRespf(fn, 400, "Invalid ID format. Could not extract module name from ID. Provided ID: %s.", id)
245 + return
246 + }
247 +
248 + mod, ok := m.Modules.Lookup(mn)
249 + if !ok {
250 + m.Warningf("dyncfg: schema: module %s not found", mn)
251 + m.dyncfgRespf(fn, 404, "The specified module '%s' is not registered.", mn)
252 + return
253 + }
254 +
255 + if mod.JobConfigSchema == "" {
256 + m.Warningf("dyncfg: schema: module %s: schema not found", mn)
257 + m.dyncfgRespf(fn, 500, "Module %s configuration schema not found.", mn)
258 + return
259 + }
260 +
261 + m.dyncfgRespPayloadJSON(fn, mod.JobConfigSchema)
262 +}
263 +
264 +func (m *Manager) dyncfgConfigGet(fn functions.Function) {
265 + id := fn.Args[0]
266 + mn, jn, ok := extractModuleJobName(id)
267 + if !ok {
268 + m.Warningf("dyncfg: get: could not extract module and job from id (%s)", id)
269 + m.dyncfgRespf(fn, 400,
270 + "Invalid ID format. Could not extract module and job name from ID. Provided ID: %s.", id)
271 + return
272 + }
273 +
274 + creator, ok := m.Modules.Lookup(mn)
275 + if !ok {
276 + m.Warningf("dyncfg: get: module %s not found", mn)
277 + m.dyncfgRespf(fn, 404, "The specified module '%s' is not registered.", mn)
278 + return
279 + }
280 +
281 + ecfg, ok := m.exposedConfigs.lookupByName(mn, jn)
282 + if !ok {
283 + m.Warningf("dyncfg: get: module %s job %s not found", mn, jn)
284 + m.dyncfgRespf(fn, 404, "The specified module '%s' job '%s' is not registered.", mn, jn)
285 + return
286 + }
287 +
288 + mod := creator.Create()
289 +
290 + if err := applyConfig(ecfg.cfg, mod); err != nil {
291 + m.Warningf("dyncfg: get: module %s job %s failed to apply config: %v", mn, jn, err)
292 + m.dyncfgRespf(fn, 400, "Invalid configuration. Failed to apply configuration: %v.", err)
293 + return
294 + }
295 +
296 + conf := mod.Configuration()
297 + if conf == nil {
298 + m.Warningf("dyncfg: get: module %s: configuration not found", mn)
299 + m.dyncfgRespf(fn, 500, "Module %s does not provide configuration.", mn)
300 + return
301 + }
302 +
303 + bs, err := json.Marshal(conf)
304 + if err != nil {
305 + m.Warningf("dyncfg: get: module %s job %s failed to json marshal config: %v", mn, jn, err)
306 + m.dyncfgRespf(fn, 500, "Failed to convert configuration into JSON: %v.", err)
307 + return
308 + }
309 +
310 + m.dyncfgRespPayloadJSON(fn, string(bs))
311 +}
312 +
313 +func (m *Manager) dyncfgConfigRestart(fn functions.Function) {
314 + id := fn.Args[0]
315 + mn, jn, ok := extractModuleJobName(id)
316 + if !ok {
317 + m.Warningf("dyncfg: restart: could not extract module from id (%s)", id)
318 + m.dyncfgRespf(fn, 400, "Invalid ID format. Could not extract module name from ID. Provided ID: %s.", id)
319 + return
320 + }
321 +
322 + ecfg, ok := m.exposedConfigs.lookupByName(mn, jn)
323 + if !ok {
324 + m.Warningf("dyncfg: restart: module %s job %s not found", mn, jn)
325 + m.dyncfgRespf(fn, 404, "The specified module '%s' job '%s' is not registered.", mn, jn)
326 + return
327 + }
328 +
329 + job, err := m.createCollectorJob(ecfg.cfg)
330 + if err != nil {
331 + m.Warningf("dyncfg: restart: module %s job %s: failed to apply config: %v", mn, jn, err)
332 + m.dyncfgRespf(fn, 400, "Invalid configuration. Failed to apply configuration: %v.", err)
333 + m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
334 + return
335 + }
336 +
337 + switch ecfg.status {
338 + case dyncfgAccepted, dyncfgDisabled:
339 + m.Warningf("dyncfg: restart: module %s job %s: restarting not allowed in '%s' state", mn, jn, ecfg.status)
340 + m.dyncfgRespf(fn, 405, "Restarting data collection job is not allowed in '%s' state.", ecfg.status)
341 + m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
342 + return
343 + case dyncfgRunning:
344 + m.FileStatus.Remove(ecfg.cfg)
345 + m.FileLock.Unlock(ecfg.cfg.FullName())
346 + m.stopRunningJob(ecfg.cfg.FullName())
347 + default:
348 + }
349 +
350 + if err := job.AutoDetection(); err != nil {
351 + job.Cleanup()
352 + ecfg.status = dyncfgFailed
353 + m.dyncfgRespf(fn, 422, "Job restart failed: %v", err)
354 + m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
355 + return
356 + }
357 +
358 + if ok, err := m.FileLock.Lock(ecfg.cfg.FullName()); !ok && err == nil {
359 + job.Cleanup()
360 + ecfg.status = dyncfgFailed
361 + m.dyncfgRespf(fn, 500, "Job restart failed: cannot filelock.")
362 + m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
363 + return
364 + }
365 +
366 + ecfg.status = dyncfgRunning
367 +
368 + if isDyncfg(ecfg.cfg) {
369 + m.FileStatus.Save(ecfg.cfg, ecfg.status.String())
370 + }
371 + m.startRunningJob(job)
372 + m.dyncfgRespf(fn, 200, "")
373 + m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
374 +}
375 +
376 +func (m *Manager) dyncfgConfigEnable(fn functions.Function) {
377 + id := fn.Args[0]
378 + mn, jn, ok := extractModuleJobName(id)
379 + if !ok {
380 + m.Warningf("dyncfg: enable: could not extract module and job from id (%s)", id)
381 + m.dyncfgRespf(fn, 400, "Invalid ID format. Could not extract module and job name from ID. Provided ID: %s.", id)
382 + return
383 + }
384 +
385 + ecfg, ok := m.exposedConfigs.lookupByName(mn, jn)
386 + if !ok {
387 + m.Warningf("dyncfg: enable: module %s job %s not found", mn, jn)
388 + m.dyncfgRespf(fn, 404, "The specified module '%s' job '%s' is not registered.", mn, jn)
389 + return
390 + }
391 +
392 + if ecfg.cfg.FullName() == m.waitCfgOnOff {
393 + m.waitCfgOnOff = ""
394 + }
395 +
396 + switch ecfg.status {
397 + case dyncfgAccepted, dyncfgDisabled, dyncfgFailed:
398 + case dyncfgRunning:
399 + // non-dyncfg update triggers enable/disable
400 + m.dyncfgRespf(fn, 200, "")
401 + m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
402 + return
403 + default:
404 + m.Warningf("dyncfg: enable: module %s job %s: enabling not allowed in %s state", mn, jn, ecfg.status)
405 + m.dyncfgRespf(fn, 405, "Enabling data collection job is not allowed in '%s' state.", ecfg.status)
406 + m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
407 + return
408 + }
409 +
410 + job, err := m.createCollectorJob(ecfg.cfg)
411 + if err != nil {
412 + ecfg.status = dyncfgFailed
413 + m.Warningf("dyncfg: enable: module %s job %s: failed to apply config: %v", mn, jn, err)
414 + m.dyncfgRespf(fn, 400, "Invalid configuration. Failed to apply configuration: %v.", err)
415 + m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
416 + return
417 + }
418 +
419 + if err := job.AutoDetection(); err != nil {
420 + job.Cleanup()
421 + ecfg.status = dyncfgFailed
422 + m.dyncfgRespf(fn, 200, "Job enable failed: %v.", err)
423 +
424 + if isStock(ecfg.cfg) {
425 + m.exposedConfigs.remove(ecfg.cfg)
426 + m.dyncfgJobRemove(ecfg.cfg)
427 + } else {
428 + m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
429 + }
430 +
431 + if job.RetryAutoDetection() && !isDyncfg(ecfg.cfg) {
432 + m.Infof("%s[%s] job detection failed, will retry in %d seconds",
433 + ecfg.cfg.Module(), ecfg.cfg.Name(), job.AutoDetectionEvery())
434 +
435 + ctx, cancel := context.WithCancel(m.ctx)
436 + m.retryingTasks.add(ecfg.cfg, &retryTask{cancel: cancel})
437 + go runRetryTask(ctx, m.addCh, ecfg.cfg)
438 + }
439 + return
440 + }
441 +
442 + if ok, err := m.FileLock.Lock(ecfg.cfg.FullName()); !ok && err == nil {
443 + job.Cleanup()
444 + ecfg.status = dyncfgFailed
445 + m.dyncfgRespf(fn, 500, "Job enable failed: can not filelock.")
446 + m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
447 + return
448 + }
449 +
450 + ecfg.status = dyncfgRunning
451 +
452 + if isDyncfg(ecfg.cfg) {
453 + m.FileStatus.Save(ecfg.cfg, ecfg.status.String())
454 + }
455 +
456 + m.startRunningJob(job)
457 + m.dyncfgRespf(fn, 200, "")
458 + m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
459 +
460 +}
461 +
462 +func (m *Manager) dyncfgConfigDisable(fn functions.Function) {
463 + id := fn.Args[0]
464 + mn, jn, ok := extractModuleJobName(id)
465 + if !ok {
466 + m.Warningf("dyncfg: disable: could not extract module from id (%s)", id)
467 + m.dyncfgRespf(fn, 400, "Invalid ID format. Could not extract module name from ID. Provided ID: %s.", id)
468 + return
469 + }
470 +
471 + ecfg, ok := m.exposedConfigs.lookupByName(mn, jn)
472 + if !ok {
473 + m.Warningf("dyncfg: disable: module %s job %s not found", mn, jn)
474 + m.dyncfgRespf(fn, 404, "The specified module '%s' job '%s' is not registered.", mn, jn)
475 + return
476 + }
477 +
478 + if ecfg.cfg.FullName() == m.waitCfgOnOff {
479 + m.waitCfgOnOff = ""
480 + }
481 +
482 + switch ecfg.status {
483 + case dyncfgDisabled:
484 + m.dyncfgRespf(fn, 200, "")
485 + m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
486 + return
487 + case dyncfgRunning:
488 + m.stopRunningJob(ecfg.cfg.FullName())
489 + if isDyncfg(ecfg.cfg) {
490 + m.FileStatus.Remove(ecfg.cfg)
491 + }
492 + m.FileLock.Unlock(ecfg.cfg.FullName())
493 + default:
494 + }
495 +
496 + ecfg.status = dyncfgDisabled
497 + m.dyncfgRespf(fn, 200, "")
498 + m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
499 +}
500 +
501 +func (m *Manager) dyncfgConfigAdd(fn functions.Function) {
502 + if len(fn.Args) < 3 {
503 + m.Warningf("dyncfg: add: missing required arguments, want 3 got %d", len(fn.Args))
504 + m.dyncfgRespf(fn, 400, "Missing required arguments. Need at least 3, but got %d.", len(fn.Args))
505 + return
506 + }
507 +
508 + id := fn.Args[0]
509 + jn := fn.Args[2]
510 + mn, ok := extractModuleName(id)
511 + if !ok {
512 + m.Warningf("dyncfg: add: could not extract module from id (%s)", id)
513 + m.dyncfgRespf(fn, 400, "Invalid ID format. Could not extract module name from ID. Provided ID: %s.", id)
514 + return
515 + }
516 +
517 + if len(fn.Payload) == 0 {
518 + m.Warningf("dyncfg: add: module %s job %s missing configuration payload.", mn, jn)
519 + m.dyncfgRespf(fn, 400, "Missing configuration payload.")
520 + return
521 + }
522 +
523 + if err := validateJobName(jn); err != nil {
524 + m.Warningf("dyncfg: add: module %s: unacceptable job name '%s': %v", mn, jn, err)
525 + m.dyncfgRespf(fn, 400, "Unacceptable job name '%s': %v.", jn, err)
526 + return
527 + }
528 +
529 + cfg, err := configFromPayload(fn)
530 + if err != nil {
531 + m.Warningf("dyncfg: add: module %s job %s: failed to create config from payload: %v", mn, jn, err)
532 + m.dyncfgRespf(fn, 400, "Invalid configuration format. Failed to create configuration from payload: %v.", err)
533 + return
534 + }
535 +
536 + m.dyncfgSetConfigMeta(cfg, mn, jn)
537 +
538 + if _, err := m.createCollectorJob(cfg); err != nil {
539 + m.Warningf("dyncfg: add: module %s job %s: failed to apply config: %v", mn, jn, err)
540 + m.dyncfgRespf(fn, 400, "Invalid configuration. Failed to apply configuration: %v.", err)
541 + return
542 + }
543 +
544 + if ecfg, ok := m.exposedConfigs.lookup(cfg); ok {
545 + if scfg, ok := m.seenConfigs.lookup(ecfg.cfg); ok && isDyncfg(scfg.cfg) {
546 + m.seenConfigs.remove(ecfg.cfg)
547 + }
548 + m.exposedConfigs.remove(ecfg.cfg)
549 + m.stopRunningJob(ecfg.cfg.FullName())
550 + }
551 +
552 + scfg := &seenConfig{cfg: cfg, status: dyncfgAccepted}
553 + ecfg := scfg
554 + m.seenConfigs.add(scfg)
555 + m.exposedConfigs.add(ecfg)
556 +
557 + m.dyncfgRespf(fn, 202, "")
558 + m.dyncfgCollectorJobCreate(ecfg.cfg, ecfg.status)
559 +}
560 +
561 +func (m *Manager) dyncfgConfigRemove(fn functions.Function) {
562 + id := fn.Args[0]
563 + mn, jn, ok := extractModuleJobName(id)
564 + if !ok {
565 + m.Warningf("dyncfg: remove: could not extract module and job from id (%s)", id)
566 + m.dyncfgRespf(fn, 400, "Invalid ID format. Could not extract module and job name from ID. Provided ID: %s.", id)
567 + return
568 + }
569 +
570 + ecfg, ok := m.exposedConfigs.lookupByName(mn, jn)
571 + if !ok {
572 + m.Warningf("dyncfg: remove: module %s job %s not found", mn, jn)
573 + m.dyncfgRespf(fn, 404, "The specified module '%s' job '%s' is not registered.", mn, jn)
574 + return
575 + }
576 +
577 + if !isDyncfg(ecfg.cfg) {
578 + m.Warningf("dyncfg: remove: module %s job %s: can not remove jobs of type %s", mn, jn, ecfg.cfg.SourceType())
579 + m.dyncfgRespf(fn, 405, "Removing jobs of type '%s' is not supported. Only 'dyncfg' jobs can be removed.", ecfg.cfg.SourceType())
580 + return
581 + }
582 +
583 + m.seenConfigs.remove(ecfg.cfg)
584 + m.exposedConfigs.remove(ecfg.cfg)
585 + m.stopRunningJob(ecfg.cfg.FullName())
586 + m.FileLock.Unlock(ecfg.cfg.FullName())
587 + m.FileStatus.Remove(ecfg.cfg)
588 +
589 + m.dyncfgRespf(fn, 200, "")
590 + m.dyncfgJobRemove(ecfg.cfg)
591 +}
592 +
593 +func (m *Manager) dyncfgConfigUpdate(fn functions.Function) {
594 + id := fn.Args[0]
595 + mn, jn, ok := extractModuleJobName(id)
596 + if !ok {
597 + m.Warningf("dyncfg: update: could not extract module from id (%s)", id)
598 + m.dyncfgRespf(fn, 400, "Invalid ID format. Could not extract module name from ID. Provided ID: %s.", id)
599 + return
600 + }
601 +
602 + ecfg, ok := m.exposedConfigs.lookupByName(mn, jn)
603 + if !ok {
604 + m.Warningf("dyncfg: update: module %s job %s not found", mn, jn)
605 + m.dyncfgRespf(fn, 404, "The specified module '%s' job '%s' is not registered.", mn, jn)
606 + return
607 + }
608 +
609 + cfg, err := configFromPayload(fn)
610 + if err != nil {
611 + m.Warningf("dyncfg: update: module %s: failed to create config from payload: %v", mn, err)
612 + m.dyncfgRespf(fn, 400, "Invalid configuration format. Failed to create configuration from payload: %v.", err)
613 + m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
614 + return
615 + }
616 +
617 + m.dyncfgSetConfigMeta(cfg, mn, jn)
618 +
619 + if ecfg.status == dyncfgRunning && ecfg.cfg.UID() == cfg.UID() {
620 + m.dyncfgRespf(fn, 200, "")
621 + m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
622 + return
623 + }
624 +
625 + job, err := m.createCollectorJob(cfg)
626 + if err != nil {
627 + m.Warningf("dyncfg: update: module %s job %s: failed to apply config: %v", mn, jn, err)
628 + m.dyncfgRespf(fn, 400, "Invalid configuration. Failed to apply configuration: %v.", err)
629 + m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
630 + return
631 + }
632 +
633 + if ecfg.status == dyncfgAccepted {
634 + m.Warningf("dyncfg: update: module %s job %s: updating not allowed in %s", mn, jn, ecfg.status)
635 + m.dyncfgRespf(fn, 403, "Updating data collection job is not allowed in '%s' state.", ecfg.status)
636 + m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
637 + return
638 + }
639 +
640 + m.exposedConfigs.remove(ecfg.cfg)
641 + m.stopRunningJob(ecfg.cfg.FullName())
642 +
643 + scfg := &seenConfig{cfg: cfg, status: dyncfgAccepted}
644 + m.seenConfigs.add(scfg)
645 + m.exposedConfigs.add(scfg)
646 +
647 + if isDyncfg(ecfg.cfg) {
648 + m.seenConfigs.remove(ecfg.cfg)
649 + } else {
650 + // Needed to update meta. There is no other way, unfortunately, but to send "create".
651 + defer m.dyncfgCollectorJobCreate(scfg.cfg, scfg.status)
652 + }
653 +
654 + if ecfg.status == dyncfgDisabled {
655 + scfg.status = dyncfgDisabled
656 + m.dyncfgRespf(fn, 200, "")
657 + m.dyncfgJobStatus(cfg, scfg.status)
658 + return
659 + }
660 +
661 + if err := job.AutoDetection(); err != nil {
662 + job.Cleanup()
663 + scfg.status = dyncfgFailed
664 + m.dyncfgRespf(fn, 200, "Job update failed: %v", err)
665 + m.dyncfgJobStatus(scfg.cfg, scfg.status)
666 + return
667 + }
668 +
669 + if ok, err := m.FileLock.Lock(scfg.cfg.FullName()); !ok && err == nil {
670 + job.Cleanup()
671 + scfg.status = dyncfgFailed
672 + m.dyncfgRespf(fn, 500, "Job update failed: cannot create file lock.")
673 + m.dyncfgJobStatus(scfg.cfg, scfg.status)
674 + return
675 + }
676 +
677 + scfg.status = dyncfgRunning
678 + m.startRunningJob(job)
679 + m.dyncfgRespf(fn, 200, "")
680 + m.dyncfgJobStatus(scfg.cfg, scfg.status)
681 +}
682 +
683 +func (m *Manager) dyncfgSetConfigMeta(cfg confgroup.Config, module, name string) {
684 + cfg.SetProvider("dyncfg")
685 + cfg.SetSource(fmt.Sprintf("type=dyncfg,module=%s,job=%s", module, name))
686 + cfg.SetSourceType("dyncfg")
687 + cfg.SetModule(module)
688 + cfg.SetName(name)
689 + if def, ok := m.ConfigDefaults.Lookup(module); ok {
690 + cfg.ApplyDefaults(def)
691 + }
692 +}
693 +
694 +func (m *Manager) dyncfgRespPayloadJSON(fn functions.Function, payload string) {
695 + m.dyncfgRespPayload(fn, payload, "application/json")
696 +}
697 +
698 +func (m *Manager) dyncfgRespPayloadYAML(fn functions.Function, payload string) {
699 + m.dyncfgRespPayload(fn, payload, "application/yaml")
700 +}
701 +
702 +func (m *Manager) dyncfgRespPayload(fn functions.Function, payload string, contentType string) {
703 + m.api.FUNCRESULT(netdataapi.FunctionResult{
704 + UID: fn.UID,
705 + ContentType: contentType,
706 + Payload: payload,
707 + Code: "200",
708 + ExpireTimestamp: strconv.FormatInt(time.Now().Unix(), 10),
709 + })
710 +}
711 +
712 +func (m *Manager) dyncfgRespf(fn functions.Function, code int, msgf string, a ...any) {
713 + if fn.UID == "" {
714 + return
715 + }
716 + bs, _ := json.Marshal(struct {
717 + Status int `json:"status"`
718 + Message string `json:"message"`
719 + }{
720 + Status: code,
721 + Message: fmt.Sprintf(msgf, a...),
722 + })
723 + m.api.FUNCRESULT(netdataapi.FunctionResult{
724 + UID: fn.UID,
725 + ContentType: "application/json",
726 + Payload: string(bs),
727 + Code: strconv.Itoa(code),
728 + ExpireTimestamp: strconv.FormatInt(time.Now().Unix(), 10),
729 + })
730 +}
731 +
732 +func userConfigFromPayload(cfg any, jobName string, fn functions.Function) ([]byte, error) {
733 + if err := unmarshalPayload(cfg, fn); err != nil {
734 + return nil, err
735 + }
736 +
737 + bs, err := yaml.Marshal(cfg)
738 + if err != nil {
739 + return nil, err
740 + }
741 +
742 + var yms yaml.MapSlice
743 + if err := yaml.Unmarshal(bs, &yms); err != nil {
744 + return nil, err
745 + }
746 +
747 + yms = append([]yaml.MapItem{{Key: "name", Value: jobName}}, yms...)
748 +
749 + v := map[string]any{
750 + "jobs": []any{yms},
751 + }
752 +
753 + bs, err = yaml.Marshal(v)
754 + if err != nil {
755 + return nil, err
756 + }
757 +
758 + return bs, nil
759 +}
760 +
761 +func configFromPayload(fn functions.Function) (confgroup.Config, error) {
762 + var cfg confgroup.Config
763 +
764 + if fn.ContentType == "application/json" {
765 + if err := json.Unmarshal(fn.Payload, &cfg); err != nil {
766 + return nil, err
767 + }
768 +
769 + return cfg.Clone()
770 + }
771 +
772 + if err := yaml.Unmarshal(fn.Payload, &cfg); err != nil {
773 + return nil, err
774 + }
775 +
776 + return cfg, nil
777 +}
778 +
779 +func extractModuleJobName(id string) (mn string, jn string, ok bool) {
780 + if mn, ok = extractModuleName(id); !ok {
781 + return "", "", false
782 + }
783 + if jn, ok = extractJobName(id); !ok {
784 + return "", "", false
785 + }
786 + return mn, jn, true
787 +}
788 +
789 +func extractModuleName(id string) (string, bool) {
790 + id = strings.TrimPrefix(id, dyncfgCollectorIDPrefix)
791 + i := strings.IndexByte(id, ':')
792 + if i == -1 {
793 + return id, id != ""
794 + }
795 + return id[:i], true
796 +}
797 +
798 +func extractJobName(id string) (string, bool) {
799 + i := strings.LastIndexByte(id, ':')
800 + if i == -1 {
801 + return "", false
802 + }
803 + return id[i+1:], true
804 +}
805 +
806 +func validateJobName(jobName string) error {
807 + for _, r := range jobName {
808 + if unicode.IsSpace(r) {
809 + return errors.New("contains spaces")
810 + }
811 + switch r {
812 + case '.', ':':
813 + return fmt.Errorf("contains '%c'", r)
814 + }
815 + }
816 + return nil
817 +}
src/go/plugin/go.d/agent/jobmgr/dyncfg_vnode.go new
+265
@@ -0,0 +1,265 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package jobmgr
4 +
5 +import (
6 + _ "embed"
7 + "encoding/json"
8 + "fmt"
9 + "strings"
10 +
11 + "github.com/google/uuid"
12 + "gopkg.in/yaml.v2"
13 +
14 + "github.com/netdata/netdata/go/plugins/pkg/netdataapi"
15 + "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/confgroup"
16 + "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/functions"
17 + "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/module"
18 + "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/vnodes"
19 +)
20 +
21 +//go:embed "vnode_config_schema.json"
22 +var vnodeConfigSchema string
23 +
24 +const (
25 + dyncfgVnodeIDPrefix = "go.d:vnode:"
26 + dyncfgVnodePath = "/vnodes/vnode"
27 +)
28 +
29 +func (m *Manager) dyncfgVnodeModuleCreate() {
30 + m.api.CONFIGCREATE(netdataapi.ConfigOpts{
31 + ID: fmt.Sprintf("%s%s", dyncfgVnodeIDPrefix, "vnode"),
32 + Status: dyncfgAccepted.String(),
33 + ConfigType: "template",
34 + Path: dyncfgVnodePath,
35 + SourceType: "internal",
36 + Source: "internal",
37 + SupportedCommands: "add schema userconfig",
38 + })
39 +}
40 +
41 +func (m *Manager) dyncfgVnodeJobCreate(cfg *vnodes.VirtualNode, status dyncfgStatus) {
42 + cmds := "userconfig schema get update"
43 + if cfg.SourceType == confgroup.TypeDyncfg {
44 + cmds += " remove"
45 + }
46 + m.api.CONFIGCREATE(netdataapi.ConfigOpts{
47 + ID: fmt.Sprintf("%s%s:%s", dyncfgVnodeIDPrefix, "vnode", cfg.Name),
48 + Status: status.String(),
49 + ConfigType: "job",
50 + Path: dyncfgVnodePath,
51 + SourceType: cfg.SourceType,
52 + Source: cfg.Source,
53 + SupportedCommands: cmds,
54 + })
55 +}
56 +
57 +func (m *Manager) dyncfgVnodeExec(fn functions.Function) {
58 + action := strings.ToLower(fn.Args[1])
59 +
60 + switch action {
61 + case "userconfig":
62 + m.dyncfgVnodeUserconfig(fn)
63 + return
64 + case "schema":
65 + m.dyncfgRespPayloadJSON(fn, vnodeConfigSchema)
66 + return
67 + case "get":
68 + m.dyncfgVnodeGet(fn)
69 + return
70 + case "add":
71 + m.dyncfgVnodeAdd(fn)
72 + return
73 + }
74 +
75 + select {
76 + case <-m.ctx.Done():
77 + m.dyncfgRespf(fn, 503, "Job manager is shutting down.")
78 + case m.dyncfgCh <- fn:
79 + }
80 +}
81 +
82 +func (m *Manager) dyncfgVnodeSeqExec(fn functions.Function) {
83 + action := strings.ToLower(fn.Args[1])
84 +
85 + switch action {
86 + case "update":
87 + m.dyncfgVnodeUpdate(fn)
88 + case "remove":
89 + m.dyncfgVnodeRemove(fn)
90 + default:
91 + m.Warningf("dyncfg: function '%s' action '%s' not implemented", fn.Name, action)
92 + m.dyncfgRespf(fn, 501, "Function '%s' action '%s' is not implemented.", fn.Name, action)
93 + }
94 +}
95 +
96 +func (m *Manager) dyncfgVnodeGet(fn functions.Function) {
97 + name := strings.TrimPrefix(fn.Args[0], fmt.Sprintf("%svnode:", dyncfgVnodeIDPrefix))
98 +
99 + cfg, ok := m.Vnodes[name]
100 + if !ok {
101 + m.Warningf("dyncfg: get: vnode %s not found", name)
102 + m.dyncfgRespf(fn, 404, "The specified vnode '%s' is not registered.", name)
103 + return
104 + }
105 +
106 + bs, err := json.Marshal(cfg)
107 + if err != nil {
108 + m.Warningf("dyncfg: get: vnode job %s failed to json marshal config: %v", name, err)
109 + m.dyncfgRespf(fn, 500, "Failed to convert configuration into JSON: %v.", err)
110 + return
111 + }
112 +
113 + m.dyncfgRespPayloadJSON(fn, string(bs))
114 +}
115 +
116 +func (m *Manager) dyncfgVnodeAdd(fn functions.Function) {
117 + if len(fn.Args) < 3 {
118 + m.Warningf("dyncfg: add: missing required arguments, want 3 got %d", len(fn.Args))
119 + m.dyncfgRespf(fn, 400, "Missing required arguments. Need at least 3, but got %d.", len(fn.Args))
120 + return
121 + }
122 +
123 + name := fn.Args[2]
124 +
125 + if len(fn.Payload) == 0 {
126 + m.Warningf("dyncfg: add: vnode job %s missing configuration payload.", name)
127 + m.dyncfgRespf(fn, 400, "Missing configuration payload.")
128 + return
129 + }
130 +
131 + cfg, err := vnodeConfigFromPayload(fn)
132 + if err != nil {
133 + m.Warningf("dyncfg: add: vnode job %s: failed to create config from payload: %v", name, err)
134 + m.dyncfgRespf(fn, 400, "Failed to create configuration from payload. Invalid configuration format: %v.", err)
135 + return
136 + }
137 +
138 + if err := uuid.Validate(cfg.GUID); err != nil {
139 + m.Warningf("dyncfg: add: vnode job %s: invalid guid: %v", name, err)
140 + m.dyncfgRespf(fn, 400, "Failed to create configuration from payload. Invalid guid format: %v.", err)
141 + return
142 + }
143 +
144 + cfg.Name = name
145 + if cfg.Hostname == "" {
146 + cfg.Hostname = name
147 + }
148 + cfg.SourceType = confgroup.TypeDyncfg
149 + cfg.Source = "type=dyncfg"
150 +
151 + m.Vnodes[name] = cfg
152 +
153 + m.dyncfgRespf(fn, 202, "")
154 + m.dyncfgVnodeJobCreate(cfg, dyncfgRunning)
155 +}
156 +
157 +func (m *Manager) dyncfgVnodeRemove(fn functions.Function) {
158 + id := fn.Args[0]
159 + name := strings.TrimPrefix(id, fmt.Sprintf("%svnode:", dyncfgVnodeIDPrefix))
160 +
161 + vnode, ok := m.Vnodes[name]
162 + if !ok {
163 + m.Warningf("dyncfg: remove: vnode %s not found", name)
164 + m.dyncfgRespf(fn, 404, "The specified vnode '%s' is not registered.", name)
165 + return
166 + }
167 + if vnode.SourceType != confgroup.TypeDyncfg {
168 + m.Warningf("dyncfg: remove: module vnode %s: can not remove vnode of type %s", vnode.Name, vnode.SourceType)
169 + m.dyncfgRespf(fn, 405, "Removing vnode of type '%s' is not supported. Only 'dyncfg' vnodes can be removed.", vnode.SourceType)
170 + return
171 + }
172 +
173 + var s strings.Builder
174 + for _, ecfg := range m.exposedConfigs.items {
175 + if ecfg.cfg.Vnode() == vnode.Name {
176 + if s.Len() > 0 {
177 + s.WriteString(", ")
178 + }
179 + s.WriteString(fmt.Sprintf("%s:%s", ecfg.cfg.Module(), ecfg.cfg.Name()))
180 + }
181 + }
182 + if s.Len() > 0 {
183 + m.Warningf("dyncfg: remove: vnode %s has running jobs (%s)", name, s.String())
184 + m.dyncfgRespf(fn, 404, "The specified vnode '%s' has running jobs (%s).", name, s.String())
185 + return
186 + }
187 +
188 + delete(m.Vnodes, name)
189 + m.api.CONFIGDELETE(id)
190 + m.dyncfgRespf(fn, 200, "")
191 +}
192 +
193 +func (m *Manager) dyncfgVnodeUpdate(fn functions.Function) {
194 + name := strings.TrimPrefix(fn.Args[0], fmt.Sprintf("%svnode:", dyncfgVnodeIDPrefix))
195 +
196 + _, ok := m.Vnodes[name]
197 + if !ok {
198 + m.Warningf("dyncfg: remove: vnode %s not found", name)
199 + m.dyncfgRespf(fn, 404, "The specified vnode '%s' is not registered.", name)
200 + return
201 + }
202 +
203 + cfg, err := vnodeConfigFromPayload(fn)
204 + if err != nil {
205 + m.Warningf("dyncfg: remove: vnode: failed to create config from payload: %v", err)
206 + m.dyncfgRespf(fn, 400, "Invalid configuration format. Failed to create configuration from payload: %v.", err)
207 + return
208 + }
209 +
210 + m.Vnodes[name] = cfg
211 +
212 + m.runningJobs.forEach(func(_ string, job *module.Job) {
213 + if job.Vnode().Name == name {
214 + job.UpdateVnode(cfg)
215 + }
216 + })
217 +
218 + m.dyncfgRespf(fn, 202, "")
219 + m.dyncfgVnodeJobCreate(cfg, dyncfgRunning)
220 +}
221 +
222 +func (m *Manager) dyncfgVnodeUserconfig(fn functions.Function) {
223 + bs, err := m.vnodeUserconfigFromPayload(fn)
224 + if err != nil {
225 + m.Warningf("dyncfg: userconfig: vnode: failed to create config from payload: %v", err)
226 + m.dyncfgRespf(fn, 400, "Invalid configuration format. Failed to create configuration from payload: %v.", err)
227 + return
228 + }
229 +
230 + m.dyncfgRespPayloadYAML(fn, string(bs))
231 +}
232 +
233 +func vnodeConfigFromPayload(fn functions.Function) (*vnodes.VirtualNode, error) {
234 + var cfg vnodes.VirtualNode
235 +
236 + if err := unmarshalPayload(&cfg, fn); err != nil {
237 + return nil, err
238 + }
239 +
240 + return &cfg, nil
241 +}
242 +
243 +func (m *Manager) vnodeUserconfigFromPayload(fn functions.Function) ([]byte, error) {
244 + cfg, err := vnodeConfigFromPayload(fn)
245 + if err != nil {
246 + return nil, err
247 + }
248 +
249 + name := "test"
250 + if len(fn.Args) > 2 {
251 + name = fn.Args[2]
252 + }
253 +
254 + cfg.Name = name
255 + if cfg.Hostname == "" {
256 + cfg.Hostname = name
257 + }
258 +
259 + bs, err := yaml.Marshal([]any{cfg})
260 + if err != nil {
261 + return nil, err
262 + }
263 +
264 + return bs, nil
265 +}
src/go/plugin/go.d/agent/jobmgr/manager.go
+29 -20
@@ -8,6 +8,7 @@ import (
8 "io"
9 "log/slog"
10 "os"
11 + "strings"
12 "sync"
13 "time"
14
@@ -18,6 +19,7 @@ import (
19 "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/confgroup"
20 "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/functions"
21 "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/module"
22 + "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/vnodes"
23
24 "github.com/mattn/go-isatty"
25 "gopkg.in/yaml.v2"
@@ -34,9 +36,10 @@ func New() *Manager {
36 FileLock: noop{},
37 FileStatus: noop{},
38 FileStatusStore: noop{},
37 - Vnodes: noop{},
39 FnReg: noop{},
40
41 + Vnodes: make(map[string]*vnodes.VirtualNode),
42 +
43 discoveredConfigs: newDiscoveredConfigsCache(),
44 seenConfigs: newSeenConfigCache(),
45 exposedConfigs: newExposedConfigCache(),
@@ -64,8 +67,8 @@ type Manager struct {
67 FileLock FileLocker
68 FileStatus FileStatus
69 FileStatusStore FileStatusStore
67 - Vnodes Vnodes
70 FnReg FunctionRegistry
71 + Vnodes map[string]*vnodes.VirtualNode
72
73 discoveredConfigs *discoveredConfigs
74 seenConfigs *seenConfigs
@@ -90,8 +93,14 @@ func (m *Manager) Run(ctx context.Context, in chan []*confgroup.Group) {
93
94 m.FnReg.Register("config", m.dyncfgConfig)
95
96 + //m.dyncfgVnodeModuleCreate()
97 + //
98 + //for _, cfg := range m.Vnodes {
99 + // m.dyncfgVnodeJobCreate(cfg, dyncfgRunning)
100 + //}
101 +
102 for name := range m.Modules {
94 - m.dyncfgModuleCreate(name)
103 + m.dyncfgCollectorModuleCreate(name)
104 }
105
106 var wg sync.WaitGroup
@@ -134,7 +143,7 @@ func (m *Manager) run() {
143 case <-m.ctx.Done():
144 return
145 case fn := <-m.dyncfgCh:
137 - m.dyncfgConfigExec(fn)
146 + m.dyncfgCollectorSeqExec(fn)
147 }
148 } else {
149 select {
@@ -145,7 +154,14 @@ func (m *Manager) run() {
154 case cfg := <-m.rmCh:
155 m.removeConfig(cfg)
156 case fn := <-m.dyncfgCh:
148 - m.dyncfgConfigExec(fn)
157 + switch id := fn.Args[0]; true {
158 + case strings.HasPrefix(id, dyncfgCollectorIDPrefix):
159 + m.dyncfgCollectorSeqExec(fn)
160 + case strings.HasPrefix(id, dyncfgVnodeIDPrefix):
161 + m.dyncfgVnodeSeqExec(fn)
162 + default:
163 + m.dyncfgRespf(fn, 503, "unknown function '%s' (%s).", fn.Name, id)
164 + }
165 }
166 }
167 }
@@ -185,7 +201,7 @@ func (m *Manager) addConfig(cfg confgroup.Config) {
201 ecfg = scfg
202 }
203
188 - m.dyncfgJobCreate(ecfg.cfg, ecfg.status)
204 + m.dyncfgCollectorJobCreate(ecfg.cfg, ecfg.status)
205
206 if isTerminal || m.PluginName == "nodyncfg" { // FIXME: quick fix of TestAgent_Run (agent_test.go)
207 m.dyncfgConfigEnable(functions.Function{Args: []string{dyncfgJobID(ecfg.cfg), "enable"}})
@@ -274,21 +290,14 @@ func (m *Manager) createCollectorJob(cfg confgroup.Config) (*module.Job, error)
290 return nil, fmt.Errorf("can not find %s module", cfg.Module())
291 }
292
277 - var vnode struct {
278 - guid string
279 - hostname string
280 - labels map[string]string
281 - }
293 + var vnode *vnodes.VirtualNode
294
295 if cfg.Vnode() != "" {
284 - n, ok := m.Vnodes.Lookup(cfg.Vnode())
285 - if !ok {
296 + n, ok := m.Vnodes[cfg.Vnode()]
297 + if !ok || n == nil {
298 return nil, fmt.Errorf("vnode '%s' is not found", cfg.Vnode())
299 }
288 -
289 - vnode.guid = n.GUID
290 - vnode.hostname = n.Hostname
291 - vnode.labels = n.Labels
300 + vnode = n
301 }
302
303 m.Debugf("creating %s[%s] job, config: %v", cfg.Module(), cfg.Name(), cfg)
@@ -311,9 +320,9 @@ func (m *Manager) createCollectorJob(cfg confgroup.Config) (*module.Job, error)
320 IsStock: cfg.SourceType() == "stock",
321 Module: mod,
322 Out: m.Out,
314 - VnodeGUID: vnode.guid,
315 - VnodeHostname: vnode.hostname,
316 - VnodeLabels: vnode.labels,
323 + }
324 + if vnode != nil {
325 + jobCfg.Vnode = *vnode.Copy()
326 }
327
328 job := module.NewJob(jobCfg)
src/go/plugin/go.d/agent/jobmgr/vnode_config_schema.json new
+41
@@ -0,0 +1,41 @@
1 +{
2 + "jsonSchema": {
3 + "$schema": "http://json-schema.org/draft-07/schema#",
4 + "title": "Virtual Node configuration.",
5 + "type": "object",
6 + "properties": {
7 + "guid": {
8 + "title": "GUID",
9 + "description": "Uniquely identifies the node.",
10 + "type": "string"
11 + },
12 + "hostname": {
13 + "title": "Hostname",
14 + "description": "The hostname of the virtual node.",
15 + "type": "string"
16 + },
17 + "labels": {
18 + "title": "Labels",
19 + "description": "Additional key-value pairs to associate with the Virtual Node.",
20 + "type": [
21 + "object",
22 + "null"
23 + ],
24 + "additionalProperties": {
25 + "type": "string"
26 + }
27 + }
28 + },
29 + "required": [
30 + "guid"
31 + ],
32 + "patternProperties": {
33 + "^name$": {}
34 + }
35 + },
36 + "uiSchema": {
37 + "uiOptions": {
38 + "fullPage": true
39 + }
40 + }
41 +}
src/go/plugin/go.d/agent/module/job.go
+46 -30
@@ -16,6 +16,7 @@ import (
16
17 "github.com/netdata/netdata/go/plugins/logger"
18 "github.com/netdata/netdata/go/plugins/pkg/netdataapi"
19 + "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/vnodes"
20 "github.com/netdata/netdata/go/plugins/plugin/go.d/pkg/metrix"
21 )
22
@@ -75,10 +76,7 @@ type JobConfig struct {
76 AutoDetectEvery int
77 Priority int
78 IsStock bool
78 -
79 - VnodeGUID string
80 - VnodeHostname string
81 - VnodeLabels map[string]string
79 + Vnode vnodes.VirtualNode
80 }
81
82 const (
@@ -114,10 +112,8 @@ func NewJob(cfg JobConfig) *Job {
112 tick: make(chan int),
113 buf: &buf,
114 api: netdataapi.New(&buf),
117 -
118 - vnodeGUID: cfg.VnodeGUID,
119 - vnodeHostname: cfg.VnodeHostname,
120 - vnodeLabels: cfg.VnodeLabels,
115 + vnode: cfg.Vnode,
116 + updVnode: make(chan *vnodes.VirtualNode, 1),
117 }
118
119 log := logger.New().With(
@@ -163,15 +159,14 @@ type Job struct {
159 buf *bytes.Buffer
160 api *netdataapi.API
161
162 + vnodeCreated bool
163 + vnode vnodes.VirtualNode
164 + updVnode chan *vnodes.VirtualNode
165 +
166 retries int
167 prevRun time.Time
168
169 stop chan struct{}
170 -
171 - vnodeCreated bool
172 - vnodeGUID string
173 - vnodeHostname string
174 - vnodeLabels map[string]string
170 }
171
172 // NetdataChartIDMaxLength is the chart ID max length. See RRD_ID_LENGTH_MAX in the netdata source code.
@@ -211,6 +206,10 @@ func (j *Job) Configuration() any {
206 return j.module.Configuration()
207 }
208
209 +func (j *Job) Vnode() vnodes.VirtualNode {
210 + return j.vnode
211 +}
212 +
213 // AutoDetection invokes init, check and postCheck. It handles panic.
214 func (j *Job) AutoDetection() (err error) {
215 defer func() {
@@ -258,6 +257,14 @@ func (j *Job) AutoDetection() (err error) {
257 return nil
258 }
259
260 +func (j *Job) UpdateVnode(vnode *vnodes.VirtualNode) {
261 + select {
262 + case <-j.updVnode:
263 + default:
264 + }
265 + j.updVnode <- vnode
266 +}
267 +
268 // Tick Tick.
269 func (j *Job) Tick(clock int) {
270 select {
@@ -305,15 +312,15 @@ func (j *Job) Cleanup() {
312 return
313 }
314
308 - if !j.vnodeCreated && j.vnodeGUID != "" {
315 + if !j.vnodeCreated && j.vnode.GUID != "" {
316 j.api.HOSTINFO(netdataapi.HostInfo{
310 - GUID: j.vnodeGUID,
311 - Hostname: j.vnodeHostname,
312 - Labels: j.vnodeLabels,
317 + GUID: j.vnode.GUID,
318 + Hostname: j.vnode.Hostname,
319 + Labels: j.vnode.Labels,
320 })
321 j.vnodeCreated = true
322 }
316 - j.api.HOST(j.vnodeGUID)
323 + j.api.HOST(j.vnode.GUID)
324
325 if j.collectStatusChart.created {
326 j.collectStatusChart.MarkRemove()
@@ -410,32 +417,41 @@ func (j *Job) collect() (result map[string]int64) {
417 }
418
419 func (j *Job) processMetrics(metrics map[string]int64, startTime time.Time, sinceLastRun int) bool {
420 + var createChart bool
421 + if j.module.VirtualNode() != nil {
422 + select {
423 + case vnode := <-j.updVnode:
424 + j.vnodeCreated = false
425 + createChart = j.vnode.GUID != vnode.GUID
426 + j.vnode = *vnode.Copy()
427 + default:
428 + }
429 + }
430 +
431 if !j.vnodeCreated {
414 - if j.vnodeGUID == "" {
432 + if j.vnode.GUID == "" {
433 if v := j.module.VirtualNode(); v != nil && v.GUID != "" && v.Hostname != "" {
416 - j.vnodeGUID = v.GUID
417 - j.vnodeHostname = v.Hostname
418 - j.vnodeLabels = v.Labels
434 + j.vnode = *v
435 }
436 }
421 - if j.vnodeGUID != "" {
437 + if j.vnode.GUID != "" {
438 j.api.HOSTINFO(netdataapi.HostInfo{
423 - GUID: j.vnodeGUID,
424 - Hostname: j.vnodeHostname,
425 - Labels: j.vnodeLabels,
439 + GUID: j.vnode.GUID,
440 + Hostname: j.vnode.Hostname,
441 + Labels: j.vnode.Labels,
442 })
443 j.vnodeCreated = true
444 }
445 }
446
431 - j.api.HOST(j.vnodeGUID)
447 + j.api.HOST(j.vnode.GUID)
448
433 - if !j.collectStatusChart.created {
449 + if !j.collectStatusChart.created || createChart {
450 j.collectStatusChart.ID = fmt.Sprintf("%s_%s_data_collection_status", cleanPluginName(j.pluginName), j.FullName())
451 j.createChart(j.collectStatusChart)
452 }
453
438 - if !j.collectDurationChart.created {
454 + if !j.collectDurationChart.created || createChart {
455 j.collectDurationChart.ID = fmt.Sprintf("%s_%s_data_collection_duration", cleanPluginName(j.pluginName), j.FullName())
456 j.createChart(j.collectDurationChart)
457 }
@@ -444,7 +460,7 @@ func (j *Job) processMetrics(metrics map[string]int64, startTime time.Time, sinc
460
461 var i, updated int
462 for _, chart := range *j.charts {
447 - if !chart.created {
463 + if !chart.created || createChart {
464 typeID := fmt.Sprintf("%s.%s", j.FullName(), chart.ID)
465 if len(typeID) >= NetdataChartIDMaxLength {
466 j.Warningf("chart 'type.id' length (%d) >= max allowed (%d), the chart is ignored (%s)",
src/go/plugin/go.d/agent/setup.go
+6 -7
@@ -161,20 +161,19 @@ func (a *Agent) buildDiscoveryConf(enabled module.Registry) discovery.Config {
161 }
162 }
163
164 -func (a *Agent) setupVnodeRegistry() *vnodes.Vnodes {
165 - a.Debugf("looking for 'vnodes/' in %v", a.VnodesConfigDir)
166 -
167 - if len(a.VnodesConfigDir) == 0 {
164 +func (a *Agent) setupVnodeRegistry() map[string]*vnodes.VirtualNode {
165 + a.Debugf("looking for 'vnodes/' in %v", a.ConfigDir)
166 + if len(a.ConfigDir) == 0 {
167 return nil
168 }
169
171 - dirPath, err := a.VnodesConfigDir.Find("vnodes/")
170 + dirPath, err := a.ConfigDir.Find("vnodes/")
171 if err != nil || dirPath == "" {
172 return nil
173 }
174
176 - reg := vnodes.New(dirPath)
177 - a.Infof("found '%s' (%d vhosts)", dirPath, reg.Len())
175 + reg := vnodes.Load(dirPath)
176 + a.Infof("found '%s' (%d vhosts)", dirPath, len(reg))
177
178 return reg
179 }
src/go/plugin/go.d/agent/vnodes/vnodes.go
+72 -45
@@ -3,67 +3,70 @@
3 package vnodes
4
5 import (
6 + "fmt"
7 "io"
8 "io/fs"
9 "log/slog"
10 + "maps"
11 "os"
12 "path/filepath"
11 -
12 - "github.com/netdata/netdata/go/plugins/logger"
13 + "strings"
14
15 "github.com/google/uuid"
16 "gopkg.in/yaml.v2"
17 +
18 + "github.com/netdata/netdata/go/plugins/logger"
19 )
20
18 -func New(confDir string) *Vnodes {
19 - vn := &Vnodes{
20 - Logger: logger.New().With(
21 - slog.String("component", "vnodes"),
22 - ),
21 +var log = logger.New().With(
22 + slog.String("component", "vnodes"),
23 +)
24
24 - confDir: confDir,
25 - vnodes: make(map[string]*VirtualNode),
26 - }
25 +func Load(dir string) map[string]*VirtualNode {
26 + return readConfDir(dir)
27 +}
28
28 - vn.readConfDir()
29 +type VirtualNode struct {
30 + Name string `yaml:"name" json:"name"`
31 + Hostname string `yaml:"hostname" json:"hostname"`
32 + GUID string `yaml:"guid" json:"guid"`
33 + Labels map[string]string `yaml:"labels,omitempty" json:"labels"`
34
30 - return vn
35 + Source string `yaml:"-" json:"-"`
36 + SourceType string `yaml:"-" json:"-"`
37 }
38
33 -type (
34 - Vnodes struct {
35 - *logger.Logger
36 -
37 - confDir string
38 - vnodes map[string]*VirtualNode
39 - }
40 - VirtualNode struct {
41 - GUID string `yaml:"guid" json:"guid"`
42 - Hostname string `yaml:"hostname" json:"hostname"`
43 - Labels map[string]string `yaml:"labels" json:"labels"`
39 +func (v *VirtualNode) Copy() *VirtualNode {
40 + if v == nil {
41 + return nil
42 }
45 -)
43
47 -func (vn *Vnodes) Lookup(key string) (*VirtualNode, bool) {
48 - v, ok := vn.vnodes[key]
49 - return v, ok
50 -}
44 + labels := make(map[string]string, len(v.Labels))
45 + maps.Copy(labels, v.Labels)
46
52 -func (vn *Vnodes) Len() int {
53 - return len(vn.vnodes)
47 + return &VirtualNode{
48 + Name: v.Name,
49 + Hostname: v.Hostname,
50 + GUID: v.GUID,
51 + Source: v.Source,
52 + SourceType: v.SourceType,
53 + Labels: labels,
54 + }
55 }
56
56 -func (vn *Vnodes) readConfDir() {
57 - _ = filepath.WalkDir(vn.confDir, func(path string, d fs.DirEntry, err error) error {
57 +func readConfDir(dir string) map[string]*VirtualNode {
58 + vnodes := make(map[string]*VirtualNode)
59 +
60 + _ = filepath.WalkDir(dir, func(path string, d fs.DirEntry, err error) error {
61 if err != nil {
59 - vn.Warning(err)
62 + log.Warning(err)
63 return nil
64 }
65
66 if d.Type()&os.ModeSymlink != 0 {
67 dst, err := os.Readlink(path)
68 if err != nil {
66 - vn.Warningf("failed to resolve symlink '%s': %v", path, err)
69 + log.Warningf("failed to resolve symlink '%s': %v", path, err)
70 return nil
71 }
72
@@ -73,52 +76,65 @@ func (vn *Vnodes) readConfDir() {
76
77 fi, err := os.Stat(dst)
78 if err != nil {
76 - vn.Warningf("failed to stat resolved path '%s': %v", dst, err)
79 + log.Warningf("failed to stat resolved path '%s': %v", dst, err)
80 return nil
81 }
82 if !fi.Mode().IsRegular() {
80 - vn.Debugf("'%s' is not a regular file, skipping it", dst)
83 + log.Debugf("'%s' is not a regular file, skipping it", dst)
84 return nil
85 }
86 path = dst
87 } else if !d.Type().IsRegular() {
85 - vn.Debugf("'%s' is not a regular file, skipping it", path)
88 + log.Debugf("'%s' is not a regular file, skipping it", path)
89 return nil
90 }
91
92 if !isConfigFile(path) {
90 - vn.Debugf("'%s' is not a config file (wrong extension), skipping it", path)
93 + log.Debugf("'%s' is not a config file (wrong extension), skipping it", path)
94 return nil
95 }
96
97 var cfg []VirtualNode
98
99 if err := loadConfigFile(&cfg, path); err != nil {
97 - vn.Warning(err)
100 + log.Warning(err)
101 return nil
102 }
103
104 for _, v := range cfg {
105 if v.Hostname == "" || v.GUID == "" {
103 - vn.Warningf("skipping virtual node '%+v': required fields are missing (%s)", v, path)
106 + log.Warningf("skipping virtual node '%+v': required fields are missing (%s)", v, path)
107 continue
108 }
109 if err := uuid.Validate(v.GUID); err != nil {
107 - vn.Warningf("skipping virtual node '%+v': invalid GUID: %v (%s)", v, err, path)
110 + log.Warningf("skipping virtual node '%+v': invalid GUID: %v (%s)", v, err, path)
111 continue
112 }
110 - if _, ok := vn.vnodes[v.Hostname]; ok {
111 - vn.Warningf("skipping virtual node '%+v': duplicate node (%s)", v, path)
113 + if _, ok := vnodes[v.Hostname]; ok {
114 + log.Warningf("skipping virtual node '%+v': duplicate node (%s)", v, path)
115 continue
116 }
117
118 v := v
116 - vn.Debugf("adding virtual node'%+v' (%s)", v, path)
117 - vn.vnodes[v.Hostname] = &v
119 +
120 + if v.Name == "" {
121 + v.Name = v.Hostname
122 + }
123 + v.Source = fmt.Sprintf("file=%s", path)
124 + if isStockConfig(path) {
125 + v.SourceType = "stock"
126 + } else {
127 + v.SourceType = "user"
128 + }
129 +
130 + log.Debugf("adding virtual node'%+v' (%s)", v, path)
131 + vnodes[v.Hostname] = &v
132 }
133
134 return nil
135 })
136 +
137 + return vnodes
138 }
139
140 func isConfigFile(path string) bool {
@@ -143,3 +159,14 @@ func loadConfigFile(conf any, path string) error {
159
160 return nil
161 }
162 +
163 +var (
164 + envNDStockConfigDir = os.Getenv("NETDATA_STOCK_CONFIG_DIR")
165 +)
166 +
167 +func isStockConfig(path string) bool {
168 + if envNDStockConfigDir == "" {
169 + return false
170 + }
171 + return strings.HasPrefix(path, envNDStockConfigDir)
172 +}
src/go/plugin/go.d/agent/vnodes/vnodes_test.go
+3 -16
@@ -8,20 +8,7 @@ import (
8 "github.com/stretchr/testify/assert"
9 )
10
11 -func TestNew(t *testing.T) {
12 - assert.NotNil(t, New("testdata"))
13 - assert.NotNil(t, New("not_exist"))
14 -}
15 -
16 -func TestVnodes_Lookup(t *testing.T) {
17 - req := New("testdata")
18 -
19 - _, ok := req.Lookup("first")
20 - assert.True(t, ok)
21 -
22 - _, ok = req.Lookup("second")
23 - assert.True(t, ok)
24 -
25 - _, ok = req.Lookup("third")
26 - assert.False(t, ok)
11 +func TestLoad(t *testing.T) {
12 + assert.NotNil(t, Load("testdata"))
13 + assert.NotNil(t, Load("not_exist"))
14 }
src/go/plugin/go.d/collector/snmp/testdata/config.json
+1
@@ -3,6 +3,7 @@
3 "hostname": "ok",
4 "create_vnode": true,
5 "vnode": {
6 + "name": "ok",
7 "guid": "ok",
8 "hostname": "ok",
9 "labels": {
src/go/plugin/go.d/collector/snmp/testdata/config.yaml
+1
@@ -2,6 +2,7 @@ update_every: 123
2 hostname: "ok"
3 create_vnode: yes
4 vnode:
5 + name: "ok"
6 guid: "ok"
7 hostname: "ok"
8 labels: