| 1 | // SPDX-License-Identifier: GPL-3.0-or-later |
| 2 | |
| 3 | package jobmgr |
| 4 | |
| 5 | import ( |
| 6 | "context" |
| 7 | "encoding/json" |
| 8 | "log/slog" |
| 9 | "time" |
| 10 | |
| 11 | "github.com/netdata/netdata/go/plugins/logger" |
| 12 | "github.com/netdata/netdata/go/plugins/plugin/framework/collectorapi" |
| 13 | "github.com/netdata/netdata/go/plugins/plugin/framework/confgroup" |
| 14 | "github.com/netdata/netdata/go/plugins/plugin/framework/dyncfg" |
| 15 | ) |
| 16 | |
| 17 | type dyncfgCmdTestTask struct { |
| 18 | fn dyncfg.Function |
| 19 | moduleName string |
| 20 | creator collectorapi.Creator |
| 21 | cfg confgroup.Config |
| 22 | timeout time.Duration |
| 23 | } |
| 24 | |
| 25 | func (m *Manager) requireModuleFromID(fn dyncfg.Function, target string) (string, collectorapi.Creator, bool) { |
| 26 | cmd := fn.Command() |
| 27 | id := fn.ID() |
| 28 | |
| 29 | moduleName, ok := m.extractModuleName(id) |
| 30 | if !ok { |
| 31 | m.Warningf("dyncfg: %s: could not extract %s from id (%s)", cmd, target, id) |
| 32 | m.dyncfgResponder.SendCodef(fn, 400, "Invalid ID format. Could not extract %s from ID. Provided ID: %s.", target, id) |
| 33 | return "", collectorapi.Creator{}, false |
| 34 | } |
| 35 | |
| 36 | creator, ok := m.modules.Lookup(moduleName) |
| 37 | if !ok { |
| 38 | m.Warningf("dyncfg: %s: module %s not found", cmd, moduleName) |
| 39 | m.dyncfgResponder.SendCodef(fn, 404, "The specified module '%s' is not registered.", moduleName) |
| 40 | return "", collectorapi.Creator{}, false |
| 41 | } |
| 42 | |
| 43 | return moduleName, creator, true |
| 44 | } |
| 45 | |
| 46 | func (m *Manager) requireTemplateModule(fn dyncfg.Function) (string, collectorapi.Creator, bool) { |
| 47 | return m.requireModuleFromID(fn, "module and job name") |
| 48 | } |
| 49 | |
| 50 | func (m *Manager) requireModule(fn dyncfg.Function) (string, collectorapi.Creator, bool) { |
| 51 | return m.requireModuleFromID(fn, "module name") |
| 52 | } |
| 53 | |
| 54 | func (m *Manager) requireModuleJob(fn dyncfg.Function) (string, string, collectorapi.Creator, bool) { |
| 55 | cmd := fn.Command() |
| 56 | id := fn.ID() |
| 57 | |
| 58 | moduleName, jobName, ok := m.extractModuleJobName(id) |
| 59 | if !ok { |
| 60 | m.Warningf("dyncfg: %s: could not extract module and job from id (%s)", cmd, id) |
| 61 | m.dyncfgResponder.SendCodef(fn, 400, "Invalid ID format. Could not extract module and job name from ID. Provided ID: %s.", id) |
| 62 | return "", "", collectorapi.Creator{}, false |
| 63 | } |
| 64 | |
| 65 | creator, ok := m.modules.Lookup(moduleName) |
| 66 | if !ok { |
| 67 | m.Warningf("dyncfg: %s: module %s not found", cmd, moduleName) |
| 68 | m.dyncfgResponder.SendCodef(fn, 404, "The specified module '%s' is not registered.", moduleName) |
| 69 | return "", "", collectorapi.Creator{}, false |
| 70 | } |
| 71 | |
| 72 | return moduleName, jobName, creator, true |
| 73 | } |
| 74 | |
| 75 | func (m *Manager) dyncfgCmdUserconfig(fn dyncfg.Function) { |
| 76 | jn := fn.JobName() |
| 77 | if jn == "" { |
| 78 | jn = "test" |
| 79 | } |
| 80 | |
| 81 | mn, creator, ok := m.requireTemplateModule(fn) |
| 82 | if !ok { |
| 83 | return |
| 84 | } |
| 85 | |
| 86 | if creator.Config == nil || creator.Config() == nil { |
| 87 | m.Warningf("dyncfg: %s: module %s: configuration not found", fn.Command(), mn) |
| 88 | m.dyncfgResponder.SendCodef(fn, 500, "Module %s does not provide configuration.", mn) |
| 89 | return |
| 90 | } |
| 91 | |
| 92 | bs, err := userConfigFromPayload(creator.Config(), jn, fn) |
| 93 | if err != nil { |
| 94 | m.Warningf("dyncfg: %s: module %s: failed to create config from payload: %v", fn.Command(), mn, err) |
| 95 | m.dyncfgResponder.SendCodef(fn, 400, "Invalid configuration format. Failed to create configuration from payload: %v.", err) |
| 96 | return |
| 97 | } |
| 98 | |
| 99 | m.dyncfgResponder.SendYAML(fn, string(bs)) |
| 100 | } |
| 101 | |
| 102 | func (m *Manager) dyncfgCmdTest(fn dyncfg.Function) { |
| 103 | cmd := fn.Command() |
| 104 | |
| 105 | mn, creator, ok := m.requireTemplateModule(fn) |
| 106 | if !ok { |
| 107 | return |
| 108 | } |
| 109 | |
| 110 | jn := fn.JobName() |
| 111 | if jn == "" { |
| 112 | jn = "test" |
| 113 | } |
| 114 | |
| 115 | m.Infof("dyncfg: %s: %s/%s job by user '%s'", cmd, mn, jn, fn.User()) |
| 116 | |
| 117 | if err := dyncfg.JobNameRuleStrict(jn); err != nil { |
| 118 | m.Warningf("dyncfg: %s: module %s: unacceptable job name '%s': %v", cmd, mn, jn, err) |
| 119 | m.dyncfgResponder.SendCodef(fn, 400, "Unacceptable job name '%s': %v.", jn, err) |
| 120 | return |
| 121 | } |
| 122 | if !fn.HasPayload() { |
| 123 | m.Warningf("dyncfg: %s: module %s: missing configuration payload", cmd, mn) |
| 124 | m.dyncfgResponder.SendCodef(fn, 400, "Missing configuration payload.") |
| 125 | return |
| 126 | } |
| 127 | |
| 128 | cfg, err := configFromPayload(fn) |
| 129 | if err != nil { |
| 130 | m.Warningf("dyncfg: %s: module %s: failed to create config from payload: %v", cmd, mn, err) |
| 131 | m.dyncfgResponder.SendCodef(fn, 400, "Invalid configuration format. Failed to create configuration from payload: %v.", err) |
| 132 | return |
| 133 | } |
| 134 | |
| 135 | if cfg.Vnode() != "" { |
| 136 | if _, ok := m.vnodesCtl.Lookup(cfg.Vnode()); !ok { |
| 137 | m.Warningf("dyncfg: %s: module %s: vnode %s not found", cmd, mn, cfg.Vnode()) |
| 138 | m.dyncfgResponder.SendCodef(fn, 400, "The specified vnode '%s' is not registered.", cfg.Vnode()) |
| 139 | return |
| 140 | } |
| 141 | } |
| 142 | |
| 143 | cfg.SetModule(mn) |
| 144 | cfg.SetName(jn) |
| 145 | |
| 146 | if err := m.baseContext().Err(); err != nil { |
| 147 | m.dyncfgResponder.SendCodef(fn, 503, "Job manager is shutting down.") |
| 148 | return |
| 149 | } |
| 150 | |
| 151 | select { |
| 152 | case m.cmdTestSem <- struct{}{}: |
| 153 | task := dyncfgCmdTestTask{ |
| 154 | fn: fn, |
| 155 | moduleName: mn, |
| 156 | creator: creator, |
| 157 | cfg: cfg, |
| 158 | timeout: m.dyncfgCmdTestTimeout(fn), |
| 159 | } |
| 160 | m.cmdTestWG.Go(func() { |
| 161 | m.runDyncfgCmdTest(task) |
| 162 | }) |
| 163 | default: |
| 164 | m.Warningf("dyncfg: %s: module %s: too many concurrent test requests", cmd, mn) |
| 165 | m.dyncfgResponder.SendCodef(fn, 503, "Too many concurrent test requests, try again later.") |
| 166 | } |
| 167 | } |
| 168 | |
| 169 | func (m *Manager) runDyncfgCmdTest(task dyncfgCmdTestTask) { |
| 170 | defer func() { <-m.cmdTestSem }() |
| 171 | |
| 172 | job, err := newConfigModule(task.creator) |
| 173 | if err != nil { |
| 174 | m.Warningf("dyncfg: test: module %s: failed to create module: %v", task.moduleName, err) |
| 175 | m.dyncfgResponder.SendCodef(task.fn, 500, "Module %s instantiation failed: %v.", task.moduleName, err) |
| 176 | return |
| 177 | } |
| 178 | |
| 179 | cleanupCtx, cleanupCancel := context.WithTimeout(m.baseContext(), cmdTestWorkerDrainWait) |
| 180 | defer cleanupCancel() |
| 181 | defer job.Cleanup(cleanupCtx) |
| 182 | |
| 183 | ctx, cancel := context.WithTimeout(m.baseContext(), task.timeout) |
| 184 | defer cancel() |
| 185 | |
| 186 | secretStoreSvc := m.secretsCtl.Service() |
| 187 | storeSnapshot := secretStoreSvc.Capture() |
| 188 | resolveCtx := collectorSecretResolveContext(ctx, m.Logger, task.cfg) |
| 189 | if err := applyConfig(resolveCtx, task.cfg, job, m.secretResolver, secretStoreSvc, storeSnapshot); err != nil { |
| 190 | m.Warningf("dyncfg: test: module %s job %s: failed to apply config: %v", task.moduleName, task.cfg.Name(), err) |
| 191 | m.dyncfgResponder.SendCodef(task.fn, 400, "Invalid configuration. Failed to apply configuration: %v.", err) |
| 192 | return |
| 193 | } |
| 194 | |
| 195 | job.GetBase().Logger = logger.New().With( |
| 196 | slog.String("collector", task.cfg.Module()), |
| 197 | slog.String("job", task.cfg.Name()), |
| 198 | ) |
| 199 | |
| 200 | if err := job.Init(ctx); err != nil { |
| 201 | m.dyncfgResponder.SendCodef(task.fn, 422, "Job initialization failed: %v", err) |
| 202 | return |
| 203 | } |
| 204 | if err := job.Check(ctx); err != nil { |
| 205 | m.dyncfgResponder.SendCodef(task.fn, 422, "Job check failed: %v", err) |
| 206 | return |
| 207 | } |
| 208 | |
| 209 | m.dyncfgResponder.SendCodef(task.fn, 200, "") |
| 210 | } |
| 211 | |
| 212 | func (m *Manager) dyncfgCmdTestTimeout(fn dyncfg.Function) time.Duration { |
| 213 | if timeout := fn.Fn().Timeout; timeout > 0 { |
| 214 | return timeout |
| 215 | } |
| 216 | return cmdTestDefaultTimeout |
| 217 | } |
| 218 | |
| 219 | func (m *Manager) dyncfgCmdSchema(fn dyncfg.Function) { |
| 220 | mn, mod, ok := m.requireModule(fn) |
| 221 | if !ok { |
| 222 | return |
| 223 | } |
| 224 | |
| 225 | m.Infof("dyncfg: %s: %s module by user '%s'", fn.Command(), mn, fn.User()) |
| 226 | |
| 227 | if mod.JobConfigSchema == "" { |
| 228 | m.Warningf("dyncfg: schema: module %s: schema not found", mn) |
| 229 | m.dyncfgResponder.SendCodef(fn, 500, "Module %s configuration schema not found.", mn) |
| 230 | return |
| 231 | } |
| 232 | |
| 233 | m.dyncfgResponder.SendJSON(fn, mod.JobConfigSchema) |
| 234 | } |
| 235 | |
| 236 | func (m *Manager) dyncfgCmdGet(fn dyncfg.Function) { |
| 237 | cmd := fn.Command() |
| 238 | |
| 239 | mn, jn, creator, ok := m.requireModuleJob(fn) |
| 240 | if !ok { |
| 241 | return |
| 242 | } |
| 243 | |
| 244 | m.Infof("dyncfg: %s: %s/%s job by user '%s'", fn.Command(), mn, jn, fn.User()) |
| 245 | |
| 246 | entry, ok := m.exposedLookupByName(mn, jn) |
| 247 | if !ok { |
| 248 | m.Warningf("dyncfg: %s: module %s job %s not found", cmd, mn, jn) |
| 249 | m.dyncfgResponder.SendCodef(fn, 404, "The specified module '%s' job '%s' is not registered.", mn, jn) |
| 250 | return |
| 251 | } |
| 252 | |
| 253 | mod, err := newConfigModule(creator) |
| 254 | if err != nil { |
| 255 | m.Warningf("dyncfg: %s: module %s job %s failed to create module: %v", cmd, mn, jn, err) |
| 256 | m.dyncfgResponder.SendCodef(fn, 500, "Module %s instantiation failed: %v.", mn, err) |
| 257 | return |
| 258 | } |
| 259 | |
| 260 | if err := applyConfigRaw(entry.Cfg, mod); err != nil { |
| 261 | m.Warningf("dyncfg: %s: module %s job %s failed to apply config: %v", cmd, mn, jn, err) |
| 262 | m.dyncfgResponder.SendCodef(fn, 400, "Invalid configuration. Failed to apply configuration: %v.", err) |
| 263 | return |
| 264 | } |
| 265 | |
| 266 | conf := mod.Configuration() |
| 267 | if conf == nil { |
| 268 | m.Warningf("dyncfg: %s: module %s: configuration not found", cmd, mn) |
| 269 | m.dyncfgResponder.SendCodef(fn, 500, "Module %s does not provide configuration.", mn) |
| 270 | return |
| 271 | } |
| 272 | |
| 273 | bs, err := json.Marshal(conf) |
| 274 | if err != nil { |
| 275 | m.Warningf("dyncfg: %s: module %s job %s failed to json marshal config: %v", cmd, mn, jn, err) |
| 276 | m.dyncfgResponder.SendCodef(fn, 500, "Failed to convert configuration into JSON: %v.", err) |
| 277 | return |
| 278 | } |
| 279 | |
| 280 | m.dyncfgResponder.SendJSON(fn, string(bs)) |
| 281 | } |