@cryptotaxi247 / netdata-1 / commits / 04545dcb5

chore(go.d): add shared dyncfg package (#21263)

Ilya Mashchenko committed Nov 4, 2025 at 18:50 UTC 04545dcb52612be795aab9d29fb7243cad6c584d
10 files changed +557 -402
src/go/plugin/go.d/agent/dyncfg/dyncfg.go new
+56
@@ -0,0 +1,56 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package dyncfg
4 +
5 +import (
6 + "strings"
7 +)
8 +
9 +// Status represents the state of a dyncfg entity
10 +type Status string
11 +
12 +const (
13 + StatusAccepted Status = "accepted"
14 + StatusRunning Status = "running"
15 + StatusFailed Status = "failed"
16 + StatusIncomplete Status = "incomplete"
17 + StatusDisabled Status = "disabled"
18 +)
19 +
20 +func (s Status) String() string {
21 + return string(s)
22 +}
23 +
24 +type ConfigType string
25 +
26 +const (
27 + ConfigTypeTemplate ConfigType = "template"
28 + ConfigTypeJob ConfigType = "job"
29 +)
30 +
31 +func (t ConfigType) String() string {
32 + return string(t)
33 +}
34 +
35 +type Command string
36 +
37 +const (
38 + CommandAdd Command = "add"
39 + CommandRemove Command = "remove"
40 + CommandGet Command = "get"
41 + CommandUpdate Command = "update"
42 + CommandRestart Command = "restart"
43 + CommandEnable Command = "enable"
44 + CommandDisable Command = "disable"
45 + CommandTest Command = "test"
46 + CommandSchema Command = "schema"
47 + CommandUserconfig Command = "userconfig"
48 +)
49 +
50 +func JoinCommands(commands ...Command) string {
51 + strs := make([]string, len(commands))
52 + for i, cmd := range commands {
53 + strs[i] = string(cmd)
54 + }
55 + return strings.Join(strs, " ")
56 +}
src/go/plugin/go.d/agent/dyncfg/responder.go new
+92
@@ -0,0 +1,92 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package dyncfg
4 +
5 +import (
6 + "encoding/json"
7 + "fmt"
8 + "strconv"
9 + "time"
10 +
11 + "github.com/netdata/netdata/go/plugins/pkg/netdataapi"
12 + "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/functions"
13 +)
14 +
15 +// Responder handles standardized responses for dyncfg operations
16 +type Responder struct {
17 + api *netdataapi.API
18 +}
19 +
20 +// NewResponder creates a new responder
21 +func NewResponder(api *netdataapi.API) *Responder {
22 + return &Responder{api: api}
23 +}
24 +
25 +// SendCodef sends a response with a specific code and message
26 +func (r *Responder) SendCodef(fn functions.Function, code int, message string, args ...any) {
27 + if fn.UID == "" {
28 + return
29 + }
30 +
31 + msg := message
32 + if len(args) > 0 {
33 + msg = fmt.Sprintf(message, args...)
34 + }
35 +
36 + response := struct {
37 + Status int `json:"status"`
38 + Message string `json:"message"`
39 + }{
40 + Status: code,
41 + Message: msg,
42 + }
43 +
44 + payload, _ := json.Marshal(response)
45 +
46 + r.api.FUNCRESULT(netdataapi.FunctionResult{
47 + UID: fn.UID,
48 + ContentType: "application/json",
49 + Payload: string(payload),
50 + Code: strconv.Itoa(code),
51 + ExpireTimestamp: strconv.FormatInt(time.Now().Unix(), 10),
52 + })
53 +}
54 +
55 +// SendJSON sends a JSON payload response
56 +func (r *Responder) SendJSON(fn functions.Function, payload string) {
57 + r.sendPayload(fn, payload, "application/json")
58 +}
59 +
60 +// SendYAML sends a YAML payload response
61 +func (r *Responder) SendYAML(fn functions.Function, payload string) {
62 + r.sendPayload(fn, payload, "application/yaml")
63 +}
64 +
65 +// sendPayload sends a response with a specific payload and content type
66 +func (r *Responder) sendPayload(fn functions.Function, payload, contentType string) {
67 + if fn.UID == "" {
68 + return
69 + }
70 +
71 + r.api.FUNCRESULT(netdataapi.FunctionResult{
72 + UID: fn.UID,
73 + ContentType: contentType,
74 + Payload: payload,
75 + Code: "200",
76 + ExpireTimestamp: strconv.FormatInt(time.Now().Unix(), 10),
77 + })
78 +}
79 +
80 +func (r *Responder) ConfigCreate(opts netdataapi.ConfigOpts) {
81 + r.api.CONFIGCREATE(opts)
82 +}
83 +
84 +// ConfigStatus sends a CONFIG STATUS command
85 +func (r *Responder) ConfigStatus(id string, status Status) {
86 + r.api.CONFIGSTATUS(id, status.String())
87 +}
88 +
89 +// ConfigDelete sends a CONFIG DELETE command
90 +func (r *Responder) ConfigDelete(id string) {
91 + r.api.CONFIGDELETE(id)
92 +}
src/go/plugin/go.d/agent/jobmgr/cache.go
+2 -1
@@ -7,6 +7,7 @@ import (
7 "sync"
8
9 "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/confgroup"
10 + "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/dyncfg"
11 "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/module"
12 )
13
@@ -57,7 +58,7 @@ type (
58 }
59 seenConfig struct {
60 cfg confgroup.Config
60 - status dyncfgStatus
61 + status dyncfg.Status
62 }
63
64 runningJobs struct {
src/go/plugin/go.d/agent/jobmgr/di.go
-8
@@ -3,7 +3,6 @@
3 package jobmgr
4
5 import (
6 - "github.com/netdata/netdata/go/plugins/pkg/netdataapi"
6 "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/functions"
7 "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/vnodes"
8 )
@@ -16,10 +15,3 @@ type FunctionRegistry interface {
15 RegisterPrefix(name, prefix string, fn func(functions.Function))
16 UnregisterPrefix(name string, prefix string)
17 }
19 -
20 -type dyncfgAPI interface {
21 - CONFIGCREATE(opts netdataapi.ConfigOpts)
22 - CONFIGDELETE(id string)
23 - CONFIGSTATUS(id, status string)
24 - FUNCRESULT(result netdataapi.FunctionResult)
25 -}
src/go/plugin/go.d/agent/jobmgr/dyncfg.go
+11 -31
@@ -10,47 +10,20 @@ import (
10
11 "gopkg.in/yaml.v2"
12
13 + "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/dyncfg"
14 "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/functions"
15 )
16
16 -type dyncfgStatus int
17 -
18 -const (
19 - _ dyncfgStatus = iota
20 - dyncfgAccepted
21 - dyncfgRunning
22 - dyncfgFailed
23 - dyncfgIncomplete
24 - dyncfgDisabled
25 -)
26 -
27 -func (s dyncfgStatus) String() string {
28 - switch s {
29 - case dyncfgAccepted:
30 - return "accepted"
31 - case dyncfgRunning:
32 - return "running"
33 - case dyncfgFailed:
34 - return "failed"
35 - case dyncfgIncomplete:
36 - return "incomplete"
37 - case dyncfgDisabled:
38 - return "disabled"
39 - default:
40 - return "unknown"
41 - }
42 -}
43 -
17 func (m *Manager) dyncfgConfig(fn functions.Function) {
18 if len(fn.Args) < 2 {
19 m.Warningf("dyncfg: %s: missing required arguments, want 3 got %d", fn.Name, len(fn.Args))
47 - m.dyncfgRespf(fn, 400, "Missing required arguments. Need at least 2, but got %d.", len(fn.Args))
20 + m.dyncfgApi.SendCodef(fn, 400, "Missing required arguments. Need at least 2, but got %d.", len(fn.Args))
21 return
22 }
23
24 select {
25 case <-m.ctx.Done():
53 - m.dyncfgRespf(fn, 503, "Job manager is shutting down.")
26 + m.dyncfgApi.SendCodef(fn, 503, "Job manager is shutting down.")
27 default:
28 }
29
@@ -62,7 +35,7 @@ func (m *Manager) dyncfgConfig(fn functions.Function) {
35 case strings.HasPrefix(id, m.dyncfgVnodePrefixValue()):
36 m.dyncfgVnodeExec(fn)
37 default:
65 - m.dyncfgRespf(fn, 503, "unknown function '%s' (%s).", fn.Name, id)
38 + m.dyncfgApi.SendCodef(fn, 503, "unknown function '%s' (%s).", fn.Name, id)
39 }
40 }
41
@@ -85,3 +58,10 @@ func getFnSourceValue(fn functions.Function, key string) string {
58 }
59 return ""
60 }
61 +
62 +func getDyncfgCommand(fn functions.Function) dyncfg.Command {
63 + if len(fn.Args) < 2 {
64 + return ""
65 + }
66 + return dyncfg.Command(strings.ToLower(fn.Args[1]))
67 +}
src/go/plugin/go.d/agent/jobmgr/dyncfg_collector.go
+211 -216
@@ -9,9 +9,7 @@ import (
9 "fmt"
10 "log/slog"
11 "slices"
12 - "strconv"
12 "strings"
14 - "time"
13 "unicode"
14
15 "gopkg.in/yaml.v2"
@@ -20,6 +18,7 @@ import (
18 "github.com/netdata/netdata/go/plugins/pkg/executable"
19 "github.com/netdata/netdata/go/plugins/pkg/netdataapi"
20 "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/confgroup"
21 + "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/dyncfg"
22 "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/functions"
23 "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/module"
24 )
@@ -41,100 +40,115 @@ func (m *Manager) dyncfgJobID(cfg confgroup.Config) string {
40 return fmt.Sprintf("%s%s:%s", m.dyncfgCollectorPrefixValue(), cfg.Module(), cfg.Name())
41 }
42
44 -func dyncfgModCmds() string {
45 - return "add schema enable disable test userconfig"
43 +func dyncfgCollectorModCmds() string {
44 + return dyncfg.JoinCommands(
45 + dyncfg.CommandAdd,
46 + dyncfg.CommandSchema,
47 + dyncfg.CommandEnable,
48 + dyncfg.CommandDisable,
49 + dyncfg.CommandTest,
50 + dyncfg.CommandUserconfig)
51 }
47 -func dyncfgJobCmds(cfg confgroup.Config) string {
48 - cmds := "schema get enable disable update restart test userconfig"
49 - if isDyncfg(cfg) {
50 - cmds += " remove"
52 +func dyncfgCollectorJobCmds(isDyncfgJob bool) string {
53 + cmds := []dyncfg.Command{
54 + dyncfg.CommandSchema,
55 + dyncfg.CommandGet,
56 + dyncfg.CommandEnable,
57 + dyncfg.CommandDisable,
58 + dyncfg.CommandUpdate,
59 + dyncfg.CommandRestart,
60 + dyncfg.CommandTest,
61 + dyncfg.CommandUserconfig,
62 }
52 - return cmds
63 + if isDyncfgJob {
64 + cmds = append(cmds, dyncfg.CommandRemove)
65 + }
66 + return dyncfg.JoinCommands(cmds...)
67 }
68
69 func (m *Manager) dyncfgCollectorModuleCreate(name string) {
56 - m.api.CONFIGCREATE(netdataapi.ConfigOpts{
70 + m.dyncfgApi.ConfigCreate(netdataapi.ConfigOpts{
71 ID: m.dyncfgModID(name),
58 - Status: dyncfgAccepted.String(),
59 - ConfigType: "template",
72 + Status: dyncfg.StatusAccepted.String(),
73 + ConfigType: dyncfg.ConfigTypeTemplate.String(),
74 Path: fmt.Sprintf(dyncfgCollectorPath, executable.Name),
75 SourceType: "internal",
76 Source: "internal",
63 - SupportedCommands: dyncfgModCmds(),
77 + SupportedCommands: dyncfgCollectorModCmds(),
78 })
79 }
80
67 -func (m *Manager) dyncfgCollectorJobCreate(cfg confgroup.Config, status dyncfgStatus) {
68 - m.api.CONFIGCREATE(netdataapi.ConfigOpts{
81 +func (m *Manager) dyncfgCollectorJobCreate(cfg confgroup.Config, status dyncfg.Status) {
82 + m.dyncfgApi.ConfigCreate(netdataapi.ConfigOpts{
83 ID: m.dyncfgJobID(cfg),
84 Status: status.String(),
71 - ConfigType: "job",
85 + ConfigType: dyncfg.ConfigTypeJob.String(),
86 Path: fmt.Sprintf(dyncfgCollectorPath, executable.Name),
87 SourceType: cfg.SourceType(),
88 Source: cfg.Source(),
75 - SupportedCommands: dyncfgJobCmds(cfg),
89 + SupportedCommands: dyncfgCollectorJobCmds(isDyncfg(cfg)),
90 })
91 }
92
93 func (m *Manager) dyncfgJobRemove(cfg confgroup.Config) {
80 - m.api.CONFIGDELETE(m.dyncfgJobID(cfg))
94 + m.dyncfgApi.ConfigDelete(m.dyncfgJobID(cfg))
95 }
96
83 -func (m *Manager) dyncfgJobStatus(cfg confgroup.Config, status dyncfgStatus) {
84 - m.api.CONFIGSTATUS(m.dyncfgJobID(cfg), status.String())
97 +func (m *Manager) dyncfgJobStatus(cfg confgroup.Config, status dyncfg.Status) {
98 + m.dyncfgApi.ConfigStatus(m.dyncfgJobID(cfg), status)
99 }
100
101 func (m *Manager) dyncfgCollectorExec(fn functions.Function) {
88 - action := strings.ToLower(fn.Args[1])
89 -
90 - switch action {
91 - case "userconfig":
102 + switch getDyncfgCommand(fn) {
103 + case dyncfg.CommandUserconfig:
104 m.dyncfgConfigUserconfig(fn)
105 return
94 - case "test":
106 + case dyncfg.CommandTest:
107 m.dyncfgConfigTest(fn)
108 return
97 - case "schema":
109 + case dyncfg.CommandSchema:
110 m.dyncfgConfigSchema(fn)
111 return
112 }
113
114 select {
115 case <-m.ctx.Done():
104 - m.dyncfgRespf(fn, 503, "Job manager is shutting down.")
116 + m.dyncfgApi.SendCodef(fn, 503, "Job manager is shutting down.")
117 case m.dyncfgCh <- fn:
118 }
119 }
120
121 func (m *Manager) dyncfgCollectorSeqExec(fn functions.Function) {
110 - action := strings.ToLower(fn.Args[1])
122 + cmd := getDyncfgCommand(fn)
123
112 - switch action {
113 - case "test":
124 + switch cmd {
125 + case dyncfg.CommandTest:
126 m.dyncfgConfigTest(fn)
115 - case "schema":
127 + case dyncfg.CommandSchema:
128 m.dyncfgConfigSchema(fn)
117 - case "get":
129 + case dyncfg.CommandGet:
130 m.dyncfgConfigGet(fn)
119 - case "restart":
131 + case dyncfg.CommandRestart:
132 m.dyncfgConfigRestart(fn)
121 - case "enable":
133 + case dyncfg.CommandEnable:
134 m.dyncfgConfigEnable(fn)
123 - case "disable":
135 + case dyncfg.CommandDisable:
136 m.dyncfgConfigDisable(fn)
125 - case "add":
137 + case dyncfg.CommandAdd:
138 m.dyncfgConfigAdd(fn)
127 - case "remove":
139 + case dyncfg.CommandRemove:
140 m.dyncfgConfigRemove(fn)
129 - case "update":
141 + case dyncfg.CommandUpdate:
142 m.dyncfgConfigUpdate(fn)
143 default:
132 - m.Warningf("dyncfg: function '%s' action '%s' not implemented", fn.Name, action)
133 - m.dyncfgRespf(fn, 501, "Function '%s' action '%s' is not implemented.", fn.Name, action)
144 + m.Warningf("dyncfg: function '%s' command '%s' not implemented", fn.Name, cmd)
145 + m.dyncfgApi.SendCodef(fn, 501, "Function '%s' command '%s' is not implemented.", fn.Name, cmd)
146 }
147 }
148
149 func (m *Manager) dyncfgConfigUserconfig(fn functions.Function) {
150 + cmd := getDyncfgCommand(fn)
151 +
152 id := fn.Args[0]
153 jn := "test"
154 if len(fn.Args) > 2 {
@@ -143,41 +157,41 @@ func (m *Manager) dyncfgConfigUserconfig(fn functions.Function) {
157
158 mn, ok := m.extractModuleName(id)
159 if !ok {
146 - m.Warningf("dyncfg: userconfig: could not extract module and job from id (%s)", id)
147 - m.dyncfgRespf(fn, 400,
148 - "Invalid ID format. Could not extract module and job name from ID. Provided ID: %s.", id)
160 + m.Warningf("dyncfg: %s: could not extract module and job from id (%s)", cmd, id)
161 + m.dyncfgApi.SendCodef(fn, 400, "Invalid ID format. Could not extract module and job name from ID. Provided ID: %s.", id)
162 return
163 }
164
165 creator, ok := m.Modules.Lookup(mn)
166 if !ok {
154 - m.Warningf("dyncfg: userconfig: module %s not found", mn)
155 - m.dyncfgRespf(fn, 404, "The specified module '%s' is not registered.", mn)
167 + m.Warningf("dyncfg: %s: module %s not found", cmd, mn)
168 + m.dyncfgApi.SendCodef(fn, 404, "The specified module '%s' is not registered.", mn)
169 return
170 }
171
172 if creator.Config == nil || creator.Config() == nil {
160 - m.Warningf("dyncfg: userconfig: module %s: configuration not found", mn)
161 - m.dyncfgRespf(fn, 500, "Module %s does not provide configuration.", mn)
173 + m.Warningf("dyncfg: %s: module %s: configuration not found", cmd, mn)
174 + m.dyncfgApi.SendCodef(fn, 500, "Module %s does not provide configuration.", mn)
175 return
176 }
177
178 bs, err := userConfigFromPayload(creator.Config(), jn, fn)
179 if err != nil {
167 - m.Warningf("dyncfg: userconfig: module %s: failed to create config from payload: %v", mn, err)
168 - m.dyncfgRespf(fn, 400, "Invalid configuration format. Failed to create configuration from payload: %v.", err)
180 + m.Warningf("dyncfg: %s: module %s: failed to create config from payload: %v", cmd, mn, err)
181 + m.dyncfgApi.SendCodef(fn, 400, "Invalid configuration format. Failed to create configuration from payload: %v.", err)
182 }
183
171 - m.dyncfgRespPayloadYAML(fn, string(bs))
184 + m.dyncfgApi.SendYAML(fn, string(bs))
185 }
186
187 func (m *Manager) dyncfgConfigTest(fn functions.Function) {
188 + cmd := getDyncfgCommand(fn)
189 +
190 id := fn.Args[0]
191 mn, ok := m.extractModuleName(id)
192 if !ok {
178 - m.Warningf("dyncfg: test: could not extract module and job from id (%s)", id)
179 - m.dyncfgRespf(fn, 400,
180 - "Invalid ID format. Could not extract module and job name from ID. Provided ID: %s.", id)
193 + m.Warningf("dyncfg: %s: could not extract module and job from id (%s)", cmd, id)
194 + m.dyncfgApi.SendCodef(fn, 400, "Invalid ID format. Could not extract module and job name from ID. Provided ID: %s.", id)
195 return
196 }
197
@@ -186,32 +200,32 @@ func (m *Manager) dyncfgConfigTest(fn functions.Function) {
200 jn = fn.Args[2]
201 }
202
189 - m.Infof("dyncfg: test: %s/%s job by user '%s'", mn, jn, getFnSourceValue(fn, "user"))
203 + m.Infof("dyncfg: %s: %s/%s job by user '%s'", cmd, mn, jn, getFnSourceValue(fn, "user"))
204
205 if err := validateJobName(jn); err != nil {
192 - m.Warningf("dyncfg: test: module %s: unacceptable job name '%s': %v", mn, jn, err)
193 - m.dyncfgRespf(fn, 400, "Unacceptable job name '%s': %v.", jn, err)
206 + m.Warningf("dyncfg: %s: module %s: unacceptable job name '%s': %v", cmd, mn, jn, err)
207 + m.dyncfgApi.SendCodef(fn, 400, "Unacceptable job name '%s': %v.", jn, err)
208 return
209 }
210
211 creator, ok := m.Modules.Lookup(mn)
212 if !ok {
199 - m.Warningf("dyncfg: test: module %s not found", mn)
200 - m.dyncfgRespf(fn, 404, "The specified module '%s' is not registered.", mn)
213 + m.Warningf("dyncfg: %s: module %s not found", cmd, mn)
214 + m.dyncfgApi.SendCodef(fn, 404, "The specified module '%s' is not registered.", mn)
215 return
216 }
217
218 cfg, err := configFromPayload(fn)
219 if err != nil {
206 - m.Warningf("dyncfg: test: module %s: failed to create config from payload: %v", mn, err)
207 - m.dyncfgRespf(fn, 400, "Invalid configuration format. Failed to create configuration from payload: %v.", err)
220 + m.Warningf("dyncfg: %s: module %s: failed to create config from payload: %v", cmd, mn, err)
221 + m.dyncfgApi.SendCodef(fn, 400, "Invalid configuration format. Failed to create configuration from payload: %v.", err)
222 return
223 }
224
225 if cfg.Vnode() != "" {
226 if _, ok := m.Vnodes[cfg.Vnode()]; !ok {
213 - m.Warningf("dyncfg: test: module %s: vnode %s not found", mn, cfg.Vnode())
214 - m.dyncfgRespf(fn, 400, "The specified vnode '%s' is not registered.", cfg.Vnode())
227 + m.Warningf("dyncfg: %s: module %s: vnode %s not found", cmd, mn, cfg.Vnode())
228 + m.dyncfgApi.SendCodef(fn, 400, "The specified vnode '%s' is not registered.", cfg.Vnode())
229 return
230 }
231 }
@@ -222,8 +236,8 @@ func (m *Manager) dyncfgConfigTest(fn functions.Function) {
236 job := creator.Create()
237
238 if err := applyConfig(cfg, job); err != nil {
225 - m.Warningf("dyncfg: test: module %s: failed to apply config: %v", mn, err)
226 - m.dyncfgRespf(fn, 400, "Invalid configuration. Failed to apply configuration: %v.", err)
239 + m.Warningf("dyncfg: %s: module %s: failed to apply config: %v", cmd, mn, err)
240 + m.dyncfgApi.SendCodef(fn, 400, "Invalid configuration. Failed to apply configuration: %v.", err)
241 return
242 }
243
@@ -235,126 +249,131 @@ func (m *Manager) dyncfgConfigTest(fn functions.Function) {
249 defer job.Cleanup(context.Background())
250
251 if err := job.Init(context.Background()); err != nil {
238 - m.dyncfgRespf(fn, 422, "Job initialization failed: %v", err)
252 + m.dyncfgApi.SendCodef(fn, 422, "Job initialization failed: %v", err)
253 return
254 }
255 if err := job.Check(context.Background()); err != nil {
242 - m.dyncfgRespf(fn, 422, "Job check failed: %v", err)
256 + m.dyncfgApi.SendCodef(fn, 422, "Job check failed: %v", err)
257 return
258 }
259
246 - m.dyncfgRespf(fn, 200, "")
260 + m.dyncfgApi.SendCodef(fn, 200, "")
261 }
262
263 func (m *Manager) dyncfgConfigSchema(fn functions.Function) {
264 + cmd := getDyncfgCommand(fn)
265 +
266 id := fn.Args[0]
267 mn, ok := m.extractModuleName(id)
268 if !ok {
253 - m.Warningf("dyncfg: schema: could not extract module from id (%s)", id)
254 - m.dyncfgRespf(fn, 400, "Invalid ID format. Could not extract module name from ID. Provided ID: %s.", id)
269 + m.Warningf("dyncfg: %s: could not extract module from id (%s)", cmd, id)
270 + m.dyncfgApi.SendCodef(fn, 400, "Invalid ID format. Could not extract module name from ID. Provided ID: %s.", id)
271 return
272 }
273
274 mod, ok := m.Modules.Lookup(mn)
275 if !ok {
260 - m.Warningf("dyncfg: schema: module %s not found", mn)
261 - m.dyncfgRespf(fn, 404, "The specified module '%s' is not registered.", mn)
276 + m.Warningf("dyncfg: %s: module %s not found", cmd, mn)
277 + m.dyncfgApi.SendCodef(fn, 404, "The specified module '%s' is not registered.", mn)
278 return
279 }
280
265 - m.Infof("dyncfg: schema: %s module by user '%s'", mn, getFnSourceValue(fn, "user"))
281 + m.Infof("dyncfg: %s: %s module by user '%s'", cmd, mn, getFnSourceValue(fn, "user"))
282
283 if mod.JobConfigSchema == "" {
284 m.Warningf("dyncfg: schema: module %s: schema not found", mn)
269 - m.dyncfgRespf(fn, 500, "Module %s configuration schema not found.", mn)
285 + m.dyncfgApi.SendCodef(fn, 500, "Module %s configuration schema not found.", mn)
286 return
287 }
288
273 - m.dyncfgRespPayloadJSON(fn, mod.JobConfigSchema)
289 + m.dyncfgApi.SendJSON(fn, mod.JobConfigSchema)
290 }
291
292 func (m *Manager) dyncfgConfigGet(fn functions.Function) {
293 + cmd := getDyncfgCommand(fn)
294 +
295 id := fn.Args[0]
296 mn, jn, ok := m.extractModuleJobName(id)
297 if !ok {
280 - m.Warningf("dyncfg: get: could not extract module and job from id (%s)", id)
281 - m.dyncfgRespf(fn, 400,
282 - "Invalid ID format. Could not extract module and job name from ID. Provided ID: %s.", id)
298 + m.Warningf("dyncfg: %s: could not extract module and job from id (%s)", cmd, id)
299 + m.dyncfgApi.SendCodef(fn, 400, "Invalid ID format. Could not extract module and job name from ID. Provided ID: %s.", id)
300 return
301 }
302
303 creator, ok := m.Modules.Lookup(mn)
304 if !ok {
288 - m.Warningf("dyncfg: get: module %s not found", mn)
289 - m.dyncfgRespf(fn, 404, "The specified module '%s' is not registered.", mn)
305 + m.Warningf("dyncfg: %s: module %s not found", cmd, mn)
306 + m.dyncfgApi.SendCodef(fn, 404, "The specified module '%s' is not registered.", mn)
307 return
308 }
309
293 - m.Infof("dyncfg: get: %s/%s job by user '%s'", mn, jn, getFnSourceValue(fn, "user"))
310 + m.Infof("dyncfg: %s: %s/%s job by user '%s'", cmd, mn, jn, getFnSourceValue(fn, "user"))
311
312 ecfg, ok := m.exposedConfigs.lookupByName(mn, jn)
313 if !ok {
297 - m.Warningf("dyncfg: get: module %s job %s not found", mn, jn)
298 - m.dyncfgRespf(fn, 404, "The specified module '%s' job '%s' is not registered.", mn, jn)
314 + m.Warningf("dyncfg: %s: module %s job %s not found", cmd, mn, jn)
315 + m.dyncfgApi.SendCodef(fn, 404, "The specified module '%s' job '%s' is not registered.", mn, jn)
316 return
317 }
318
319 mod := creator.Create()
320
321 if err := applyConfig(ecfg.cfg, mod); err != nil {
305 - m.Warningf("dyncfg: get: module %s job %s failed to apply config: %v", mn, jn, err)
306 - m.dyncfgRespf(fn, 400, "Invalid configuration. Failed to apply configuration: %v.", err)
322 + m.Warningf("dyncfg: %s: module %s job %s failed to apply config: %v", cmd, mn, jn, err)
323 + m.dyncfgApi.SendCodef(fn, 400, "Invalid configuration. Failed to apply configuration: %v.", err)
324 return
325 }
326
327 conf := mod.Configuration()
328 if conf == nil {
312 - m.Warningf("dyncfg: get: module %s: configuration not found", mn)
313 - m.dyncfgRespf(fn, 500, "Module %s does not provide configuration.", mn)
329 + m.Warningf("dyncfg: %s: module %s: configuration not found", cmd, mn)
330 + m.dyncfgApi.SendCodef(fn, 500, "Module %s does not provide configuration.", mn)
331 return
332 }
333
334 bs, err := json.Marshal(conf)
335 if err != nil {
319 - m.Warningf("dyncfg: get: module %s job %s failed to json marshal config: %v", mn, jn, err)
320 - m.dyncfgRespf(fn, 500, "Failed to convert configuration into JSON: %v.", err)
336 + m.Warningf("dyncfg: %s: module %s job %s failed to json marshal config: %v", cmd, mn, jn, err)
337 + m.dyncfgApi.SendCodef(fn, 500, "Failed to convert configuration into JSON: %v.", err)
338 return
339 }
340
324 - m.dyncfgRespPayloadJSON(fn, string(bs))
341 + m.dyncfgApi.SendJSON(fn, string(bs))
342 }
343
344 func (m *Manager) dyncfgConfigRestart(fn functions.Function) {
345 + cmd := getDyncfgCommand(fn)
346 +
347 id := fn.Args[0]
348 mn, jn, ok := m.extractModuleJobName(id)
349 if !ok {
331 - m.Warningf("dyncfg: restart: could not extract module from id (%s)", id)
332 - m.dyncfgRespf(fn, 400, "Invalid ID format. Could not extract module name from ID. Provided ID: %s.", id)
350 + m.Warningf("dyncfg: %s: could not extract module from id (%s)", cmd, id)
351 + m.dyncfgApi.SendCodef(fn, 400, "Invalid ID format. Could not extract module name from ID. Provided ID: %s.", id)
352 return
353 }
354
355 ecfg, ok := m.exposedConfigs.lookupByName(mn, jn)
356 if !ok {
338 - m.Warningf("dyncfg: restart: module %s job %s not found", mn, jn)
339 - m.dyncfgRespf(fn, 404, "The specified module '%s' job '%s' is not registered.", mn, jn)
357 + m.Warningf("dyncfg: %s: module %s job %s not found", cmd, mn, jn)
358 + m.dyncfgApi.SendCodef(fn, 404, "The specified module '%s' job '%s' is not registered.", mn, jn)
359 return
360 }
361
362 job, err := m.createCollectorJob(ecfg.cfg)
363 if err != nil {
345 - m.Warningf("dyncfg: restart: module %s job %s: failed to apply config: %v", mn, jn, err)
346 - m.dyncfgRespf(fn, 400, "Invalid configuration. Failed to apply configuration: %v.", err)
364 + m.Warningf("dyncfg: %s: module %s job %s: failed to apply config: %v", cmd, mn, jn, err)
365 + m.dyncfgApi.SendCodef(fn, 400, "Invalid configuration. Failed to apply configuration: %v.", err)
366 m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
367 return
368 }
369
370 switch ecfg.status {
352 - case dyncfgAccepted, dyncfgDisabled:
353 - m.Warningf("dyncfg: restart: module %s job %s: restarting not allowed in '%s' state", mn, jn, ecfg.status)
354 - m.dyncfgRespf(fn, 405, "Restarting data collection job is not allowed in '%s' state.", ecfg.status)
371 + case dyncfg.StatusAccepted, dyncfg.StatusDisabled:
372 + m.Warningf("dyncfg: %s: module %s job %s: restarting not allowed in '%s' state", cmd, mn, jn, ecfg.status)
373 + m.dyncfgApi.SendCodef(fn, 405, "Restarting data collection job is not allowed in '%s' state.", ecfg.status)
374 m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
375 return
357 - case dyncfgRunning:
376 + case dyncfg.StatusRunning:
377 m.fileStatus.remove(ecfg.cfg)
378 m.stopRunningJob(ecfg.cfg.FullName())
379 default:
@@ -362,40 +381,43 @@ func (m *Manager) dyncfgConfigRestart(fn functions.Function) {
381
382 m.retryingTasks.remove(ecfg.cfg)
383
365 - m.Infof("dyncfg: restart: %s/%s job by user '%s'", mn, jn, getFnSourceValue(fn, "user"))
384 + m.Infof("dyncfg: %s: %s/%s job by user '%s'", cmd, mn, jn, getFnSourceValue(fn, "user"))
385
386 if err := job.AutoDetection(); err != nil {
387 job.Cleanup()
369 - ecfg.status = dyncfgFailed
370 - m.dyncfgRespf(fn, 422, "Job restart failed: %v", err)
388 + ecfg.status = dyncfg.StatusFailed
389 + m.dyncfgApi.SendCodef(fn, 422, "Job restart failed: %v", err)
390 m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
391 m.runRetryTask(ecfg, job)
392 return
393 }
394
376 - ecfg.status = dyncfgRunning
395 + ecfg.status = dyncfg.StatusRunning
396
397 if isDyncfg(ecfg.cfg) {
398 m.fileStatus.add(ecfg.cfg, ecfg.status.String())
399 }
400 m.startRunningJob(job)
382 - m.dyncfgRespf(fn, 200, "")
401 +
402 + m.dyncfgApi.SendCodef(fn, 200, "")
403 m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
404 }
405
406 func (m *Manager) dyncfgConfigEnable(fn functions.Function) {
407 + cmd := getDyncfgCommand(fn)
408 +
409 id := fn.Args[0]
410 mn, jn, ok := m.extractModuleJobName(id)
411 if !ok {
390 - m.Warningf("dyncfg: enable: could not extract module and job from id (%s)", id)
391 - m.dyncfgRespf(fn, 400, "Invalid ID format. Could not extract module and job name from ID. Provided ID: %s.", id)
412 + m.Warningf("dyncfg: %s: could not extract module and job from id (%s)", cmd, id)
413 + m.dyncfgApi.SendCodef(fn, 400, "Invalid ID format. Could not extract module and job name from ID. Provided ID: %s.", id)
414 return
415 }
416
417 ecfg, ok := m.exposedConfigs.lookupByName(mn, jn)
418 if !ok {
397 - m.Warningf("dyncfg: enable: module %s job %s not found", mn, jn)
398 - m.dyncfgRespf(fn, 404, "The specified module '%s' job '%s' is not registered.", mn, jn)
419 + m.Warningf("dyncfg: %s: module %s job %s not found", cmd, mn, jn)
420 + m.dyncfgApi.SendCodef(fn, 404, "The specified module '%s' job '%s' is not registered.", mn, jn)
421 return
422 }
423
@@ -404,38 +426,38 @@ func (m *Manager) dyncfgConfigEnable(fn functions.Function) {
426 }
427
428 switch ecfg.status {
407 - case dyncfgAccepted, dyncfgDisabled, dyncfgFailed:
408 - case dyncfgRunning:
429 + case dyncfg.StatusAccepted, dyncfg.StatusDisabled, dyncfg.StatusFailed:
430 + case dyncfg.StatusRunning:
431 // non-dyncfg update triggers enable/disable
410 - m.dyncfgRespf(fn, 200, "")
432 + m.dyncfgApi.SendCodef(fn, 200, "")
433 m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
434 return
435 default:
414 - m.Warningf("dyncfg: enable: module %s job %s: enabling not allowed in %s state", mn, jn, ecfg.status)
415 - m.dyncfgRespf(fn, 405, "Enabling data collection job is not allowed in '%s' state.", ecfg.status)
436 + m.Warningf("dyncfg: %s: module %s job %s: enabling not allowed in %s state", cmd, mn, jn, ecfg.status)
437 + m.dyncfgApi.SendCodef(fn, 405, "Enabling data collection job is not allowed in '%s' state.", ecfg.status)
438 m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
439 return
440 }
441
442 job, err := m.createCollectorJob(ecfg.cfg)
443 if err != nil {
422 - ecfg.status = dyncfgFailed
423 - m.Warningf("dyncfg: enable: module %s job %s: failed to apply config: %v", mn, jn, err)
424 - m.dyncfgRespf(fn, 400, "Invalid configuration. Failed to apply configuration: %v.", err)
444 + ecfg.status = dyncfg.StatusFailed
445 + m.Warningf("dyncfg: %s: module %s job %s: failed to apply config: %v", cmd, mn, jn, err)
446 + m.dyncfgApi.SendCodef(fn, 400, "Invalid configuration. Failed to apply configuration: %v.", err)
447 m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
448 return
449 }
450
429 - if ecfg.status == dyncfgDisabled {
430 - m.Infof("dyncfg: enable: %s/%s job by user '%s'", mn, jn, getFnSourceValue(fn, "user"))
451 + if ecfg.status == dyncfg.StatusDisabled {
452 + m.Infof("dyncfg: %s: %s/%s job by user '%s'", cmd, mn, jn, getFnSourceValue(fn, "user"))
453 }
454
455 m.retryingTasks.remove(ecfg.cfg)
456
457 if err := job.AutoDetection(); err != nil {
458 job.Cleanup()
437 - ecfg.status = dyncfgFailed
438 - m.dyncfgRespf(fn, 200, "Job enable failed: %v.", err)
459 + ecfg.status = dyncfg.StatusFailed
460 + m.dyncfgApi.SendCodef(fn, 200, "Job enable failed: %v.", err)
461
462 if isStock(ecfg.cfg) {
463 m.exposedConfigs.remove(ecfg.cfg)
@@ -448,31 +470,33 @@ func (m *Manager) dyncfgConfigEnable(fn functions.Function) {
470 return
471 }
472
451 - ecfg.status = dyncfgRunning
473 + ecfg.status = dyncfg.StatusRunning
474
475 if isDyncfg(ecfg.cfg) {
476 m.fileStatus.add(ecfg.cfg, ecfg.status.String())
477 }
478
479 m.startRunningJob(job)
458 - m.dyncfgRespf(fn, 200, "")
459 - m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
480
481 + m.dyncfgApi.SendCodef(fn, 200, "")
482 + m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
483 }
484
485 func (m *Manager) dyncfgConfigDisable(fn functions.Function) {
486 + cmd := getDyncfgCommand(fn)
487 +
488 id := fn.Args[0]
489 mn, jn, ok := m.extractModuleJobName(id)
490 if !ok {
467 - m.Warningf("dyncfg: disable: could not extract module from id (%s)", id)
468 - m.dyncfgRespf(fn, 400, "Invalid ID format. Could not extract module name from ID. Provided ID: %s.", id)
491 + m.Warningf("dyncfg: %s: could not extract module from id (%s)", cmd, id)
492 + m.dyncfgApi.SendCodef(fn, 400, "Invalid ID format. Could not extract module name from ID. Provided ID: %s.", id)
493 return
494 }
495
496 ecfg, ok := m.exposedConfigs.lookupByName(mn, jn)
497 if !ok {
474 - m.Warningf("dyncfg: disable: module %s job %s not found", mn, jn)
475 - m.dyncfgRespf(fn, 404, "The specified module '%s' job '%s' is not registered.", mn, jn)
498 + m.Warningf("dyncfg: %s: module %s job %s not found", cmd, mn, jn)
499 + m.dyncfgApi.SendCodef(fn, 404, "The specified module '%s' job '%s' is not registered.", mn, jn)
500 return
501 }
502
@@ -481,11 +505,11 @@ func (m *Manager) dyncfgConfigDisable(fn functions.Function) {
505 }
506
507 switch ecfg.status {
484 - case dyncfgDisabled:
485 - m.dyncfgRespf(fn, 200, "")
508 + case dyncfg.StatusDisabled:
509 + m.dyncfgApi.SendCodef(fn, 200, "")
510 m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
511 return
488 - case dyncfgRunning:
512 + case dyncfg.StatusRunning:
513 m.stopRunningJob(ecfg.cfg.FullName())
514 if isDyncfg(ecfg.cfg) {
515 m.fileStatus.remove(ecfg.cfg)
@@ -495,17 +519,19 @@ func (m *Manager) dyncfgConfigDisable(fn functions.Function) {
519
520 m.retryingTasks.remove(ecfg.cfg)
521
498 - m.Infof("dyncfg: disable: %s/%s job by user '%s'", mn, jn, getFnSourceValue(fn, "user"))
522 + m.Infof("dyncfg: %s: %s/%s job by user '%s'", cmd, mn, jn, getFnSourceValue(fn, "user"))
523
500 - ecfg.status = dyncfgDisabled
501 - m.dyncfgRespf(fn, 200, "")
524 + ecfg.status = dyncfg.StatusDisabled
525 + m.dyncfgApi.SendCodef(fn, 200, "")
526 m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
527 }
528
529 func (m *Manager) dyncfgConfigAdd(fn functions.Function) {
530 + cmd := getDyncfgCommand(fn)
531 +
532 if len(fn.Args) < 3 {
507 - m.Warningf("dyncfg: add: missing required arguments, want 3 got %d", len(fn.Args))
508 - m.dyncfgRespf(fn, 400, "Missing required arguments. Need at least 3, but got %d.", len(fn.Args))
533 + m.Warningf("dyncfg: %s: missing required arguments, want 3 got %d", cmd, len(fn.Args))
534 + m.dyncfgApi.SendCodef(fn, 400, "Missing required arguments. Need at least 3, but got %d.", len(fn.Args))
535 return
536 }
537
@@ -513,39 +539,39 @@ func (m *Manager) dyncfgConfigAdd(fn functions.Function) {
539 jn := fn.Args[2]
540 mn, ok := m.extractModuleName(id)
541 if !ok {
516 - m.Warningf("dyncfg: add: could not extract module from id (%s)", id)
517 - m.dyncfgRespf(fn, 400, "Invalid ID format. Could not extract module name from ID. Provided ID: %s.", id)
542 + m.Warningf("dyncfg: %s: could not extract module from id (%s)", cmd, id)
543 + m.dyncfgApi.SendCodef(fn, 400, "Invalid ID format. Could not extract module name from ID. Provided ID: %s.", id)
544 return
545 }
546
547 if len(fn.Payload) == 0 {
522 - m.Warningf("dyncfg: add: module %s job %s missing configuration payload.", mn, jn)
523 - m.dyncfgRespf(fn, 400, "Missing configuration payload.")
548 + m.Warningf("dyncfg: %s: module %s job %s missing configuration payload.", cmd, mn, jn)
549 + m.dyncfgApi.SendCodef(fn, 400, "Missing configuration payload.")
550 return
551 }
552
553 if err := validateJobName(jn); err != nil {
528 - m.Warningf("dyncfg: add: module %s: unacceptable job name '%s': %v", mn, jn, err)
529 - m.dyncfgRespf(fn, 400, "Unacceptable job name '%s': %v.", jn, err)
554 + m.Warningf("dyncfg: %s: module %s: unacceptable job name '%s': %v", cmd, mn, jn, err)
555 + m.dyncfgApi.SendCodef(fn, 400, "Unacceptable job name '%s': %v.", jn, err)
556 return
557 }
558
559 cfg, err := configFromPayload(fn)
560 if err != nil {
535 - m.Warningf("dyncfg: add: module %s job %s: failed to create config from payload: %v", mn, jn, err)
536 - m.dyncfgRespf(fn, 400, "Invalid configuration format. Failed to create configuration from payload: %v.", err)
561 + m.Warningf("dyncfg: %s: module %s job %s: failed to create config from payload: %v", cmd, mn, jn, err)
562 + m.dyncfgApi.SendCodef(fn, 400, "Invalid configuration format. Failed to create configuration from payload: %v.", err)
563 return
564 }
565
566 m.dyncfgSetConfigMeta(cfg, mn, jn, fn)
567
568 if _, err := m.createCollectorJob(cfg); err != nil {
543 - m.Warningf("dyncfg: add: module %s job %s: failed to apply config: %v", mn, jn, err)
544 - m.dyncfgRespf(fn, 400, "Invalid configuration. Failed to apply configuration: %v.", err)
569 + m.Warningf("dyncfg: %s: module %s job %s: failed to apply config: %v", cmd, mn, jn, err)
570 + m.dyncfgApi.SendCodef(fn, 400, "Invalid configuration. Failed to apply configuration: %v.", err)
571 return
572 }
573
548 - m.Infof("dyncfg: add: %s/%s job by user '%s'", mn, jn, getFnSourceValue(fn, "user"))
574 + m.Infof("dyncfg: %s: %s/%s job by user '%s'", cmd, mn, jn, getFnSourceValue(fn, "user"))
575
576 if ecfg, ok := m.exposedConfigs.lookup(cfg); ok {
577 if scfg, ok := m.seenConfigs.lookup(ecfg.cfg); ok && isDyncfg(scfg.cfg) {
@@ -556,38 +582,40 @@ func (m *Manager) dyncfgConfigAdd(fn functions.Function) {
582 m.stopRunningJob(ecfg.cfg.FullName())
583 }
584
559 - scfg := &seenConfig{cfg: cfg, status: dyncfgAccepted}
585 + scfg := &seenConfig{cfg: cfg, status: dyncfg.StatusAccepted}
586 ecfg := scfg
587 m.seenConfigs.add(scfg)
588 m.exposedConfigs.add(ecfg)
589
564 - m.dyncfgRespf(fn, 202, "")
590 + m.dyncfgApi.SendCodef(fn, 202, "")
591 m.dyncfgCollectorJobCreate(ecfg.cfg, ecfg.status)
592 }
593
594 func (m *Manager) dyncfgConfigRemove(fn functions.Function) {
595 + cmd := getDyncfgCommand(fn)
596 +
597 id := fn.Args[0]
598 mn, jn, ok := m.extractModuleJobName(id)
599 if !ok {
572 - m.Warningf("dyncfg: remove: could not extract module and job from id (%s)", id)
573 - m.dyncfgRespf(fn, 400, "Invalid ID format. Could not extract module and job name from ID. Provided ID: %s.", id)
600 + m.Warningf("dyncfg: %s: could not extract module and job from id (%s)", cmd, id)
601 + m.dyncfgApi.SendCodef(fn, 400, "Invalid ID format. Could not extract module and job name from ID. Provided ID: %s.", id)
602 return
603 }
604
605 ecfg, ok := m.exposedConfigs.lookupByName(mn, jn)
606 if !ok {
579 - m.Warningf("dyncfg: remove: module %s job %s not found", mn, jn)
580 - m.dyncfgRespf(fn, 404, "The specified module '%s' job '%s' is not registered.", mn, jn)
607 + m.Warningf("dyncfg: %s: module %s job %s not found", cmd, mn, jn)
608 + m.dyncfgApi.SendCodef(fn, 404, "The specified module '%s' job '%s' is not registered.", mn, jn)
609 return
610 }
611
612 if !isDyncfg(ecfg.cfg) {
585 - m.Warningf("dyncfg: remove: module %s job %s: can not remove jobs of type %s", mn, jn, ecfg.cfg.SourceType())
586 - m.dyncfgRespf(fn, 405, "Removing jobs of type '%s' is not supported. Only 'dyncfg' jobs can be removed.", ecfg.cfg.SourceType())
613 + m.Warningf("dyncfg: %s: module %s job %s: can not remove jobs of type %s", cmd, mn, jn, ecfg.cfg.SourceType())
614 + m.dyncfgApi.SendCodef(fn, 405, "Removing jobs of type '%s' is not supported. Only 'dyncfg' jobs can be removed.", ecfg.cfg.SourceType())
615 return
616 }
617
590 - m.Infof("dyncfg: remove: %s/%s job by user '%s'", mn, jn, getFnSourceValue(fn, "user"))
618 + m.Infof("dyncfg: %s: %s/%s job by user '%s'", cmd, mn, jn, getFnSourceValue(fn, "user"))
619
620 m.retryingTasks.remove(ecfg.cfg)
621 m.seenConfigs.remove(ecfg.cfg)
@@ -595,63 +623,65 @@ func (m *Manager) dyncfgConfigRemove(fn functions.Function) {
623 m.stopRunningJob(ecfg.cfg.FullName())
624 m.fileStatus.remove(ecfg.cfg)
625
598 - m.dyncfgRespf(fn, 200, "")
626 + m.dyncfgApi.SendCodef(fn, 200, "")
627 m.dyncfgJobRemove(ecfg.cfg)
628 }
629
630 func (m *Manager) dyncfgConfigUpdate(fn functions.Function) {
631 + cmd := getDyncfgCommand(fn)
632 +
633 id := fn.Args[0]
634 mn, jn, ok := m.extractModuleJobName(id)
635 if !ok {
606 - m.Warningf("dyncfg: update: could not extract module from id (%s)", id)
607 - m.dyncfgRespf(fn, 400, "Invalid ID format. Could not extract module name from ID. Provided ID: %s.", id)
636 + m.Warningf("dyncfg: %s: could not extract module from id (%s)", cmd, id)
637 + m.dyncfgApi.SendCodef(fn, 400, "Invalid ID format. Could not extract module name from ID. Provided ID: %s.", id)
638 return
639 }
640
641 ecfg, ok := m.exposedConfigs.lookupByName(mn, jn)
642 if !ok {
613 - m.Warningf("dyncfg: update: module %s job %s not found", mn, jn)
614 - m.dyncfgRespf(fn, 404, "The specified module '%s' job '%s' is not registered.", mn, jn)
643 + m.Warningf("dyncfg: %s: module %s job %s not found", cmd, mn, jn)
644 + m.dyncfgApi.SendCodef(fn, 404, "The specified module '%s' job '%s' is not registered.", mn, jn)
645 return
646 }
647
648 cfg, err := configFromPayload(fn)
649 if err != nil {
620 - m.Warningf("dyncfg: update: module %s: failed to create config from payload: %v", mn, err)
621 - m.dyncfgRespf(fn, 400, "Invalid configuration format. Failed to create configuration from payload: %v.", err)
650 + m.Warningf("dyncfg: %s: module %s: failed to create config from payload: %v", cmd, mn, err)
651 + m.dyncfgApi.SendCodef(fn, 400, "Invalid configuration format. Failed to create configuration from payload: %v.", err)
652 m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
653 return
654 }
655
656 m.dyncfgSetConfigMeta(cfg, mn, jn, fn)
657
628 - if ecfg.status == dyncfgRunning && ecfg.cfg.UID() == cfg.UID() {
629 - m.dyncfgRespf(fn, 200, "")
658 + if ecfg.status == dyncfg.StatusRunning && ecfg.cfg.UID() == cfg.UID() {
659 + m.dyncfgApi.SendCodef(fn, 200, "")
660 m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
661 return
662 }
663
664 job, err := m.createCollectorJob(cfg)
665 if err != nil {
636 - m.Warningf("dyncfg: update: module %s job %s: failed to apply config: %v", mn, jn, err)
637 - m.dyncfgRespf(fn, 400, "Invalid configuration. Failed to apply configuration: %v.", err)
666 + m.Warningf("dyncfg: %s: module %s job %s: failed to apply config: %v", cmd, mn, jn, err)
667 + m.dyncfgApi.SendCodef(fn, 400, "Invalid configuration. Failed to apply configuration: %v.", err)
668 m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
669 return
670 }
671
642 - if ecfg.status == dyncfgAccepted {
643 - m.Warningf("dyncfg: update: module %s job %s: updating not allowed in %s", mn, jn, ecfg.status)
644 - m.dyncfgRespf(fn, 403, "Updating data collection job is not allowed in '%s' state.", ecfg.status)
672 + if ecfg.status == dyncfg.StatusAccepted {
673 + m.Warningf("dyncfg: %s: module %s job %s: updating not allowed in %s", cmd, mn, jn, ecfg.status)
674 + m.dyncfgApi.SendCodef(fn, 403, "Updating data collection job is not allowed in '%s' state.", ecfg.status)
675 m.dyncfgJobStatus(ecfg.cfg, ecfg.status)
676 return
677 }
678
649 - m.Infof("dyncfg: update: %s/%s job by user '%s'", mn, jn, getFnSourceValue(fn, "user"))
679 + m.Infof("dyncfg: %s: %s/%s job by user '%s'", cmd, mn, jn, getFnSourceValue(fn, "user"))
680
681 m.exposedConfigs.remove(ecfg.cfg)
682 m.stopRunningJob(ecfg.cfg.FullName())
683
654 - scfg := &seenConfig{cfg: cfg, status: dyncfgAccepted}
684 + scfg := &seenConfig{cfg: cfg, status: dyncfg.StatusAccepted}
685 m.seenConfigs.add(scfg)
686 m.exposedConfigs.add(scfg)
687
@@ -662,9 +692,10 @@ func (m *Manager) dyncfgConfigUpdate(fn functions.Function) {
692 defer m.dyncfgCollectorJobCreate(scfg.cfg, scfg.status)
693 }
694
665 - if ecfg.status == dyncfgDisabled {
666 - scfg.status = dyncfgDisabled
667 - m.dyncfgRespf(fn, 200, "")
695 + if ecfg.status == dyncfg.StatusDisabled {
696 + scfg.status = dyncfg.StatusDisabled
697 +
698 + m.dyncfgApi.SendCodef(fn, 200, "")
699 m.dyncfgJobStatus(cfg, scfg.status)
700 return
701 }
@@ -673,16 +704,18 @@ func (m *Manager) dyncfgConfigUpdate(fn functions.Function) {
704
705 if err := job.AutoDetection(); err != nil {
706 job.Cleanup()
676 - scfg.status = dyncfgFailed
677 - m.dyncfgRespf(fn, 200, "Job update failed: %v", err)
707 + scfg.status = dyncfg.StatusFailed
708 +
709 + m.dyncfgApi.SendCodef(fn, 200, "Job update failed: %v", err)
710 m.dyncfgJobStatus(scfg.cfg, scfg.status)
711 m.runRetryTask(scfg, job)
712 return
713 }
714
683 - scfg.status = dyncfgRunning
715 + scfg.status = dyncfg.StatusRunning
716 m.startRunningJob(job)
685 - m.dyncfgRespf(fn, 200, "")
717 +
718 + m.dyncfgApi.SendCodef(fn, 200, "")
719 m.dyncfgJobStatus(scfg.cfg, scfg.status)
720 }
721
@@ -697,44 +730,6 @@ func (m *Manager) dyncfgSetConfigMeta(cfg confgroup.Config, module, name string,
730 }
731 }
732
700 -func (m *Manager) dyncfgRespPayloadJSON(fn functions.Function, payload string) {
701 - m.dyncfgRespPayload(fn, payload, "application/json")
702 -}
703 -
704 -func (m *Manager) dyncfgRespPayloadYAML(fn functions.Function, payload string) {
705 - m.dyncfgRespPayload(fn, payload, "application/yaml")
706 -}
707 -
708 -func (m *Manager) dyncfgRespPayload(fn functions.Function, payload string, contentType string) {
709 - m.api.FUNCRESULT(netdataapi.FunctionResult{
710 - UID: fn.UID,
711 - ContentType: contentType,
712 - Payload: payload,
713 - Code: "200",
714 - ExpireTimestamp: strconv.FormatInt(time.Now().Unix(), 10),
715 - })
716 -}
717 -
718 -func (m *Manager) dyncfgRespf(fn functions.Function, code int, msgf string, a ...any) {
719 - if fn.UID == "" {
720 - return
721 - }
722 - bs, _ := json.Marshal(struct {
723 - Status int `json:"status"`
724 - Message string `json:"message"`
725 - }{
726 - Status: code,
727 - Message: fmt.Sprintf(msgf, a...),
728 - })
729 - m.api.FUNCRESULT(netdataapi.FunctionResult{
730 - UID: fn.UID,
731 - ContentType: "application/json",
732 - Payload: string(bs),
733 - Code: strconv.Itoa(code),
734 - ExpireTimestamp: strconv.FormatInt(time.Now().Unix(), 10),
735 - })
736 -}
737 -
733 func (m *Manager) runRetryTask(ecfg *seenConfig, job *module.Job) {
734 if !job.RetryAutoDetection() {
735 return
src/go/plugin/go.d/agent/jobmgr/dyncfg_vnode.go
+110 -76
@@ -14,6 +14,7 @@ import (
14 "github.com/netdata/netdata/go/plugins/pkg/executable"
15 "github.com/netdata/netdata/go/plugins/pkg/netdataapi"
16 "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/confgroup"
17 + "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/dyncfg"
18 "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/functions"
19 "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/module"
20 "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/vnodes"
@@ -28,133 +29,155 @@ func (m *Manager) dyncfgVnodePrefixValue() string {
29 return fmt.Sprintf(dyncfgVnodeIDf, executable.Name)
30 }
31
32 +func dyncfgVnodeModCmds() string {
33 + return dyncfg.JoinCommands(
34 + dyncfg.CommandAdd,
35 + dyncfg.CommandSchema,
36 + dyncfg.CommandUserconfig,
37 + dyncfg.CommandTest,
38 + )
39 +}
40 +func dyncfgVnodeJobCmds(isDyncfgJob bool) string {
41 + cmds := []dyncfg.Command{
42 + dyncfg.CommandUserconfig,
43 + dyncfg.CommandSchema,
44 + dyncfg.CommandGet,
45 + dyncfg.CommandUpdate,
46 + dyncfg.CommandTest,
47 + }
48 + if isDyncfgJob {
49 + cmds = append(cmds, dyncfg.CommandRemove)
50 + }
51 + return dyncfg.JoinCommands(cmds...)
52 +}
53 +
54 func (m *Manager) dyncfgVnodeModuleCreate() {
32 - m.api.CONFIGCREATE(netdataapi.ConfigOpts{
55 + m.dyncfgApi.ConfigCreate(netdataapi.ConfigOpts{
56 ID: m.dyncfgVnodePrefixValue(),
34 - Status: dyncfgAccepted.String(),
35 - ConfigType: "template",
57 + Status: dyncfg.StatusAccepted.String(),
58 + ConfigType: dyncfg.ConfigTypeTemplate.String(),
59 Path: fmt.Sprintf(dyncfgVnodePath, executable.Name),
60 SourceType: "internal",
61 Source: "internal",
39 - SupportedCommands: "add schema userconfig test",
62 + SupportedCommands: dyncfgVnodeModCmds(),
63 })
64 }
65
43 -func (m *Manager) dyncfgVnodeJobCreate(cfg *vnodes.VirtualNode, status dyncfgStatus) {
44 - cmds := "userconfig schema get update test"
45 - if cfg.SourceType == confgroup.TypeDyncfg {
46 - cmds += " remove"
47 - }
48 - m.api.CONFIGCREATE(netdataapi.ConfigOpts{
66 +func (m *Manager) dyncfgVnodeJobCreate(cfg *vnodes.VirtualNode, status dyncfg.Status) {
67 + m.dyncfgApi.ConfigCreate(netdataapi.ConfigOpts{
68 ID: fmt.Sprintf("%s:%s", m.dyncfgVnodePrefixValue(), cfg.Name),
69 Status: status.String(),
51 - ConfigType: "job",
70 + ConfigType: dyncfg.ConfigTypeJob.String(),
71 Path: fmt.Sprintf(dyncfgVnodePath, executable.Name),
72 SourceType: cfg.SourceType,
73 Source: cfg.Source,
55 - SupportedCommands: cmds,
74 + SupportedCommands: dyncfgVnodeJobCmds(cfg.SourceType == confgroup.TypeDyncfg),
75 })
76 }
77
78 func (m *Manager) dyncfgVnodeExec(fn functions.Function) {
60 - action := strings.ToLower(fn.Args[1])
79 + cmd := dyncfg.Command(strings.ToLower(fn.Args[1]))
80
62 - switch action {
63 - case "userconfig":
81 + switch cmd {
82 + case dyncfg.CommandUserconfig:
83 m.dyncfgVnodeUserconfig(fn)
84 return
66 - case "schema":
67 - m.dyncfgRespPayloadJSON(fn, vnodes.ConfigSchema)
85 + case dyncfg.CommandSchema:
86 + m.dyncfgApi.SendJSON(fn, vnodes.ConfigSchema)
87 return
88 }
89
90 select {
91 case <-m.ctx.Done():
73 - m.dyncfgRespf(fn, 503, "Job manager is shutting down.")
92 + m.dyncfgApi.SendCodef(fn, 503, "Job manager is shutting down.")
93 case m.dyncfgCh <- fn:
94 }
95 }
96
97 func (m *Manager) dyncfgVnodeSeqExec(fn functions.Function) {
79 - action := strings.ToLower(fn.Args[1])
98 + cmd := dyncfg.Command(strings.ToLower(fn.Args[1]))
99
81 - switch action {
82 - case "test":
100 + switch cmd {
101 + case dyncfg.CommandTest:
102 m.dyncfgVnodeTest(fn)
84 - case "get":
103 + case dyncfg.CommandGet:
104 m.dyncfgVnodeGet(fn)
86 - case "add":
105 + case dyncfg.CommandAdd:
106 m.dyncfgVnodeAdd(fn)
88 - case "update":
107 + case dyncfg.CommandUpdate:
108 m.dyncfgVnodeUpdate(fn)
90 - case "remove":
109 + case dyncfg.CommandRemove:
110 m.dyncfgVnodeRemove(fn)
111 default:
93 - m.Warningf("dyncfg: function '%s' action '%s' not implemented", fn.Name, action)
94 - m.dyncfgRespf(fn, 501, "Function '%s' action '%s' is not implemented.", fn.Name, action)
112 + m.Warningf("dyncfg: function '%s' command '%s' not implemented", fn.Name, cmd)
113 + m.dyncfgApi.SendCodef(fn, 501, "Function '%s' command '%s' is not implemented.", fn.Name, cmd)
114 }
115 }
116
117 func (m *Manager) dyncfgVnodeGet(fn functions.Function) {
118 + cmd := dyncfg.CommandGet
119 +
120 id := fn.Args[0]
121 name := strings.TrimPrefix(id, m.dyncfgVnodePrefixValue()+":")
122
123 cfg, ok := m.Vnodes[name]
124 if !ok {
104 - m.Warningf("dyncfg: get: vnode %s not found", name)
105 - m.dyncfgRespf(fn, 404, "The specified vnode '%s' is not registered.", name)
125 + m.Warningf("dyncfg: %s: vnode %s not found", cmd, name)
126 + m.dyncfgApi.SendCodef(fn, 404, "The specified vnode '%s' is not registered.", name)
127 return
128 }
129
130 bs, err := json.Marshal(cfg)
131 if err != nil {
111 - m.Warningf("dyncfg: get: vnode job %s failed to json marshal config: %v", name, err)
112 - m.dyncfgRespf(fn, 500, "Failed to convert configuration into JSON: %v.", err)
132 + m.Warningf("dyncfg: %s: vnode job %s failed to json marshal config: %v", cmd, name, err)
133 + m.dyncfgApi.SendCodef(fn, 500, "Failed to convert configuration into JSON: %v.", err)
134 return
135 }
136
116 - m.dyncfgRespPayloadJSON(fn, string(bs))
137 + m.dyncfgApi.SendJSON(fn, string(bs))
138 }
139
140 func (m *Manager) dyncfgVnodeAdd(fn functions.Function) {
141 + cmd := dyncfg.CommandAdd
142 +
143 if len(fn.Args) < 3 {
121 - m.Warningf("dyncfg: add: missing required arguments, want 3 got %d", len(fn.Args))
122 - m.dyncfgRespf(fn, 400, "Missing required arguments. Need at least 3, but got %d.", len(fn.Args))
144 + m.Warningf("dyncfg: %s: missing required arguments, want 3 got %d", cmd, len(fn.Args))
145 + m.dyncfgApi.SendCodef(fn, 400, "Missing required arguments. Need at least 3, but got %d.", len(fn.Args))
146 return
147 }
148
149 name := fn.Args[2]
150
151 if len(fn.Payload) == 0 {
129 - m.Warningf("dyncfg: add: vnode job %s missing configuration payload.", name)
130 - m.dyncfgRespf(fn, 400, "Missing configuration payload.")
152 + m.Warningf("dyncfg: %s: vnode job %s missing configuration payload.", cmd, name)
153 + m.dyncfgApi.SendCodef(fn, 400, "Missing configuration payload.")
154 return
155 }
156
157 cfg, err := vnodeConfigFromPayload(fn)
158 if err != nil {
136 - m.Warningf("dyncfg: add: vnode job %s: failed to create config from payload: %v", name, err)
137 - m.dyncfgRespf(fn, 400, "Failed to create configuration from payload. Invalid configuration format: %v.", err)
159 + m.Warningf("dyncfg: %s: vnode job %s: failed to create config from payload: %v", cmd, name, err)
160 + m.dyncfgApi.SendCodef(fn, 400, "Failed to create configuration from payload. Invalid configuration format: %v.", err)
161 return
162 }
163
164 if err := uuid.Validate(cfg.GUID); err != nil {
142 - m.Warningf("dyncfg: add: vnode job %s: invalid guid: %v", name, err)
143 - m.dyncfgRespf(fn, 400, "Failed to create configuration from payload. Invalid guid format: %v.", err)
165 + m.Warningf("dyncfg: %s: vnode job %s: invalid guid: %v", cmd, name, err)
166 + m.dyncfgApi.SendCodef(fn, 400, "Failed to create configuration from payload. Invalid guid format: %v.", err)
167 return
168 }
169
170 dyncfgUpdateVnodeConfig(cfg, name, fn)
171
172 if err := m.verifyVnodeUnique(cfg); err != nil {
150 - m.Warningf("dyncfg: add: vnode job %s: %v", name, err)
151 - m.dyncfgRespf(fn, 400, "Failed to create configuration from payload: %v.", err)
173 + m.Warningf("dyncfg: %s: vnode job %s: %v", cmd, name, err)
174 + m.dyncfgApi.SendCodef(fn, 400, "Failed to create configuration from payload: %v.", err)
175 return
176 }
177
178 if orig, ok := m.Vnodes[name]; ok && orig.Equal(cfg) {
156 - m.dyncfgRespf(fn, 202, "")
157 - m.dyncfgVnodeJobCreate(cfg, dyncfgRunning)
179 + m.dyncfgApi.SendCodef(fn, 202, "")
180 + m.dyncfgVnodeJobCreate(cfg, dyncfg.StatusRunning)
181 return
182 }
183
@@ -165,41 +188,47 @@ func (m *Manager) dyncfgVnodeAdd(fn functions.Function) {
188 job.UpdateVnode(cfg)
189 }
190 })
168 - m.dyncfgRespf(fn, 202, "")
169 - m.dyncfgVnodeJobCreate(cfg, dyncfgRunning)
191 +
192 + m.dyncfgApi.SendCodef(fn, 202, "")
193 + m.dyncfgVnodeJobCreate(cfg, dyncfg.StatusRunning)
194 }
195
196 func (m *Manager) dyncfgVnodeRemove(fn functions.Function) {
197 + cmd := dyncfg.CommandRemove
198 +
199 id := fn.Args[0]
200 name := strings.TrimPrefix(id, m.dyncfgVnodePrefixValue()+":")
201
202 vnode, ok := m.Vnodes[name]
203 if !ok {
178 - m.Warningf("dyncfg: remove: vnode %s not found", name)
179 - m.dyncfgRespf(fn, 404, "The specified vnode '%s' is not registered.", name)
204 + m.Warningf("dyncfg: %s: vnode %s not found", cmd, name)
205 + m.dyncfgApi.SendCodef(fn, 404, "The specified vnode '%s' is not registered.", name)
206 return
207 }
208 if vnode.SourceType != confgroup.TypeDyncfg {
183 - m.Warningf("dyncfg: remove: module vnode %s: can not remove vnode of type %s", vnode.Name, vnode.SourceType)
184 - m.dyncfgRespf(fn, 405, "Removing vnode of type '%s' is not supported. Only 'dyncfg' vnodes can be removed.", vnode.SourceType)
209 + m.Warningf("dyncfg: %s: module vnode %s: can not remove vnode of type %s", cmd, vnode.Name, vnode.SourceType)
210 + m.dyncfgApi.SendCodef(fn, 405, "Removing vnode of type '%s' is not supported. Only 'dyncfg' vnodes can be removed.", vnode.SourceType)
211 return
212 }
213
214 if s := m.dyncfgVnodeAffectedJobs(vnode.Name); s != "" {
189 - m.Warningf("dyncfg: remove: vnode %s has running jobs (%s)", name, s)
190 - m.dyncfgRespf(fn, 404, "The specified vnode '%s' has running jobs (%s).", name, s)
215 + m.Warningf("dyncfg: %s: vnode %s has running jobs (%s)", cmd, name, s)
216 + m.dyncfgApi.SendCodef(fn, 404, "The specified vnode '%s' has running jobs (%s).", name, s)
217 return
218 }
219
220 delete(m.Vnodes, name)
195 - m.api.CONFIGDELETE(id)
196 - m.dyncfgRespf(fn, 200, "")
221 +
222 + m.dyncfgApi.ConfigDelete(id)
223 + m.dyncfgApi.SendCodef(fn, 200, "")
224 }
225
226 func (m *Manager) dyncfgVnodeTest(fn functions.Function) {
227 + cmd := dyncfg.CommandTest
228 +
229 if len(fn.Args) < 3 {
201 - m.Warningf("dyncfg: test: missing required arguments, want 3 got %d", len(fn.Args))
202 - m.dyncfgRespf(fn, 400, "Missing required arguments. Need at least 3, but got %d.", len(fn.Args))
230 + m.Warningf("dyncfg: %s: missing required arguments, want 3 got %d", cmd, len(fn.Args))
231 + m.dyncfgApi.SendCodef(fn, 400, "Missing required arguments. Need at least 3, but got %d.", len(fn.Args))
232 return
233 }
234
@@ -207,60 +236,62 @@ func (m *Manager) dyncfgVnodeTest(fn functions.Function) {
236
237 cfg, err := vnodeConfigFromPayload(fn)
238 if err != nil {
210 - m.Warningf("dyncfg: test: vnode: failed to create config from payload: %v", err)
211 - m.dyncfgRespf(fn, 400, "Invalid configuration format. Failed to create configuration from payload: %v.", err)
239 + m.Warningf("dyncfg: %s: vnode: failed to create config from payload: %v", cmd, err)
240 + m.dyncfgApi.SendCodef(fn, 400, "Invalid configuration format. Failed to create configuration from payload: %v.", err)
241 return
242 }
243
244 if err := uuid.Validate(cfg.GUID); err != nil {
216 - m.Warningf("dyncfg: test: vnode job %s: invalid guid: %v", name, err)
217 - m.dyncfgRespf(fn, 400, "Failed to create configuration from payload. Invalid guid format: %v.", err)
245 + m.Warningf("dyncfg: %s: vnode job %s: invalid guid: %v", cmd, name, err)
246 + m.dyncfgApi.SendCodef(fn, 400, "Failed to create configuration from payload. Invalid guid format: %v.", err)
247 return
248 }
249
250 dyncfgUpdateVnodeConfig(cfg, name, fn)
251
252 if err := m.verifyVnodeUnique(cfg); err != nil {
224 - m.Warningf("dyncfg: test: vnode job %s: %v", name, err)
225 - m.dyncfgRespf(fn, 400, "Failed to create configuration from payload: %v.", err)
253 + m.Warningf("dyncfg: %s: vnode job %s: %v", cmd, name, err)
254 + m.dyncfgApi.SendCodef(fn, 400, "Failed to create configuration from payload: %v.", err)
255 return
256 }
257
258 if s := m.dyncfgVnodeAffectedJobs(cfg.Name); s != "" {
230 - m.dyncfgRespf(fn, 202, "Updated configuration will affect: %s.", s)
259 + m.dyncfgApi.SendCodef(fn, 202, "Updated configuration will affect: %s.", s)
260 } else {
232 - m.dyncfgRespf(fn, 202, "No jobs will be affected by this change.")
261 + m.dyncfgApi.SendCodef(fn, 202, "No jobs will be affected by this change.")
262 }
263 }
264
265 func (m *Manager) dyncfgVnodeUpdate(fn functions.Function) {
266 + cmd := dyncfg.CommandUpdate
267 +
268 id := fn.Args[0]
269 name := strings.TrimPrefix(id, m.dyncfgVnodePrefixValue()+":")
270
271 orig, ok := m.Vnodes[name]
272 if !ok {
242 - m.Warningf("dyncfg: remove: vnode %s not found", name)
243 - m.dyncfgRespf(fn, 404, "The specified vnode '%s' is not registered.", name)
273 + m.Warningf("dyncfg: %s: vnode %s not found", cmd, name)
274 + m.dyncfgApi.SendCodef(fn, 404, "The specified vnode '%s' is not registered.", name)
275 return
276 }
277
278 cfg, err := vnodeConfigFromPayload(fn)
279 if err != nil {
249 - m.Warningf("dyncfg: remove: vnode: failed to create config from payload: %v", err)
250 - m.dyncfgRespf(fn, 400, "Invalid configuration format. Failed to create configuration from payload: %v.", err)
280 + m.Warningf("dyncfg: %s: vnode: failed to create config from payload: %v", cmd, err)
281 + m.dyncfgApi.SendCodef(fn, 400, "Invalid configuration format. Failed to create configuration from payload: %v.", err)
282 return
283 }
284
285 if err := uuid.Validate(cfg.GUID); err != nil {
255 - m.Warningf("dyncfg: update: vnode job %s: invalid guid: %v", name, err)
256 - m.dyncfgRespf(fn, 400, "Failed to create configuration from payload. Invalid guid format: %v.", err)
286 + m.Warningf("dyncfg: %s: vnode job %s: invalid guid: %v", cmd, name, err)
287 + m.dyncfgApi.SendCodef(fn, 400, "Failed to create configuration from payload. Invalid guid format: %v.", err)
288 return
289 }
290
291 dyncfgUpdateVnodeConfig(cfg, name, fn)
292
293 if orig.Equal(cfg) {
263 - m.dyncfgRespf(fn, 202, "")
294 + m.dyncfgApi.SendCodef(fn, 202, "")
295 return
296 }
297
@@ -271,19 +302,22 @@ func (m *Manager) dyncfgVnodeUpdate(fn functions.Function) {
302 job.UpdateVnode(cfg)
303 }
304 })
274 - m.dyncfgRespf(fn, 202, "")
275 - m.dyncfgVnodeJobCreate(cfg, dyncfgRunning)
305 +
306 + m.dyncfgApi.SendCodef(fn, 202, "")
307 + m.dyncfgVnodeJobCreate(cfg, dyncfg.StatusRunning)
308 }
309
310 func (m *Manager) dyncfgVnodeUserconfig(fn functions.Function) {
311 + cmd := dyncfg.CommandUserconfig
312 +
313 bs, err := vnodeUserconfigFromPayload(fn)
314 if err != nil {
281 - m.Warningf("dyncfg: userconfig: vnode: failed to create config from payload: %v", err)
282 - m.dyncfgRespf(fn, 400, "Invalid configuration format. Failed to create configuration from payload: %v.", err)
315 + m.Warningf("dyncfg: %s: vnode: failed to create config from payload: %v", cmd, err)
316 + m.dyncfgApi.SendCodef(fn, 400, "Invalid configuration format. Failed to create configuration from payload: %v.", err)
317 return
318 }
319
286 - m.dyncfgRespPayloadYAML(fn, string(bs))
320 + m.dyncfgApi.SendYAML(fn, string(bs))
321 }
322
323 func (m *Manager) dyncfgVnodeAffectedJobs(vnode string) string {
src/go/plugin/go.d/agent/jobmgr/manager.go
+18 -15
@@ -19,6 +19,7 @@ import (
19 "github.com/netdata/netdata/go/plugins/pkg/safewriter"
20 "github.com/netdata/netdata/go/plugins/pkg/ticker"
21 "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/confgroup"
22 + "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/dyncfg"
23 "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/functions"
24 "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/module"
25 "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/vnodes"
@@ -45,11 +46,11 @@ func New() *Manager {
46 runningJobs: newRunningJobsCache(),
47 retryingTasks: newRetryingTasksCache(),
48
48 - started: make(chan struct{}),
49 - api: netdataapi.New(safewriter.Stdout),
50 - addCh: make(chan confgroup.Config),
51 - rmCh: make(chan confgroup.Config),
52 - dyncfgCh: make(chan functions.Function),
49 + started: make(chan struct{}),
50 + addCh: make(chan confgroup.Config),
51 + rmCh: make(chan confgroup.Config),
52 + dyncfgCh: make(chan functions.Function),
53 + dyncfgApi: dyncfg.NewResponder(netdataapi.New(safewriter.Stdout)),
54 }
55
56 return mgr
@@ -80,14 +81,16 @@ type Manager struct {
81 retryingTasks *retryingTasks
82 runningJobs *runningJobs
83
83 - ctx context.Context
84 - started chan struct{}
85 - api dyncfgAPI
84 + ctx context.Context
85 + started chan struct{}
86 + //api dyncfgAPI
87 addCh chan confgroup.Config
88 rmCh chan confgroup.Config
89 dyncfgCh chan functions.Function
90
91 waitCfgOnOff string // block processing of discovered configs until "enable"/"disable" is received from Netdata
92 +
93 + dyncfgApi *dyncfg.Responder
94 }
95
96 func (m *Manager) Run(ctx context.Context, in chan []*confgroup.Group) {
@@ -101,7 +104,7 @@ func (m *Manager) Run(ctx context.Context, in chan []*confgroup.Group) {
104 m.dyncfgVnodeModuleCreate()
105
106 for _, cfg := range m.Vnodes {
104 - m.dyncfgVnodeJobCreate(cfg, dyncfgRunning)
107 + m.dyncfgVnodeJobCreate(cfg, dyncfg.StatusRunning)
108 }
109
110 for name := range m.Modules {
@@ -175,7 +178,7 @@ func (m *Manager) run() {
178 case strings.HasPrefix(id, m.dyncfgVnodePrefixValue()):
179 m.dyncfgVnodeSeqExec(fn)
180 default:
178 - m.dyncfgRespf(fn, 503, "unknown function '%s' (%s).", fn.Name, id)
181 + m.dyncfgApi.SendCodef(fn, 503, "unknown function '%s' (%s).", fn.Name, id)
182 }
183 }
184 }
@@ -197,19 +200,19 @@ func (m *Manager) addConfig(cfg confgroup.Config) {
200
201 ecfg, ok := m.exposedConfigs.lookup(cfg)
202 if !ok {
200 - scfg.status = dyncfgAccepted
203 + scfg.status = dyncfg.StatusAccepted
204 ecfg = scfg
205 m.exposedConfigs.add(ecfg)
206 } else {
207 sp, ep := scfg.cfg.SourceTypePriority(), ecfg.cfg.SourceTypePriority()
205 - if ep > sp || (ep == sp && ecfg.status == dyncfgRunning) {
208 + if ep > sp || (ep == sp && ecfg.status == dyncfg.StatusRunning) {
209 return
210 }
208 - if ecfg.status == dyncfgRunning {
211 + if ecfg.status == dyncfg.StatusRunning {
212 m.stopRunningJob(ecfg.cfg.FullName())
213 m.fileStatus.remove(ecfg.cfg)
214 }
212 - scfg.status = dyncfgAccepted
215 + scfg.status = dyncfg.StatusAccepted
216 m.exposedConfigs.add(scfg) // replace existing exposed
217 ecfg = scfg
218 }
@@ -241,7 +244,7 @@ func (m *Manager) removeConfig(cfg confgroup.Config) {
244 m.stopRunningJob(cfg.FullName())
245 m.fileStatus.remove(cfg)
246
244 - if !isStock(cfg) || ecfg.status == dyncfgRunning {
247 + if !isStock(cfg) || ecfg.status == dyncfg.StatusRunning {
248 m.dyncfgJobRemove(cfg)
249 }
250 }
src/go/plugin/go.d/agent/jobmgr/manager_test.go
+55 -54
@@ -8,6 +8,7 @@ import (
8 "testing"
9
10 "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/confgroup"
11 + "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/dyncfg"
12 "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/functions"
13 )
14
@@ -61,7 +62,7 @@ CONFIG test:collector:success:name delete
62 },
63 wantDiscovered: []confgroup.Config{cfg},
64 wantSeen: []seenConfig{
64 - {cfg: cfg, status: dyncfgFailed},
65 + {cfg: cfg, status: dyncfg.StatusFailed},
66 },
67 wantExposed: nil,
68 wantRunning: nil,
@@ -267,13 +268,13 @@ CONFIG test:collector:fail:name delete
268 discCfg,
269 },
270 wantSeen: []seenConfig{
270 - {cfg: stockCfg, status: dyncfgFailed},
271 - {cfg: discCfg, status: dyncfgFailed},
272 - {cfg: userCfg, status: dyncfgFailed},
271 + {cfg: stockCfg, status: dyncfg.StatusFailed},
272 + {cfg: discCfg, status: dyncfg.StatusFailed},
273 + {cfg: userCfg, status: dyncfg.StatusFailed},
274 },
275 wantExposed: []seenConfig{
275 - {cfg: discCfg, status: dyncfgFailed},
276 - {cfg: userCfg, status: dyncfgFailed},
276 + {cfg: discCfg, status: dyncfg.StatusFailed},
277 + {cfg: userCfg, status: dyncfg.StatusFailed},
278 },
279 wantRunning: nil,
280 wantDyncfg: `
@@ -336,12 +337,12 @@ CONFIG test:collector:fail:user status failed
337 discCfg,
338 },
339 wantSeen: []seenConfig{
339 - {cfg: stockCfg, status: dyncfgFailed},
340 - {cfg: discCfg, status: dyncfgFailed},
341 - {cfg: userCfg, status: dyncfgFailed},
340 + {cfg: stockCfg, status: dyncfg.StatusFailed},
341 + {cfg: discCfg, status: dyncfg.StatusFailed},
342 + {cfg: userCfg, status: dyncfg.StatusFailed},
343 },
344 wantExposed: []seenConfig{
344 - {cfg: userCfg, status: dyncfgFailed},
345 + {cfg: userCfg, status: dyncfg.StatusFailed},
346 },
347 wantRunning: nil,
348 wantDyncfg: `
@@ -459,12 +460,12 @@ CONFIG test:collector:fail:name delete
460 discCfg,
461 },
462 wantSeen: []seenConfig{
462 - {cfg: userCfg, status: dyncfgFailed},
463 + {cfg: userCfg, status: dyncfg.StatusFailed},
464 {cfg: discCfg},
465 {cfg: stockCfg},
466 },
467 wantExposed: []seenConfig{
467 - {cfg: userCfg, status: dyncfgFailed},
468 + {cfg: userCfg, status: dyncfg.StatusFailed},
469 },
470 wantRunning: nil,
471 wantDyncfg: `
@@ -578,10 +579,10 @@ FUNCTION_RESULT_END
579 },
580 wantDiscovered: nil,
581 wantSeen: []seenConfig{
581 - {cfg: cfg, status: dyncfgAccepted},
582 + {cfg: cfg, status: dyncfg.StatusAccepted},
583 },
584 wantExposed: []seenConfig{
584 - {cfg: cfg, status: dyncfgAccepted},
585 + {cfg: cfg, status: dyncfg.StatusAccepted},
586 },
587 wantRunning: nil,
588 wantDyncfg: `
@@ -692,10 +693,10 @@ func TestManager_Run_Dyncfg_Add(t *testing.T) {
693 },
694 wantDiscovered: nil,
695 wantSeen: []seenConfig{
695 - {cfg: cfg, status: dyncfgAccepted},
696 + {cfg: cfg, status: dyncfg.StatusAccepted},
697 },
698 wantExposed: []seenConfig{
698 - {cfg: cfg, status: dyncfgAccepted},
699 + {cfg: cfg, status: dyncfg.StatusAccepted},
700 },
701 wantRunning: nil,
702 wantDyncfg: `
@@ -724,10 +725,10 @@ CONFIG test:collector:success:test create accepted job /collectors/test/Jobs dyn
725 },
726 wantDiscovered: nil,
727 wantSeen: []seenConfig{
727 - {cfg: cfg, status: dyncfgAccepted},
728 + {cfg: cfg, status: dyncfg.StatusAccepted},
729 },
730 wantExposed: []seenConfig{
730 - {cfg: cfg, status: dyncfgAccepted},
731 + {cfg: cfg, status: dyncfg.StatusAccepted},
732 },
733 wantRunning: nil,
734 wantDyncfg: `
@@ -762,10 +763,10 @@ CONFIG test:collector:fail:test create accepted job /collectors/test/Jobs dyncfg
763 },
764 wantDiscovered: nil,
765 wantSeen: []seenConfig{
765 - {cfg: cfg, status: dyncfgAccepted},
766 + {cfg: cfg, status: dyncfg.StatusAccepted},
767 },
768 wantExposed: []seenConfig{
768 - {cfg: cfg, status: dyncfgAccepted},
769 + {cfg: cfg, status: dyncfg.StatusAccepted},
770 },
771 wantRunning: nil,
772 wantDyncfg: `
@@ -842,10 +843,10 @@ FUNCTION_RESULT_END
843 },
844 wantDiscovered: nil,
845 wantSeen: []seenConfig{
845 - {cfg: cfg, status: dyncfgRunning},
846 + {cfg: cfg, status: dyncfg.StatusRunning},
847 },
848 wantExposed: []seenConfig{
848 - {cfg: cfg, status: dyncfgRunning},
849 + {cfg: cfg, status: dyncfg.StatusRunning},
850 },
851 wantRunning: []string{cfg.FullName()},
852 wantDyncfg: `
@@ -888,10 +889,10 @@ CONFIG test:collector:success:test status running
889 },
890 wantDiscovered: nil,
891 wantSeen: []seenConfig{
891 - {cfg: cfg, status: dyncfgRunning},
892 + {cfg: cfg, status: dyncfg.StatusRunning},
893 },
894 wantExposed: []seenConfig{
894 - {cfg: cfg, status: dyncfgRunning},
895 + {cfg: cfg, status: dyncfg.StatusRunning},
896 },
897 wantRunning: []string{cfg.FullName()},
898 wantDyncfg: `
@@ -936,10 +937,10 @@ CONFIG test:collector:success:test status running
937 },
938 wantDiscovered: nil,
939 wantSeen: []seenConfig{
939 - {cfg: cfg, status: dyncfgFailed},
940 + {cfg: cfg, status: dyncfg.StatusFailed},
941 },
942 wantExposed: []seenConfig{
942 - {cfg: cfg, status: dyncfgFailed},
943 + {cfg: cfg, status: dyncfg.StatusFailed},
944 },
945 wantRunning: nil,
946 wantDyncfg: `
@@ -982,10 +983,10 @@ CONFIG test:collector:fail:test status failed
983 },
984 wantDiscovered: nil,
985 wantSeen: []seenConfig{
985 - {cfg: cfg, status: dyncfgFailed},
986 + {cfg: cfg, status: dyncfg.StatusFailed},
987 },
988 wantExposed: []seenConfig{
988 - {cfg: cfg, status: dyncfgFailed},
989 + {cfg: cfg, status: dyncfg.StatusFailed},
990 },
991 wantRunning: nil,
992 wantDyncfg: `
@@ -1068,10 +1069,10 @@ FUNCTION_RESULT_END
1069 },
1070 wantDiscovered: nil,
1071 wantSeen: []seenConfig{
1071 - {cfg: cfg, status: dyncfgDisabled},
1072 + {cfg: cfg, status: dyncfg.StatusDisabled},
1073 },
1074 wantExposed: []seenConfig{
1074 - {cfg: cfg, status: dyncfgDisabled},
1075 + {cfg: cfg, status: dyncfg.StatusDisabled},
1076 },
1077 wantRunning: nil,
1078 wantDyncfg: `
@@ -1114,10 +1115,10 @@ CONFIG test:collector:success:test status disabled
1115 },
1116 wantDiscovered: nil,
1117 wantSeen: []seenConfig{
1117 - {cfg: cfg, status: dyncfgDisabled},
1118 + {cfg: cfg, status: dyncfg.StatusDisabled},
1119 },
1120 wantExposed: []seenConfig{
1120 - {cfg: cfg, status: dyncfgDisabled},
1121 + {cfg: cfg, status: dyncfg.StatusDisabled},
1122 },
1123 wantRunning: nil,
1124 wantDyncfg: `
@@ -1162,10 +1163,10 @@ CONFIG test:collector:success:test status disabled
1163 },
1164 wantDiscovered: nil,
1165 wantSeen: []seenConfig{
1165 - {cfg: cfg, status: dyncfgDisabled},
1166 + {cfg: cfg, status: dyncfg.StatusDisabled},
1167 },
1168 wantExposed: []seenConfig{
1168 - {cfg: cfg, status: dyncfgDisabled},
1169 + {cfg: cfg, status: dyncfg.StatusDisabled},
1170 },
1171 wantRunning: nil,
1172 wantDyncfg: `
@@ -1208,10 +1209,10 @@ CONFIG test:collector:fail:test status disabled
1209 },
1210 wantDiscovered: nil,
1211 wantSeen: []seenConfig{
1211 - {cfg: cfg, status: dyncfgDisabled},
1212 + {cfg: cfg, status: dyncfg.StatusDisabled},
1213 },
1214 wantExposed: []seenConfig{
1214 - {cfg: cfg, status: dyncfgDisabled},
1215 + {cfg: cfg, status: dyncfg.StatusDisabled},
1216 },
1217 wantRunning: nil,
1218 wantDyncfg: `
@@ -1294,10 +1295,10 @@ FUNCTION_RESULT_END
1295 },
1296 wantDiscovered: nil,
1297 wantSeen: []seenConfig{
1297 - {cfg: cfg, status: dyncfgAccepted},
1298 + {cfg: cfg, status: dyncfg.StatusAccepted},
1299 },
1300 wantExposed: []seenConfig{
1300 - {cfg: cfg, status: dyncfgAccepted},
1301 + {cfg: cfg, status: dyncfg.StatusAccepted},
1302 },
1303 wantRunning: nil,
1304 wantDyncfg: `
@@ -1340,10 +1341,10 @@ CONFIG test:collector:success:test status accepted
1341 },
1342 wantDiscovered: nil,
1343 wantSeen: []seenConfig{
1343 - {cfg: cfg, status: dyncfgRunning},
1344 + {cfg: cfg, status: dyncfg.StatusRunning},
1345 },
1346 wantExposed: []seenConfig{
1346 - {cfg: cfg, status: dyncfgRunning},
1347 + {cfg: cfg, status: dyncfg.StatusRunning},
1348 },
1349 wantRunning: []string{cfg.FullName()},
1350 wantDyncfg: `
@@ -1392,10 +1393,10 @@ CONFIG test:collector:success:test status running
1393 },
1394 wantDiscovered: nil,
1395 wantSeen: []seenConfig{
1395 - {cfg: cfg, status: dyncfgDisabled},
1396 + {cfg: cfg, status: dyncfg.StatusDisabled},
1397 },
1398 wantExposed: []seenConfig{
1398 - {cfg: cfg, status: dyncfgDisabled},
1399 + {cfg: cfg, status: dyncfg.StatusDisabled},
1400 },
1401 wantRunning: nil,
1402 wantDyncfg: `
@@ -1448,10 +1449,10 @@ CONFIG test:collector:success:test status disabled
1449 },
1450 wantDiscovered: nil,
1451 wantSeen: []seenConfig{
1451 - {cfg: cfg, status: dyncfgRunning},
1452 + {cfg: cfg, status: dyncfg.StatusRunning},
1453 },
1454 wantExposed: []seenConfig{
1454 - {cfg: cfg, status: dyncfgRunning},
1455 + {cfg: cfg, status: dyncfg.StatusRunning},
1456 },
1457 wantRunning: []string{cfg.FullName()},
1458 wantDyncfg: `
@@ -1566,14 +1567,14 @@ FUNCTION_RESULT_END
1567 discCfg,
1568 },
1569 wantSeen: []seenConfig{
1569 - {cfg: stockCfg, status: dyncfgRunning},
1570 - {cfg: userCfg, status: dyncfgRunning},
1571 - {cfg: discCfg, status: dyncfgRunning},
1570 + {cfg: stockCfg, status: dyncfg.StatusRunning},
1571 + {cfg: userCfg, status: dyncfg.StatusRunning},
1572 + {cfg: discCfg, status: dyncfg.StatusRunning},
1573 },
1574 wantExposed: []seenConfig{
1574 - {cfg: stockCfg, status: dyncfgRunning},
1575 - {cfg: userCfg, status: dyncfgRunning},
1576 - {cfg: discCfg, status: dyncfgRunning},
1575 + {cfg: stockCfg, status: dyncfg.StatusRunning},
1576 + {cfg: userCfg, status: dyncfg.StatusRunning},
1577 + {cfg: discCfg, status: dyncfg.StatusRunning},
1578 },
1579 wantRunning: []string{stockCfg.FullName(), userCfg.FullName(), discCfg.FullName()},
1580 wantDyncfg: `
@@ -1771,10 +1772,10 @@ FUNCTION_RESULT_END
1772 },
1773 wantDiscovered: nil,
1774 wantSeen: []seenConfig{
1774 - {cfg: updCfg, status: dyncfgRunning},
1775 + {cfg: updCfg, status: dyncfg.StatusRunning},
1776 },
1777 wantExposed: []seenConfig{
1777 - {cfg: updCfg, status: dyncfgRunning},
1778 + {cfg: updCfg, status: dyncfg.StatusRunning},
1779 },
1780 wantRunning: []string{updCfg.FullName()},
1781 wantDyncfg: `
@@ -1830,10 +1831,10 @@ CONFIG test:collector:success:test status running
1831 },
1832 wantDiscovered: nil,
1833 wantSeen: []seenConfig{
1833 - {cfg: updCfg, status: dyncfgDisabled},
1834 + {cfg: updCfg, status: dyncfg.StatusDisabled},
1835 },
1836 wantExposed: []seenConfig{
1836 - {cfg: updCfg, status: dyncfgDisabled},
1837 + {cfg: updCfg, status: dyncfg.StatusDisabled},
1838 },
1839 wantRunning: nil,
1840 wantDyncfg: `
src/go/plugin/go.d/agent/jobmgr/sim_test.go
+2 -1
@@ -13,6 +13,7 @@ import (
13 "github.com/netdata/netdata/go/plugins/pkg/netdataapi"
14 "github.com/netdata/netdata/go/plugins/pkg/safewriter"
15 "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/confgroup"
16 + "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/dyncfg"
17 "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/module"
18
19 "github.com/stretchr/testify/assert"
@@ -36,7 +37,7 @@ func (s *runSim) run(t *testing.T) {
37
38 var buf bytes.Buffer
39 mgr := New()
39 - mgr.api = netdataapi.New(safewriter.New(&buf))
40 + mgr.dyncfgApi = dyncfg.NewResponder(netdataapi.New(safewriter.New(&buf)))
41 mgr.Modules = prepareMockRegistry()
42
43 done := make(chan struct{})