@cryptotaxi247 / netdata-1 / commits / 8653926e5

feat(go.d.plugin/sql): add function support for interactive table views (#21666)

Ilya Mashchenko committed Jan 29, 2026 at 21:05 UTC 8653926e58637fec88901b487c50853df68ccad4
15 files changed +2150 -31
src/go/pkg/funcapi/columns.go
+2 -1
@@ -16,7 +16,8 @@ type ValueOptions struct {
16 type Column struct {
17 // Index is the 0-based position in each row array and must match data order.
18 Index int
19 - // Name is the header label shown in the UI.
19 + // Name is the column tooltip shown in the UI.
20 + // The column header is the map key used in the columns map, not this field.
21 Name string
22 // Type controls the base data type and default rendering.
23 Type FieldType
src/go/pkg/netdataapi/api.go
+8
@@ -198,3 +198,11 @@ func (a *API) FUNCTIONGLOBAL(opts FunctionGlobalOpts) {
198 strconv.Itoa(opts.Priority) + " " +
199 strconv.Itoa(opts.Version) + "\n\n"))
200 }
201 +
202 +// FUNCTIONREMOVE removes a function from Netdata.
203 +// NOTE: This is a no-op placeholder - Netdata core does not yet support function removal.
204 +// When Netdata implements this, the protocol format will be added here.
205 +func (a *API) FUNCTIONREMOVE(name string) {
206 + // TODO: Implement when Netdata core supports function removal
207 + // For now, this is intentionally a no-op
208 +}
src/go/plugin/go.d/agent/dyncfg/responder.go
+22 -8
@@ -33,16 +33,25 @@ func (r *Responder) SendCodef(fn functions.Function, code int, message string, a
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,
36 + var payload []byte
37 + if code >= 400 && code < 600 {
38 + payload, _ = json.Marshal(struct {
39 + Status int `json:"status"`
40 + ErrorMessage string `json:"errorMessage"`
41 + }{
42 + Status: code,
43 + ErrorMessage: msg,
44 + })
45 + } else {
46 + payload, _ = json.Marshal(struct {
47 + Status int `json:"status"`
48 + Message string `json:"message"`
49 + }{
50 + Status: code,
51 + Message: msg,
52 + })
53 }
54
44 - payload, _ := json.Marshal(response)
45 -
55 r.api.FUNCRESULT(netdataapi.FunctionResult{
56 UID: fn.UID,
57 ContentType: "application/json",
@@ -110,3 +119,8 @@ func (r *Responder) ConfigDelete(id string) {
119 func (r *Responder) FunctionGlobal(opts netdataapi.FunctionGlobalOpts) {
120 r.api.FUNCTIONGLOBAL(opts)
121 }
122 +
123 +// FunctionRemove removes a function from Netdata (no-op until Netdata core supports it)
124 +func (r *Responder) FunctionRemove(name string) {
125 + r.api.FUNCTIONREMOVE(name)
126 +}
src/go/plugin/go.d/agent/functions/manager.go
+20 -7
@@ -146,13 +146,26 @@ func (m *Manager) handlePrefixRouting(f Function, fs *functionSet) {
146 }
147
148 func (m *Manager) respf(fn *Function, code int, msgf string, a ...any) {
149 - bs, _ := json.Marshal(struct {
150 - Status int `json:"status"`
151 - Message string `json:"message"`
152 - }{
153 - Status: code,
154 - Message: fmt.Sprintf(msgf, a...),
155 - })
149 + msg := fmt.Sprintf(msgf, a...)
150 +
151 + var bs []byte
152 + if code >= 400 && code < 600 {
153 + bs, _ = json.Marshal(struct {
154 + Status int `json:"status"`
155 + ErrorMessage string `json:"errorMessage"`
156 + }{
157 + Status: code,
158 + ErrorMessage: msg,
159 + })
160 + } else {
161 + bs, _ = json.Marshal(struct {
162 + Status int `json:"status"`
163 + Message string `json:"message"`
164 + }{
165 + Status: code,
166 + Message: msg,
167 + })
168 + }
169
170 m.api.FUNCRESULT(netdataapi.FunctionResult{
171 UID: fn.UID,
src/go/plugin/go.d/agent/jobmgr/funcshandler.go
+207
@@ -458,3 +458,210 @@ func buildAcceptedParams(methodParams []funcapi.ParamConfig) []string {
458 }
459 return accepted
460 }
461 +
462 +// makeJobMethodFuncHandler creates a function handler for a job-specific method.
463 +// Unlike makeMethodFuncHandler, this handler routes directly to a specific job
464 +// without needing the __job parameter (the job is known from the function name).
465 +func (m *Manager) makeJobMethodFuncHandler(moduleName, jobName, methodID string) func(functions.Function) {
466 + return func(fn functions.Function) {
467 + // Check for "info" request
468 + if slices.Contains(fn.Args, "info") {
469 + m.handleJobMethodFuncInfo(moduleName, jobName, methodID, fn)
470 + return
471 + }
472 +
473 + methodCfg, ok := m.moduleFuncs.getJobMethod(moduleName, jobName, methodID)
474 + if !ok {
475 + m.respondError(fn, 404, "unknown method '%s' for job '%s:%s'", methodID, moduleName, jobName)
476 + return
477 + }
478 +
479 + // Get job WITH generation for race condition detection
480 + job, jobGen := m.moduleFuncs.getJobWithGeneration(moduleName, jobName)
481 + if job == nil {
482 + m.respondError(fn, 503, "job '%s:%s' is not running", moduleName, jobName)
483 + return
484 + }
485 +
486 + // Create context with timeout from function request
487 + ctx, cancel := context.WithTimeout(context.Background(), fn.Timeout)
488 + defer cancel()
489 +
490 + // Verify job is still running before calling handler
491 + if !job.IsRunning() {
492 + m.respondError(fn, 503, "job '%s:%s' is no longer running", moduleName, jobName)
493 + return
494 + }
495 +
496 + // Get the creator for this module to call MethodHandler
497 + creator, ok := m.moduleFuncs.getCreator(moduleName)
498 + if !ok || creator.MethodHandler == nil {
499 + m.respondError(fn, 500, "module '%s' does not implement MethodHandler", moduleName)
500 + return
501 + }
502 +
503 + // Get the handler for this job
504 + handler := creator.MethodHandler(job)
505 + if handler == nil {
506 + m.respondError(fn, 500, "module '%s' returned nil handler for job '%s'", moduleName, jobName)
507 + return
508 + }
509 +
510 + payload := parsePayload(fn.Payload)
511 + argValues := parseArgsParams(fn.Args)
512 +
513 + // Resolve method-specific required params
514 + methodParams, paramsFromJob, err := m.resolveJobMethodParams(ctx, methodCfg, handler, methodID)
515 + if err != nil {
516 + m.respondError(fn, 503, "job '%s:%s' cannot provide parameters: %v", moduleName, jobName, err)
517 + return
518 + }
519 +
520 + // Validate provided param values
521 + if paramsFromJob {
522 + if err := validateParamValues(methodParams, argValues, payload, jobName); err != nil {
523 + m.respondError(fn, 400, "%v", err)
524 + return
525 + }
526 + }
527 +
528 + methodParamValues := make(map[string][]string, len(methodParams))
529 + for _, paramCfg := range methodParams {
530 + methodParamValues[paramCfg.ID] = paramValues(argValues, payload, paramCfg.ID)
531 + }
532 + resolvedParams := funcapi.ResolveParams(methodParams, methodParamValues)
533 +
534 + // Route to the module's handler
535 + dataResp := handler.Handle(ctx, methodID, resolvedParams)
536 +
537 + // Verify job was not replaced during handler execution
538 + if !m.moduleFuncs.verifyJobGeneration(moduleName, jobName, jobGen) {
539 + m.respondError(fn, 503, "job '%s:%s' was replaced during request, please retry", moduleName, jobName)
540 + return
541 + }
542 +
543 + updateEvery := 1
544 + if methodCfg.UpdateEvery > 1 {
545 + updateEvery = methodCfg.UpdateEvery
546 + }
547 + m.respondJobMethodWithParams(fn, dataResp, methodParams, updateEvery)
548 + }
549 +}
550 +
551 +// handleJobMethodFuncInfo handles "info" requests for a job-specific method
552 +func (m *Manager) handleJobMethodFuncInfo(moduleName, jobName, methodID string, fn functions.Function) {
553 + methodCfg, ok := m.moduleFuncs.getJobMethod(moduleName, jobName, methodID)
554 + if !ok {
555 + m.respondError(fn, 404, "unknown method '%s' for job '%s:%s'", methodID, moduleName, jobName)
556 + return
557 + }
558 +
559 + methodParams := methodCfg.RequiredParams
560 + help := methodCfg.Help
561 + if help == "" {
562 + help = fmt.Sprintf("%s %s data function", moduleName, methodID)
563 + }
564 +
565 + updateEvery := 1
566 + if methodCfg.UpdateEvery > 1 {
567 + updateEvery = methodCfg.UpdateEvery
568 + }
569 +
570 + resp := map[string]any{
571 + "v": 3,
572 + "update_every": updateEvery,
573 + "status": 200,
574 + "type": "table",
575 + "has_history": false,
576 + "help": help,
577 + "accepted_params": buildJobMethodAcceptedParams(methodParams),
578 + "required_params": buildJobMethodRequiredParams(methodParams),
579 + }
580 +
581 + m.respondJSON(fn, resp)
582 +}
583 +
584 +// resolveJobMethodParams resolves method parameters for a job-specific method
585 +func (m *Manager) resolveJobMethodParams(ctx context.Context, methodCfg *funcapi.MethodConfig, handler funcapi.MethodHandler, methodID string) ([]funcapi.ParamConfig, bool, error) {
586 + methodParams := methodCfg.RequiredParams
587 +
588 + jobParams, err := handler.MethodParams(ctx, methodID)
589 + if err != nil {
590 + return nil, false, err
591 + }
592 + if len(jobParams) == 0 {
593 + return methodParams, true, nil
594 + }
595 +
596 + return funcapi.MergeParamConfigs(methodParams, jobParams), true, nil
597 +}
598 +
599 +// respondJobMethodWithParams wraps the module's data response for job-specific methods
600 +func (m *Manager) respondJobMethodWithParams(fn functions.Function, dataResp *funcapi.FunctionResponse, methodParams []funcapi.ParamConfig, updateEvery int) {
601 + if dataResp == nil {
602 + m.respondError(fn, 500, "internal error: module returned nil response")
603 + return
604 + }
605 +
606 + if dataResp.Status >= 400 {
607 + m.respondError(fn, dataResp.Status, "%s", dataResp.Message)
608 + return
609 + }
610 +
611 + paramsForResponse := methodParams
612 + if len(dataResp.RequiredParams) > 0 {
613 + paramsForResponse = funcapi.MergeParamConfigs(paramsForResponse, dataResp.RequiredParams)
614 + }
615 +
616 + resp := map[string]any{
617 + "v": 3,
618 + "update_every": updateEvery,
619 + "status": dataResp.Status,
620 + "type": "table",
621 + "has_history": false,
622 + "help": dataResp.Help,
623 + "accepted_params": buildJobMethodAcceptedParams(paramsForResponse),
624 + "required_params": buildJobMethodRequiredParams(paramsForResponse),
625 + }
626 +
627 + if dataResp.Columns != nil {
628 + resp["columns"] = dataResp.Columns
629 + }
630 + if dataResp.Data != nil {
631 + resp["data"] = dataResp.Data
632 + }
633 + if dataResp.DefaultSortColumn != "" {
634 + resp["default_sort_column"] = dataResp.DefaultSortColumn
635 + }
636 + if len(dataResp.Charts) > 0 {
637 + resp["charts"] = dataResp.Charts
638 + }
639 + if len(dataResp.DefaultCharts) > 0 {
640 + resp["default_charts"] = dataResp.DefaultCharts.Build()
641 + }
642 + if len(dataResp.GroupBy) > 0 {
643 + resp["group_by"] = dataResp.GroupBy
644 + }
645 +
646 + m.respondJSON(fn, resp)
647 +}
648 +
649 +// buildJobMethodAcceptedParams creates accepted_params for job-specific methods (no __job)
650 +func buildJobMethodAcceptedParams(methodParams []funcapi.ParamConfig) []string {
651 + accepted := make([]string, 0, len(methodParams))
652 + for _, p := range methodParams {
653 + if !slices.Contains(accepted, p.ID) {
654 + accepted = append(accepted, p.ID)
655 + }
656 + }
657 + return accepted
658 +}
659 +
660 +// buildJobMethodRequiredParams creates required_params for job-specific methods (no __job)
661 +func buildJobMethodRequiredParams(methodParams []funcapi.ParamConfig) []map[string]any {
662 + required := make([]map[string]any, 0, len(methodParams))
663 + for _, cfg := range methodParams {
664 + required = append(required, cfg.RequiredParam())
665 + }
666 + return required
667 +}
src/go/plugin/go.d/agent/jobmgr/manager.go
+90 -3
@@ -15,6 +15,7 @@ import (
15 "time"
16
17 "github.com/netdata/netdata/go/plugins/logger"
18 + "github.com/netdata/netdata/go/plugins/pkg/funcapi"
19 "github.com/netdata/netdata/go/plugins/pkg/netdataapi"
20 "github.com/netdata/netdata/go/plugins/pkg/safewriter"
21 "github.com/netdata/netdata/go/plugins/pkg/ticker"
@@ -122,9 +123,13 @@ func (m *Manager) Run(ctx context.Context, in chan []*confgroup.Group) {
123 for name, creator := range m.Modules {
124 m.dyncfgCollectorModuleCreate(name)
125
125 - // Register module-level function if this module provides methods
126 - if creator.Methods != nil {
126 + // Register module if it provides static methods OR per-job methods
127 + if creator.Methods != nil || creator.JobMethods != nil {
128 m.moduleFuncs.registerModule(name, creator)
129 + }
130 +
131 + // Register static module-level functions
132 + if creator.Methods != nil {
133 methods := creator.Methods()
134 for _, method := range methods {
135 if method.ID == "" {
@@ -157,6 +162,7 @@ func (m *Manager) Run(ctx context.Context, in chan []*confgroup.Group) {
162 })
163 }
164 }
165 + // Note: Per-job methods (JobMethods) are registered in startRunningJob
166 }
167
168 m.loadFileStatus()
@@ -343,6 +349,15 @@ func (m *Manager) startRunningJob(job *module.Job) {
349
350 // Track job for module function routing
351 m.moduleFuncs.addJob(job.ModuleName(), job.Name(), job)
352 +
353 + // Register job-specific methods if module provides JobMethods callback
354 + creator, ok := m.Modules.Lookup(job.ModuleName())
355 + if ok && creator.JobMethods != nil {
356 + methods := creator.JobMethods(job)
357 + if len(methods) > 0 {
358 + m.registerJobMethods(job, methods)
359 + }
360 + }
361 }
362
363 func (m *Manager) stopRunningJob(name string) {
@@ -354,6 +369,9 @@ func (m *Manager) stopRunningJob(name string) {
369 m.runningJobs.unlock()
370
371 if ok {
372 + // Unregister job-specific methods
373 + m.unregisterJobMethods(job)
374 +
375 // Remove job from module function registry
376 m.moduleFuncs.removeJob(job.ModuleName(), job.Name())
377 job.Stop()
@@ -385,6 +403,75 @@ func (m *Manager) cleanup() {
403 })
404 }
405
406 +// registerJobMethods registers methods for a specific job with Netdata
407 +func (m *Manager) registerJobMethods(job *module.Job, methods []funcapi.MethodConfig) {
408 + for _, method := range methods {
409 + if method.ID == "" {
410 + m.Warningf("skipping job method registration for %s[%s]: empty method ID", job.ModuleName(), job.Name())
411 + continue
412 + }
413 +
414 + funcName := fmt.Sprintf("%s:%s", job.ModuleName(), method.ID)
415 +
416 + // Register Go handler for this function
417 + m.FnReg.Register(funcName, m.makeJobMethodFuncHandler(job.ModuleName(), job.Name(), method.ID))
418 +
419 + // Notify Netdata about this function
420 + help := method.Help
421 + if help == "" {
422 + help = fmt.Sprintf("%s %s data function", job.ModuleName(), method.ID)
423 + }
424 +
425 + const cloudAccess = "0x0013" // SIGNED_ID | SAME_SPACE | SENSITIVE_DATA
426 + access := "0x0000"
427 + if method.RequireCloud {
428 + access = cloudAccess
429 + }
430 +
431 + m.dyncfgApi.FunctionGlobal(netdataapi.FunctionGlobalOpts{
432 + Name: funcName,
433 + Timeout: 60,
434 + Help: help,
435 + Tags: "top",
436 + Access: access,
437 + Priority: 100,
438 + Version: 3,
439 + })
440 +
441 + m.Debugf("registered job method: %s for job %s[%s]", funcName, job.ModuleName(), job.Name())
442 + }
443 +
444 + // Store methods in registry for later unregistration
445 + m.moduleFuncs.registerJobMethods(job.ModuleName(), job.Name(), methods)
446 +}
447 +
448 +// unregisterJobMethods unregisters methods for a specific job
449 +func (m *Manager) unregisterJobMethods(job *module.Job) {
450 + methods := m.moduleFuncs.getJobMethods(job.ModuleName(), job.Name())
451 + if len(methods) == 0 {
452 + return
453 + }
454 +
455 + for _, method := range methods {
456 + if method.ID == "" {
457 + continue
458 + }
459 +
460 + funcName := fmt.Sprintf("%s:%s", job.ModuleName(), method.ID)
461 +
462 + // Unregister Go handler
463 + m.FnReg.Unregister(funcName)
464 +
465 + // Notify Netdata to remove function (no-op until Netdata supports it)
466 + m.dyncfgApi.FunctionRemove(funcName)
467 +
468 + m.Debugf("unregistered job method: %s for job %s[%s]", funcName, job.ModuleName(), job.Name())
469 + }
470 +
471 + // Remove from registry
472 + m.moduleFuncs.unregisterJobMethods(job.ModuleName(), job.Name())
473 +}
474 +
475 func (m *Manager) createCollectorJob(cfg confgroup.Config) (*module.Job, error) {
476 creator, ok := m.Modules[cfg.Module()]
477 if !ok {
@@ -396,7 +483,7 @@ func (m *Manager) createCollectorJob(cfg confgroup.Config) (*module.Job, error)
483
484 // Reject if config sets function_only but module has no methods
485 // Note: module-level FunctionOnly without Methods is caught at registration time
399 - if cfg.FunctionOnly() && creator.Methods == nil {
486 + if cfg.FunctionOnly() && creator.Methods == nil && creator.JobMethods == nil {
487 return nil, fmt.Errorf("function_only is set but %s module has no methods defined", cfg.Module())
488 }
489
src/go/plugin/go.d/agent/jobmgr/modulefuncs.go
+61 -2
@@ -23,8 +23,9 @@ type moduleFunc struct {
23 creator module.Creator // The module creator (has Methods())
24 methods []funcapi.MethodConfig // Static methods from creator (ordered)
25 methodsByID map[string]funcapi.MethodConfig
26 - jobs map[string]*jobEntry // jobName → job entry with generation
27 - lastGeneration map[string]uint64 // jobName → last known generation (persists across removals)
26 + jobs map[string]*jobEntry // jobName → job entry with generation
27 + lastGeneration map[string]uint64 // jobName → last known generation (persists across removals)
28 + jobMethods map[string][]funcapi.MethodConfig // jobName → methods registered for that job
29 }
30
31 // jobEntry wraps a job with a generation number for race detection
@@ -55,6 +56,7 @@ func (r *moduleFuncRegistry) registerModule(name string, creator module.Creator)
56 methodsByID: indexMethods(methods),
57 jobs: make(map[string]*jobEntry),
58 lastGeneration: make(map[string]uint64),
59 + jobMethods: make(map[string][]funcapi.MethodConfig),
60 }
61 }
62
@@ -231,3 +233,60 @@ func (r *moduleFuncRegistry) isModuleRegistered(moduleName string) bool {
233 _, ok := r.modules[moduleName]
234 return ok
235 }
236 +
237 +// registerJobMethods stores methods registered for a specific job
238 +func (r *moduleFuncRegistry) registerJobMethods(moduleName, jobName string, methods []funcapi.MethodConfig) {
239 + r.mu.Lock()
240 + defer r.mu.Unlock()
241 +
242 + mf, ok := r.modules[moduleName]
243 + if !ok {
244 + return
245 + }
246 + mf.jobMethods[jobName] = methods
247 +}
248 +
249 +// unregisterJobMethods removes methods registered for a specific job
250 +func (r *moduleFuncRegistry) unregisterJobMethods(moduleName, jobName string) {
251 + r.mu.Lock()
252 + defer r.mu.Unlock()
253 +
254 + mf, ok := r.modules[moduleName]
255 + if !ok {
256 + return
257 + }
258 + delete(mf.jobMethods, jobName)
259 +}
260 +
261 +// getJobMethods returns methods registered for a specific job
262 +func (r *moduleFuncRegistry) getJobMethods(moduleName, jobName string) []funcapi.MethodConfig {
263 + r.mu.RLock()
264 + defer r.mu.RUnlock()
265 +
266 + mf, ok := r.modules[moduleName]
267 + if !ok {
268 + return nil
269 + }
270 + return mf.jobMethods[jobName]
271 +}
272 +
273 +// getJobMethod returns a specific method registered for a job by method ID
274 +func (r *moduleFuncRegistry) getJobMethod(moduleName, jobName, methodID string) (*funcapi.MethodConfig, bool) {
275 + r.mu.RLock()
276 + defer r.mu.RUnlock()
277 +
278 + mf, ok := r.modules[moduleName]
279 + if !ok {
280 + return nil, false
281 + }
282 + methods, ok := mf.jobMethods[jobName]
283 + if !ok {
284 + return nil, false
285 + }
286 + for i := range methods {
287 + if methods[i].ID == methodID {
288 + return &methods[i], true
289 + }
290 + }
291 + return nil, false
292 +}
src/go/plugin/go.d/agent/module/registry.go
+11 -2
@@ -43,6 +43,12 @@ type (
43 // When nil, methods are disabled for this module.
44 MethodHandler func(job *Job) funcapi.MethodHandler
45
46 + // Optional: JobMethods returns methods to register when a job starts.
47 + // Each method is registered as "moduleName:methodID" and unregistered when the job stops.
48 + // This enables per-job function registration instead of static module-level functions.
49 + // If nil, no per-job methods are registered.
50 + JobMethods func(job *Job) []funcapi.MethodConfig
51 +
52 // FunctionOnly indicates this module provides only functions, no metrics.
53 // Jobs created from this module skip data collection and chart creation.
54 // The module must still implement Init() and Check() for connectivity validation.
@@ -65,8 +71,11 @@ func (r Registry) Register(name string, creator Creator) {
71 if _, ok := r[name]; ok {
72 panic(fmt.Sprintf("%s is already in registry", name))
73 }
68 - if creator.FunctionOnly && creator.Methods == nil {
69 - panic(fmt.Sprintf("%s is FunctionOnly but has no Methods defined", name))
74 + if creator.Methods != nil && creator.JobMethods != nil {
75 + panic(fmt.Sprintf("%s has both Methods and JobMethods defined (mutually exclusive)", name))
76 + }
77 + if creator.FunctionOnly && creator.Methods == nil && creator.JobMethods == nil {
78 + panic(fmt.Sprintf("%s is FunctionOnly but has no Methods or JobMethods defined", name))
79 }
80 r[name] = creator
81 }
src/go/plugin/go.d/collector/sql/collector.go
+44 -2
@@ -7,6 +7,7 @@ import (
7 "database/sql"
8 _ "embed"
9 "errors"
10 + "sync"
11 "time"
12
13 "github.com/netdata/netdata/go/plugins/pkg/confopt"
@@ -21,6 +22,8 @@ func init() {
22 Create: func() module.Module { return New() },
23 JobConfigSchema: configSchema,
24 Config: func() any { return &Config{} },
25 + JobMethods: sqlJobMethods,
26 + MethodHandler: sqlMethodHandler,
27 })
28 }
29
@@ -41,9 +44,14 @@ type Collector struct {
44
45 charts *module.Charts
46
44 - db *sql.DB
47 + dbMu sync.RWMutex
48 + db *sql.DB
49 + dbCtx context.Context
50 + dbCancel context.CancelFunc
51
52 seenCharts map[string]bool
53 +
54 + funcTable *funcTable
55 }
56
57 func (c *Collector) Configuration() any {
@@ -51,14 +59,35 @@ func (c *Collector) Configuration() any {
59 }
60
61 func (c *Collector) Charts() *module.Charts {
62 + if c.Config.FunctionOnly {
63 + return nil
64 + }
65 return c.charts
66 }
67
68 func (c *Collector) Init(context.Context) error {
58 - return c.validateConfig()
69 + if err := c.validateConfig(); err != nil {
70 + return err
71 + }
72 +
73 + c.funcTable = newFuncTable(c)
74 +
75 + return nil
76 }
77
78 func (c *Collector) Check(ctx context.Context) error {
79 + if c.db == nil {
80 + if err := c.openConnection(ctx); err != nil {
81 + return err
82 + }
83 + // Create cancellable context for function queries
84 + c.dbCtx, c.dbCancel = context.WithCancel(context.Background())
85 + }
86 +
87 + if c.Config.FunctionOnly {
88 + return nil
89 + }
90 +
91 mx, err := c.collect(ctx)
92 if err != nil {
93 return err
@@ -70,6 +99,10 @@ func (c *Collector) Check(ctx context.Context) error {
99 }
100
101 func (c *Collector) Collect(ctx context.Context) map[string]int64 {
102 + if c.Config.FunctionOnly {
103 + return nil
104 + }
105 +
106 mx, err := c.collect(ctx)
107 if err != nil {
108 c.Error(err)
@@ -82,6 +115,15 @@ func (c *Collector) Collect(ctx context.Context) map[string]int64 {
115 }
116
117 func (c *Collector) Cleanup(context.Context) {
118 + // Cancel context first to signal in-flight queries to abort
119 + if c.dbCancel != nil {
120 + c.dbCancel()
121 + }
122 +
123 + // Acquire write lock - should be quick since queries are aborting
124 + c.dbMu.Lock()
125 + defer c.dbMu.Unlock()
126 +
127 if c.db != nil {
128 _ = c.db.Close()
129 c.db = nil
src/go/plugin/go.d/collector/sql/collector_test.go
+174
@@ -22,7 +22,181 @@ func TestCollector_ConfigurationSerialize(t *testing.T) {
22 }
23
24 func TestCollector_Charts(t *testing.T) {
25 + // Default: Charts() returns non-nil (metrics mode)
26 assert.NotNil(t, New().Charts())
27 +
28 + // With metrics configured, Charts() returns non-nil
29 + c := New()
30 + c.Config.Metrics = []ConfigMetricBlock{{ID: "test"}}
31 + assert.NotNil(t, c.Charts())
32 +
33 + // Function-only mode: Charts() returns nil
34 + c2 := New()
35 + c2.Config.FunctionOnly = true
36 + assert.Nil(t, c2.Charts())
37 +
38 + // Combined mode (function_only: false with both): Charts() returns non-nil
39 + c3 := New()
40 + c3.Config.Metrics = []ConfigMetricBlock{{ID: "test"}}
41 + c3.Config.Functions = []ConfigFunction{{ID: "test", Query: "SELECT 1"}}
42 + assert.NotNil(t, c3.Charts())
43 +}
44 +
45 +func TestCollector_Init_ConfigValidation(t *testing.T) {
46 + tests := map[string]struct {
47 + setup func(*Collector)
48 + wantFail bool
49 + }{
50 + "no metrics fails": {
51 + setup: func(c *Collector) {
52 + c.Driver = "pgx"
53 + c.DSN = "postgres://user:pass@localhost/db"
54 + // no metrics, function_only not set
55 + },
56 + wantFail: true,
57 + },
58 + "metrics only succeeds": {
59 + setup: func(c *Collector) {
60 + c.Driver = "pgx"
61 + c.DSN = "postgres://user:pass@localhost/db"
62 + c.Metrics = []ConfigMetricBlock{{
63 + ID: "test",
64 + Mode: "columns",
65 + Query: "SELECT 1 AS val",
66 + Charts: []ConfigChartConfig{{
67 + Title: "test", Context: "test", Family: "test", Units: "x",
68 + Dims: []ConfigDimConfig{{Name: "val", Source: "val"}},
69 + }},
70 + }}
71 + },
72 + wantFail: false,
73 + },
74 + "function_only with functions succeeds": {
75 + setup: func(c *Collector) {
76 + c.Driver = "pgx"
77 + c.DSN = "postgres://user:pass@localhost/db"
78 + c.FunctionOnly = true
79 + c.Functions = []ConfigFunction{{ID: "test", Query: "SELECT 1"}}
80 + },
81 + wantFail: false,
82 + },
83 + "function_only without functions fails": {
84 + setup: func(c *Collector) {
85 + c.Driver = "pgx"
86 + c.DSN = "postgres://user:pass@localhost/db"
87 + c.FunctionOnly = true
88 + // no functions
89 + },
90 + wantFail: true,
91 + },
92 + "function_only with metrics fails": {
93 + setup: func(c *Collector) {
94 + c.Driver = "pgx"
95 + c.DSN = "postgres://user:pass@localhost/db"
96 + c.FunctionOnly = true
97 + c.Functions = []ConfigFunction{{ID: "func", Query: "SELECT 1"}}
98 + c.Metrics = []ConfigMetricBlock{{
99 + ID: "test",
100 + Mode: "columns",
101 + Query: "SELECT 1 AS val",
102 + Charts: []ConfigChartConfig{{
103 + Title: "test", Context: "test", Family: "test", Units: "x",
104 + Dims: []ConfigDimConfig{{Name: "val", Source: "val"}},
105 + }},
106 + }}
107 + },
108 + wantFail: true,
109 + },
110 + "combined metrics and functions succeeds": {
111 + setup: func(c *Collector) {
112 + c.Driver = "pgx"
113 + c.DSN = "postgres://user:pass@localhost/db"
114 + c.Metrics = []ConfigMetricBlock{{
115 + ID: "test",
116 + Mode: "columns",
117 + Query: "SELECT 1 AS val",
118 + Charts: []ConfigChartConfig{{
119 + Title: "test", Context: "test", Family: "test", Units: "x",
120 + Dims: []ConfigDimConfig{{Name: "val", Source: "val"}},
121 + }},
122 + }}
123 + c.Functions = []ConfigFunction{{ID: "func", Query: "SELECT 1"}}
124 + },
125 + wantFail: false,
126 + },
127 + "missing driver fails": {
128 + setup: func(c *Collector) {
129 + c.Driver = ""
130 + c.DSN = "postgres://user:pass@localhost/db"
131 + c.FunctionOnly = true
132 + c.Functions = []ConfigFunction{{ID: "test", Query: "SELECT 1"}}
133 + },
134 + wantFail: true,
135 + },
136 + "missing dsn fails": {
137 + setup: func(c *Collector) {
138 + c.Driver = "pgx"
139 + c.DSN = ""
140 + c.FunctionOnly = true
141 + c.Functions = []ConfigFunction{{ID: "test", Query: "SELECT 1"}}
142 + },
143 + wantFail: true,
144 + },
145 + }
146 +
147 + for name, tc := range tests {
148 + t.Run(name, func(t *testing.T) {
149 + c := New()
150 + tc.setup(c)
151 +
152 + err := c.Init(context.Background())
153 +
154 + if tc.wantFail {
155 + assert.Error(t, err)
156 + } else {
157 + assert.NoError(t, err)
158 + }
159 + })
160 + }
161 +}
162 +
163 +func TestCollector_Check_FunctionOnly(t *testing.T) {
164 + db, mock, err := sqlmock.New()
165 + require.NoError(t, err)
166 + defer func() { _ = db.Close() }()
167 +
168 + c := New()
169 + c.db = db
170 + c.Driver = "pgx"
171 + c.DSN = "postgres://user:pass@localhost/db"
172 + c.FunctionOnly = true
173 + c.Functions = []ConfigFunction{{ID: "test", Query: "SELECT 1"}}
174 +
175 + require.NoError(t, c.Init(context.Background()))
176 +
177 + // Check should succeed without collecting metrics
178 + err = c.Check(context.Background())
179 + assert.NoError(t, err)
180 + assert.NoError(t, mock.ExpectationsWereMet())
181 +}
182 +
183 +func TestCollector_Collect_FunctionOnly(t *testing.T) {
184 + db, _, err := sqlmock.New()
185 + require.NoError(t, err)
186 + defer func() { _ = db.Close() }()
187 +
188 + c := New()
189 + c.db = db
190 + c.Driver = "pgx"
191 + c.DSN = "postgres://user:pass@localhost/db"
192 + c.FunctionOnly = true
193 + c.Functions = []ConfigFunction{{ID: "test", Query: "SELECT 1"}}
194 +
195 + require.NoError(t, c.Init(context.Background()))
196 +
197 + // Collect should return nil in function-only mode
198 + mx := c.Collect(context.Background())
199 + assert.Nil(t, mx)
200 }
201
202 func TestCollector_Cleanup(t *testing.T) {
src/go/plugin/go.d/collector/sql/config.go
+107 -4
@@ -12,8 +12,9 @@ import (
12 )
13
14 type Config struct {
15 - UpdateEvery int `yaml:"update_every,omitempty" json:"update_every"`
16 - AutoDetectionRetry int `yaml:"autodetection_retry,omitempty" json:"autodetection_retry"`
15 + Name string `yaml:"name,omitempty" json:"name,omitempty"`
16 + UpdateEvery int `yaml:"update_every,omitempty" json:"update_every"`
17 + AutoDetectionRetry int `yaml:"autodetection_retry,omitempty" json:"autodetection_retry"`
18
19 Driver string `yaml:"driver" json:"driver"`
20 DSN string `yaml:"dsn" json:"dsn"`
@@ -22,6 +23,50 @@ type Config struct {
23 StaticLabels map[string]string `yaml:"static_labels,omitempty" json:"static_labels"`
24 Queries []ConfigQueryDef `yaml:"queries,omitempty" json:"queries"`
25 Metrics []ConfigMetricBlock `yaml:"metrics,omitempty" json:"metrics"`
26 + Functions []ConfigFunction `yaml:"functions,omitempty" json:"functions,omitempty"`
27 + FunctionOnly bool `yaml:"function_only,omitempty" json:"function_only,omitempty"`
28 +}
29 +
30 +const (
31 + defaultFunctionLimit = 100
32 + maxFunctionLimit = 10000
33 +)
34 +
35 +type ConfigFunction struct {
36 + ID string `yaml:"id" json:"id"`
37 + Name string `yaml:"name,omitempty" json:"name,omitempty"`
38 + Description string `yaml:"description,omitempty" json:"description,omitempty"`
39 + Query string `yaml:"query" json:"query"`
40 + Timeout confopt.Duration `yaml:"timeout,omitempty" json:"timeout,omitempty"`
41 + Limit int `yaml:"limit,omitempty" json:"limit,omitempty"`
42 + DefaultSort string `yaml:"default_sort,omitempty" json:"default_sort,omitempty"`
43 + DefaultSortDesc *bool `yaml:"default_sort_desc,omitempty" json:"default_sort_desc,omitempty"`
44 + Columns map[string]ConfigFuncColumn `yaml:"columns,omitempty" json:"columns,omitempty"`
45 +}
46 +
47 +type ConfigFuncColumn struct {
48 + Type string `yaml:"type,omitempty" json:"type,omitempty"`
49 + Units string `yaml:"units,omitempty" json:"units,omitempty"`
50 + Tooltip string `yaml:"tooltip,omitempty" json:"tooltip,omitempty"`
51 + Visible *bool `yaml:"visible,omitempty" json:"visible,omitempty"`
52 + Sortable *bool `yaml:"sortable,omitempty" json:"sortable,omitempty"`
53 +}
54 +
55 +func (f *ConfigFunction) derivedName() string {
56 + if f.Name != "" {
57 + return f.Name
58 + }
59 + return deriveNameFromID(f.ID)
60 +}
61 +
62 +func deriveNameFromID(id string) string {
63 + words := strings.Split(strings.ReplaceAll(id, "_", "-"), "-")
64 + for i, w := range words {
65 + if len(w) > 0 {
66 + words[i] = strings.ToUpper(w[:1]) + w[1:]
67 + }
68 + }
69 + return strings.Join(words, " ")
70 }
71
72 type (
@@ -86,8 +131,17 @@ func (c *Collector) validateConfig() error {
131 errs = append(errs, errors.New("dsn required"))
132 }
133
89 - if len(c.Metrics) == 0 {
90 - errs = append(errs, errors.New("missing metrics"))
134 + if c.FunctionOnly {
135 + if len(c.Metrics) > 0 {
136 + errs = append(errs, errors.New("function_only is set but metrics are defined"))
137 + }
138 + if len(c.Functions) == 0 {
139 + errs = append(errs, errors.New("function_only is set but no functions defined"))
140 + }
141 + } else {
142 + if len(c.Metrics) == 0 {
143 + errs = append(errs, errors.New("metrics required (or set function_only: true)"))
144 + }
145 }
146
147 queryIdx := map[string]bool{}
@@ -99,9 +153,58 @@ func (c *Collector) validateConfig() error {
153 errs = append(errs, c.Metrics[i].validate(i, queryIdx)...)
154 }
155
156 + funcIdx := map[string]bool{}
157 + for i := range c.Functions {
158 + errs = append(errs, c.Functions[i].validate(i, funcIdx)...)
159 + }
160 +
161 return errors.Join(errs...)
162 }
163
164 +func (f *ConfigFunction) validate(idx int, seen map[string]bool) []error {
165 + var errs []error
166 + fidx := idx + 1
167 +
168 + if f.ID == "" {
169 + errs = append(errs, fmt.Errorf("functions[%d] missing id", fidx))
170 + }
171 + if f.Query == "" {
172 + errs = append(errs, fmt.Errorf("functions[%d] missing query", fidx))
173 + }
174 +
175 + if f.ID != "" {
176 + if strings.Contains(f.ID, ":") {
177 + errs = append(errs, fmt.Errorf("functions[%d] id %q cannot contain ':'", fidx, f.ID))
178 + }
179 + if _, dup := seen[f.ID]; dup {
180 + errs = append(errs, fmt.Errorf("functions[%d] duplicate id %q", fidx, f.ID))
181 + }
182 + seen[f.ID] = true
183 + }
184 +
185 + if f.Limit < 0 {
186 + errs = append(errs, fmt.Errorf("functions[%d] limit cannot be negative", fidx))
187 + }
188 + if f.Limit > maxFunctionLimit {
189 + errs = append(errs, fmt.Errorf("functions[%d] limit exceeds maximum (%d)", fidx, maxFunctionLimit))
190 + }
191 + if f.Timeout.Duration() < 0 {
192 + errs = append(errs, fmt.Errorf("functions[%d] timeout cannot be negative", fidx))
193 + }
194 +
195 + validTypes := map[string]bool{
196 + "string": true, "integer": true, "float": true,
197 + "boolean": true, "duration": true, "timestamp": true,
198 + }
199 + for colName, col := range f.Columns {
200 + if col.Type != "" && !validTypes[col.Type] {
201 + errs = append(errs, fmt.Errorf("functions[%d] column %q invalid type %q", fidx, colName, col.Type))
202 + }
203 + }
204 +
205 + return errs
206 +}
207 +
208 // ---- Per-struct validation helpers ----
209
210 func (q *ConfigQueryDef) validate(idx int, seen map[string]bool) []error {
src/go/plugin/go.d/collector/sql/config_schema.json
+126
@@ -290,6 +290,107 @@
290 }
291 }
292 },
293 + "function_only": {
294 + "title": "Function Only",
295 + "description": "Set to true if this job only provides functions (no metrics). When enabled, metrics configuration is ignored and no charts are created.",
296 + "type": "boolean",
297 + "default": false
298 + },
299 + "functions": {
300 + "title": "Functions",
301 + "description": "SQL functions that expose query results as table views in the Netdata UI.",
302 + "type": "array",
303 + "items": {
304 + "title": "Function",
305 + "type": "object",
306 + "properties": {
307 + "id": {
308 + "title": "ID",
309 + "description": "Unique identifier for this function.",
310 + "type": "string"
311 + },
312 + "name": {
313 + "title": "Name",
314 + "description": "Display name shown in the UI. Auto-derived from ID if empty.",
315 + "type": "string"
316 + },
317 + "description": {
318 + "title": "Description",
319 + "description": "Help text shown in the UI.",
320 + "type": "string"
321 + },
322 + "query": {
323 + "title": "Query",
324 + "description": "SQL query to execute when this function is called.",
325 + "type": "string"
326 + },
327 + "timeout": {
328 + "title": "Timeout",
329 + "description": "Query timeout in seconds. Uses collector default if not set.",
330 + "type": "number",
331 + "minimum": 0
332 + },
333 + "limit": {
334 + "title": "Row Limit",
335 + "description": "Maximum rows to return.",
336 + "type": "integer",
337 + "minimum": 1,
338 + "maximum": 10000,
339 + "default": 100
340 + },
341 + "default_sort": {
342 + "title": "Default Sort Column",
343 + "description": "Column name for initial sort order.",
344 + "type": "string"
345 + },
346 + "default_sort_desc": {
347 + "title": "Sort Descending",
348 + "description": "Sort in descending order by default.",
349 + "type": "boolean",
350 + "default": true
351 + },
352 + "columns": {
353 + "title": "Column Overrides",
354 + "description": "Override auto-detected column metadata.",
355 + "type": "object",
356 + "additionalProperties": {
357 + "type": "object",
358 + "properties": {
359 + "type": {
360 + "title": "Type",
361 + "description": "Column data type.",
362 + "type": "string",
363 + "enum": ["string", "integer", "float", "boolean", "duration", "timestamp"]
364 + },
365 + "units": {
366 + "title": "Units",
367 + "description": "Unit label for the column (e.g., milliseconds, bytes).",
368 + "type": "string"
369 + },
370 + "tooltip": {
371 + "title": "Tooltip",
372 + "description": "Hover text shown in the UI.",
373 + "type": "string"
374 + },
375 + "visible": {
376 + "title": "Visible",
377 + "description": "Show column by default.",
378 + "type": "boolean",
379 + "default": true
380 + },
381 + "sortable": {
382 + "title": "Sortable",
383 + "description": "Allow sorting by this column.",
384 + "type": "boolean",
385 + "default": true
386 + }
387 + }
388 + }
389 + }
390 + },
391 + "required": ["id", "query"]
392 + }
393 + },
394 "vnode": {
395 "title": "Vnode",
396 "description": "Name of the Virtual Node this job should send its metrics to. Leave empty to use the local node.",
@@ -336,6 +437,13 @@
437 "fields": [
438 "static_labels"
439 ]
440 + },
441 + {
442 + "title": "Functions",
443 + "fields": [
444 + "function_only",
445 + "functions"
446 + ]
447 }
448 ]
449 },
@@ -396,6 +504,24 @@
504 }
505 }
506 }
507 + },
508 + "functions": {
509 + "items": {
510 + "ui:order": [
511 + "id",
512 + "name",
513 + "description",
514 + "query",
515 + "timeout",
516 + "limit",
517 + "default_sort",
518 + "default_sort_desc",
519 + "columns"
520 + ],
521 + "query": {
522 + "ui:widget": "textarea"
523 + }
524 + }
525 }
526 }
527 }
src/go/plugin/go.d/collector/sql/func_table.go new
+412
@@ -0,0 +1,412 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package sql
4 +
5 +import (
6 + "context"
7 + "database/sql"
8 + "errors"
9 + "strings"
10 + "time"
11 +
12 + "github.com/netdata/netdata/go/plugins/pkg/funcapi"
13 + "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/module"
14 +)
15 +
16 +type funcTable struct {
17 + collector *Collector
18 +}
19 +
20 +func newFuncTable(c *Collector) *funcTable {
21 + return &funcTable{collector: c}
22 +}
23 +
24 +var _ funcapi.MethodHandler = (*funcTable)(nil)
25 +
26 +// sqlJobMethods returns method configs for a specific SQL job.
27 +// Each configured function becomes a separate method: "jobName:functionID"
28 +// This results in functions like "sql:postgres_test:active-queries"
29 +func sqlJobMethods(job *module.Job) []funcapi.MethodConfig {
30 + c, ok := job.Module().(*Collector)
31 + if !ok || len(c.Config.Functions) == 0 {
32 + return nil
33 + }
34 +
35 + methods := make([]funcapi.MethodConfig, 0, len(c.Config.Functions))
36 + for _, fn := range c.Config.Functions {
37 + // Method ID format: "jobName:functionID" (e.g., "postgres_test:active-queries")
38 + // Full function name will be: "sql:postgres_test:active-queries"
39 + methodID := job.Name() + ":" + fn.ID
40 +
41 + methodName := fn.derivedName()
42 + help := fn.Description
43 + if help == "" {
44 + help = "Execute SQL query: " + fn.ID
45 + }
46 +
47 + methods = append(methods, funcapi.MethodConfig{
48 + ID: methodID,
49 + Name: methodName,
50 + Help: help,
51 + UpdateEvery: 10,
52 + })
53 + }
54 +
55 + return methods
56 +}
57 +
58 +func sqlMethodHandler(job *module.Job) funcapi.MethodHandler {
59 + c, ok := job.Module().(*Collector)
60 + if !ok {
61 + return nil
62 + }
63 + return c.funcTable
64 +}
65 +
66 +func (f *funcTable) Cleanup(context.Context) {
67 + // No-op: DB connection managed by collector
68 +}
69 +
70 +func (f *funcTable) MethodParams(ctx context.Context, method string) ([]funcapi.ParamConfig, error) {
71 + // Each function is now a separate method endpoint, no __function selector needed
72 + return nil, nil
73 +}
74 +
75 +func (f *funcTable) Handle(ctx context.Context, method string, params funcapi.ResolvedParams) *funcapi.FunctionResponse {
76 + // Check if collector is shutting down (fast path)
77 + if f.collector.dbCtx != nil && f.collector.dbCtx.Err() != nil {
78 + return funcapi.ErrorResponse(503, "collector is shutting down")
79 + }
80 +
81 + // Acquire read lock to prevent DB close during query
82 + f.collector.dbMu.RLock()
83 + defer f.collector.dbMu.RUnlock()
84 +
85 + if f.collector.db == nil {
86 + return funcapi.ErrorResponse(503, "database connection not initialized")
87 + }
88 +
89 + // Method format is "jobName:functionID" (e.g., "postgres_test:active-queries")
90 + // Extract the functionID part after the last colon
91 + functionID := method
92 + if idx := strings.LastIndex(method, ":"); idx != -1 {
93 + functionID = method[idx+1:]
94 + }
95 +
96 + funcCfg := f.findFunction(functionID)
97 + if funcCfg == nil {
98 + return funcapi.ErrorResponse(404, "unknown function: %s", functionID)
99 + }
100 +
101 + // Merge request context with collector's shutdown context
102 + // Query cancels if either request times out OR collector shuts down
103 + queryCtx, queryCancel := context.WithCancel(ctx)
104 + defer queryCancel()
105 +
106 + if f.collector.dbCtx != nil {
107 + stop := context.AfterFunc(f.collector.dbCtx, queryCancel)
108 + defer stop()
109 + }
110 +
111 + return f.executeFunction(queryCtx, funcCfg)
112 +}
113 +
114 +func (f *funcTable) findFunction(id string) *ConfigFunction {
115 + for i := range f.collector.Config.Functions {
116 + if f.collector.Config.Functions[i].ID == id {
117 + return &f.collector.Config.Functions[i]
118 + }
119 + }
120 + return nil
121 +}
122 +
123 +func (f *funcTable) executeFunction(ctx context.Context, cfg *ConfigFunction) *funcapi.FunctionResponse {
124 + timeout := f.collector.Timeout.Duration()
125 + if cfg.Timeout.Duration() > 0 {
126 + timeout = cfg.Timeout.Duration()
127 + }
128 + queryCtx, cancel := context.WithTimeout(ctx, timeout)
129 + defer cancel()
130 +
131 + rows, err := f.collector.db.QueryContext(queryCtx, cfg.Query)
132 + if err != nil {
133 + if errors.Is(queryCtx.Err(), context.DeadlineExceeded) {
134 + return funcapi.ErrorResponse(504, "query timeout after %v", timeout)
135 + }
136 + return funcapi.ErrorResponse(500, "query failed: %v", err)
137 + }
138 + defer rows.Close()
139 +
140 + colTypes, err := rows.ColumnTypes()
141 + if err != nil {
142 + return funcapi.ErrorResponse(500, "failed to get column types: %v", err)
143 + }
144 +
145 + sortDesc := cfg.DefaultSortDesc == nil || *cfg.DefaultSortDesc
146 + columns := f.buildColumnMetadata(colTypes, cfg.Columns, cfg.DefaultSort, sortDesc)
147 +
148 + limit := cfg.Limit
149 + if limit <= 0 {
150 + limit = defaultFunctionLimit
151 + }
152 + if limit > maxFunctionLimit {
153 + limit = maxFunctionLimit
154 + }
155 +
156 + data := make([][]any, 0, limit)
157 + for rows.Next() && len(data) < limit {
158 + row, err := f.scanRow(rows, len(colTypes))
159 + if err != nil {
160 + f.collector.Warningf("scan row failed: %v", err)
161 + continue
162 + }
163 + data = append(data, row)
164 + }
165 +
166 + if err := rows.Err(); err != nil {
167 + if errors.Is(queryCtx.Err(), context.DeadlineExceeded) {
168 + return funcapi.ErrorResponse(504, "query timeout during iteration")
169 + }
170 + return funcapi.ErrorResponse(500, "row iteration failed: %v", err)
171 + }
172 +
173 + defaultSort := cfg.DefaultSort
174 + if defaultSort != "" {
175 + found := false
176 + for _, ct := range colTypes {
177 + if ct.Name() == defaultSort {
178 + found = true
179 + break
180 + }
181 + }
182 + if !found {
183 + f.collector.Warningf("function %q: default_sort column %q not in query results", cfg.ID, defaultSort)
184 + defaultSort = ""
185 + }
186 + }
187 +
188 + return &funcapi.FunctionResponse{
189 + Status: 200,
190 + Help: cfg.Description,
191 + Columns: columns,
192 + Data: data,
193 + DefaultSortColumn: defaultSort,
194 + }
195 +}
196 +
197 +func (f *funcTable) scanRow(rows *sql.Rows, numCols int) ([]any, error) {
198 + values := make([]any, numCols)
199 + ptrs := make([]any, numCols)
200 + for i := range values {
201 + ptrs[i] = &values[i]
202 + }
203 + if err := rows.Scan(ptrs...); err != nil {
204 + return nil, err
205 + }
206 +
207 + for i, v := range values {
208 + values[i] = normalizeValue(v)
209 + }
210 + return values, nil
211 +}
212 +
213 +func normalizeValue(v any) any {
214 + if v == nil {
215 + return nil
216 + }
217 + switch val := v.(type) {
218 + case []byte:
219 + return string(val)
220 + case time.Time:
221 + return val.UnixMilli()
222 + case int:
223 + return int64(val)
224 + case int32:
225 + return int64(val)
226 + case float32:
227 + return float64(val)
228 + default:
229 + return v
230 + }
231 +}
232 +
233 +// typeMapping maps database type names to funcapi field types.
234 +// Covers MySQL, PostgreSQL (pgx), SQL Server, and Oracle drivers.
235 +var typeMapping = map[string]funcapi.FieldType{
236 + // MySQL (uppercase)
237 + "INT": funcapi.FieldTypeInteger,
238 + "BIGINT": funcapi.FieldTypeInteger,
239 + "TINYINT": funcapi.FieldTypeInteger,
240 + "SMALLINT": funcapi.FieldTypeInteger,
241 + "MEDIUMINT": funcapi.FieldTypeInteger,
242 + "FLOAT": funcapi.FieldTypeFloat,
243 + "DOUBLE": funcapi.FieldTypeFloat,
244 + "DECIMAL": funcapi.FieldTypeFloat,
245 + "VARCHAR": funcapi.FieldTypeString,
246 + "CHAR": funcapi.FieldTypeString,
247 + "TEXT": funcapi.FieldTypeString,
248 + "DATETIME": funcapi.FieldTypeTimestamp,
249 + "TIMESTAMP": funcapi.FieldTypeTimestamp,
250 + "DATE": funcapi.FieldTypeTimestamp,
251 + "TIME": funcapi.FieldTypeDuration,
252 +
253 + // PostgreSQL (pgx) - lowercase
254 + "int2": funcapi.FieldTypeInteger,
255 + "int4": funcapi.FieldTypeInteger,
256 + "int8": funcapi.FieldTypeInteger,
257 + "smallint": funcapi.FieldTypeInteger,
258 + "integer": funcapi.FieldTypeInteger,
259 + "bigint": funcapi.FieldTypeInteger,
260 + "float4": funcapi.FieldTypeFloat,
261 + "float8": funcapi.FieldTypeFloat,
262 + "numeric": funcapi.FieldTypeFloat,
263 + "decimal": funcapi.FieldTypeFloat,
264 + "varchar": funcapi.FieldTypeString,
265 + "char": funcapi.FieldTypeString,
266 + "text": funcapi.FieldTypeString,
267 + "bpchar": funcapi.FieldTypeString,
268 + "timestamp": funcapi.FieldTypeTimestamp,
269 + "timestamptz": funcapi.FieldTypeTimestamp,
270 + "date": funcapi.FieldTypeTimestamp,
271 + "bool": funcapi.FieldTypeBoolean,
272 + "boolean": funcapi.FieldTypeBoolean,
273 + "interval": funcapi.FieldTypeDuration,
274 +
275 + // SQL Server
276 + "NVARCHAR": funcapi.FieldTypeString,
277 + "NCHAR": funcapi.FieldTypeString,
278 + "DATETIME2": funcapi.FieldTypeTimestamp,
279 + "BIT": funcapi.FieldTypeBoolean,
280 + "REAL": funcapi.FieldTypeFloat,
281 +
282 + // Oracle
283 + "VARCHAR2": funcapi.FieldTypeString,
284 + "NVARCHAR2": funcapi.FieldTypeString,
285 + "CLOB": funcapi.FieldTypeString,
286 + "NUMBER": funcapi.FieldTypeFloat, // Could be int, safer as float
287 + "BINARY_FLOAT": funcapi.FieldTypeFloat,
288 + "BINARY_DOUBLE": funcapi.FieldTypeFloat,
289 +}
290 +
291 +func inferType(dbTypeName string) funcapi.FieldType {
292 + if t, ok := typeMapping[dbTypeName]; ok {
293 + return t
294 + }
295 + if t, ok := typeMapping[strings.ToUpper(dbTypeName)]; ok {
296 + return t
297 + }
298 + if t, ok := typeMapping[strings.ToLower(dbTypeName)]; ok {
299 + return t
300 + }
301 + return funcapi.FieldTypeString
302 +}
303 +
304 +func parseFieldType(s string) funcapi.FieldType {
305 + switch strings.ToLower(s) {
306 + case "integer":
307 + return funcapi.FieldTypeInteger
308 + case "float":
309 + return funcapi.FieldTypeFloat
310 + case "boolean":
311 + return funcapi.FieldTypeBoolean
312 + case "duration":
313 + return funcapi.FieldTypeDuration
314 + case "timestamp":
315 + return funcapi.FieldTypeTimestamp
316 + default:
317 + return funcapi.FieldTypeString
318 + }
319 +}
320 +
321 +func deriveTransform(fieldType funcapi.FieldType) funcapi.FieldTransform {
322 + switch fieldType {
323 + case funcapi.FieldTypeInteger, funcapi.FieldTypeFloat:
324 + return funcapi.FieldTransformNumber
325 + case funcapi.FieldTypeDuration:
326 + return funcapi.FieldTransformDuration
327 + case funcapi.FieldTypeTimestamp:
328 + return funcapi.FieldTransformDatetime
329 + default:
330 + return funcapi.FieldTransformNone
331 + }
332 +}
333 +
334 +func deriveFilterSummary(fieldType funcapi.FieldType) (funcapi.FieldFilter, funcapi.FieldSummary) {
335 + switch fieldType {
336 + case funcapi.FieldTypeInteger, funcapi.FieldTypeFloat, funcapi.FieldTypeDuration:
337 + return funcapi.FieldFilterRange, funcapi.FieldSummarySum
338 + case funcapi.FieldTypeTimestamp:
339 + return funcapi.FieldFilterRange, funcapi.FieldSummaryMax
340 + case funcapi.FieldTypeBoolean:
341 + return funcapi.FieldFilterMultiselect, funcapi.FieldSummaryCount
342 + default:
343 + return funcapi.FieldFilterMultiselect, funcapi.FieldSummaryCount
344 + }
345 +}
346 +
347 +// buildColumnMetadata constructs column definitions from SQL query results.
348 +//
349 +// This uses funcapi.Column directly instead of funcapi.ColumnMeta because:
350 +// - Columns are discovered dynamically at runtime from query results
351 +// - There are no static, pre-defined columns to embed ColumnMeta into
352 +// - ColumnMeta is designed for collectors with static columns and custom Value extractors
353 +func (f *funcTable) buildColumnMetadata(colTypes []*sql.ColumnType, overrides map[string]ConfigFuncColumn, defaultSort string, sortDesc bool) map[string]any {
354 + columns := make(map[string]any, len(colTypes))
355 +
356 + for i, ct := range colTypes {
357 + colName := ct.Name()
358 +
359 + fieldType := inferType(ct.DatabaseTypeName())
360 +
361 + override, hasOverride := overrides[colName]
362 + if hasOverride && override.Type != "" {
363 + fieldType = parseFieldType(override.Type)
364 + }
365 +
366 + transform := deriveTransform(fieldType)
367 + filter, summary := deriveFilterSummary(fieldType)
368 +
369 + visible := true
370 + if hasOverride && override.Visible != nil {
371 + visible = *override.Visible
372 + }
373 +
374 + sortable := true
375 + if hasOverride && override.Sortable != nil {
376 + sortable = *override.Sortable
377 + }
378 +
379 + sort := funcapi.FieldSortAscending
380 + if colName == defaultSort && sortDesc {
381 + sort = funcapi.FieldSortDescending
382 + }
383 +
384 + units := ""
385 + if hasOverride {
386 + units = override.Units
387 + }
388 +
389 + tooltip := colName
390 + if hasOverride && override.Tooltip != "" {
391 + tooltip = override.Tooltip
392 + }
393 +
394 + col := funcapi.Column{
395 + Index: i,
396 + Name: tooltip,
397 + Type: fieldType,
398 + Units: units,
399 + Visible: visible,
400 + Sortable: sortable,
401 + Sort: sort,
402 + Filter: filter,
403 + Summary: summary,
404 + ValueOptions: funcapi.ValueOptions{
405 + Transform: transform,
406 + },
407 + }
408 + columns[colName] = col.BuildColumn()
409 + }
410 +
411 + return columns
412 +}
src/go/plugin/go.d/collector/sql/func_table_test.go new
+593
@@ -0,0 +1,593 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package sql
4 +
5 +import (
6 + "context"
7 + "database/sql"
8 + "io"
9 + "testing"
10 + "time"
11 +
12 + "github.com/DATA-DOG/go-sqlmock"
13 + "github.com/stretchr/testify/assert"
14 + "github.com/stretchr/testify/require"
15 +
16 + "github.com/netdata/netdata/go/plugins/pkg/confopt"
17 + "github.com/netdata/netdata/go/plugins/pkg/funcapi"
18 + "github.com/netdata/netdata/go/plugins/plugin/go.d/agent/module"
19 +)
20 +
21 +func TestNormalizeValue(t *testing.T) {
22 + tests := []struct {
23 + name string
24 + input any
25 + expected any
26 + }{
27 + {"nil", nil, nil},
28 + {"string", "hello", "hello"},
29 + {"int64", int64(42), int64(42)},
30 + {"int", int(42), int64(42)},
31 + {"int32", int32(42), int64(42)},
32 + {"float64", float64(3.14), float64(3.14)},
33 + {"float32", float32(3.14), float64(float32(3.14))},
34 + {"bytes", []byte("hello"), "hello"},
35 + {"bool", true, true},
36 + }
37 +
38 + for _, tc := range tests {
39 + t.Run(tc.name, func(t *testing.T) {
40 + result := normalizeValue(tc.input)
41 + assert.Equal(t, tc.expected, result)
42 + })
43 + }
44 +}
45 +
46 +func TestNormalizeValue_Time(t *testing.T) {
47 + ts := time.Date(2024, 1, 15, 10, 30, 0, 0, time.UTC)
48 + result := normalizeValue(ts)
49 + assert.Equal(t, ts.UnixMilli(), result)
50 +}
51 +
52 +func TestConfigFunction_Validate(t *testing.T) {
53 + tests := []struct {
54 + name string
55 + cfg ConfigFunction
56 + expectError bool
57 + errorCount int
58 + }{
59 + {
60 + name: "valid config",
61 + cfg: ConfigFunction{ID: "test", Query: "SELECT 1"},
62 + expectError: false,
63 + },
64 + {
65 + name: "valid config with limit",
66 + cfg: ConfigFunction{ID: "test", Query: "SELECT 1", Limit: 500},
67 + expectError: false,
68 + },
69 + {
70 + name: "missing id",
71 + cfg: ConfigFunction{Query: "SELECT 1"},
72 + expectError: true,
73 + errorCount: 1,
74 + },
75 + {
76 + name: "missing query",
77 + cfg: ConfigFunction{ID: "test"},
78 + expectError: true,
79 + errorCount: 1,
80 + },
81 + {
82 + name: "missing both",
83 + cfg: ConfigFunction{},
84 + expectError: true,
85 + errorCount: 2,
86 + },
87 + {
88 + name: "id contains colon",
89 + cfg: ConfigFunction{ID: "test:query", Query: "SELECT 1"},
90 + expectError: true,
91 + errorCount: 1,
92 + },
93 + {
94 + name: "negative limit",
95 + cfg: ConfigFunction{ID: "test", Query: "SELECT 1", Limit: -1},
96 + expectError: true,
97 + errorCount: 1,
98 + },
99 + {
100 + name: "limit exceeds maximum",
101 + cfg: ConfigFunction{ID: "test", Query: "SELECT 1", Limit: 10001},
102 + expectError: true,
103 + errorCount: 1,
104 + },
105 + {
106 + name: "negative timeout",
107 + cfg: ConfigFunction{ID: "test", Query: "SELECT 1", Timeout: confopt.Duration(-time.Second)},
108 + expectError: true,
109 + errorCount: 1,
110 + },
111 + {
112 + name: "invalid column type",
113 + cfg: ConfigFunction{
114 + ID: "test",
115 + Query: "SELECT 1",
116 + Columns: map[string]ConfigFuncColumn{"col1": {Type: "invalid"}},
117 + },
118 + expectError: true,
119 + errorCount: 1,
120 + },
121 + {
122 + name: "valid column types",
123 + cfg: ConfigFunction{
124 + ID: "test",
125 + Query: "SELECT 1",
126 + Columns: map[string]ConfigFuncColumn{
127 + "col1": {Type: "string"},
128 + "col2": {Type: "integer"},
129 + "col3": {Type: "float"},
130 + "col4": {Type: "boolean"},
131 + "col5": {Type: "duration"},
132 + "col6": {Type: "timestamp"},
133 + },
134 + },
135 + expectError: false,
136 + },
137 + }
138 +
139 + for _, tc := range tests {
140 + t.Run(tc.name, func(t *testing.T) {
141 + seen := make(map[string]bool)
142 + errs := tc.cfg.validate(0, seen)
143 + if tc.expectError {
144 + assert.Len(t, errs, tc.errorCount)
145 + } else {
146 + assert.Empty(t, errs)
147 + }
148 + })
149 + }
150 +}
151 +
152 +func TestConfigFunction_Validate_DuplicateID(t *testing.T) {
153 + seen := map[string]bool{"existing": true}
154 + cfg := ConfigFunction{ID: "existing", Query: "SELECT 1"}
155 +
156 + errs := cfg.validate(0, seen)
157 + assert.Len(t, errs, 1)
158 + assert.Contains(t, errs[0].Error(), "duplicate id")
159 +}
160 +
161 +func TestFuncTable_FindFunction(t *testing.T) {
162 + c := &Collector{}
163 + c.Config.Functions = []ConfigFunction{
164 + {ID: "func1", Query: "SELECT 1"},
165 + {ID: "func2", Query: "SELECT 2"},
166 + }
167 +
168 + ft := &funcTable{collector: c}
169 +
170 + // Found
171 + f := ft.findFunction("func1")
172 + assert.NotNil(t, f)
173 + assert.Equal(t, "func1", f.ID)
174 +
175 + f = ft.findFunction("func2")
176 + assert.NotNil(t, f)
177 + assert.Equal(t, "func2", f.ID)
178 +
179 + // Not found
180 + f = ft.findFunction("nonexistent")
181 + assert.Nil(t, f)
182 +}
183 +
184 +func TestSqlJobMethods(t *testing.T) {
185 + tests := map[string]struct {
186 + setupCollector func() *Collector
187 + jobName string
188 + wantLen int
189 + wantMethodIDs []string
190 + }{
191 + "collector with single function": {
192 + setupCollector: func() *Collector {
193 + c := New()
194 + c.Config.Functions = []ConfigFunction{
195 + {ID: "test-query", Query: "SELECT 1", Description: "Test query"},
196 + }
197 + return c
198 + },
199 + jobName: "postgres_test",
200 + wantLen: 1,
201 + wantMethodIDs: []string{"postgres_test:test-query"},
202 + },
203 + "collector with multiple functions": {
204 + setupCollector: func() *Collector {
205 + c := New()
206 + c.Config.Functions = []ConfigFunction{
207 + {ID: "active-queries", Query: "SELECT 1"},
208 + {ID: "databases", Query: "SELECT 2"},
209 + {ID: "roles", Query: "SELECT 3"},
210 + }
211 + return c
212 + },
213 + jobName: "pg_main",
214 + wantLen: 3,
215 + wantMethodIDs: []string{
216 + "pg_main:active-queries",
217 + "pg_main:databases",
218 + "pg_main:roles",
219 + },
220 + },
221 + "collector without functions": {
222 + setupCollector: func() *Collector {
223 + return New()
224 + },
225 + jobName: "empty_job",
226 + wantLen: 0,
227 + },
228 + }
229 +
230 + for name, tc := range tests {
231 + t.Run(name, func(t *testing.T) {
232 + c := tc.setupCollector()
233 + job := module.NewJob(module.JobConfig{
234 + Name: tc.jobName,
235 + ModuleName: "sql",
236 + FullName: "sql_" + tc.jobName,
237 + Module: c,
238 + Out: io.Discard,
239 + })
240 + methods := sqlJobMethods(job)
241 +
242 + assert.Len(t, methods, tc.wantLen)
243 + for i, wantID := range tc.wantMethodIDs {
244 + assert.Equal(t, wantID, methods[i].ID)
245 + assert.Equal(t, 10, methods[i].UpdateEvery)
246 + }
247 + })
248 + }
249 +}
250 +
251 +func TestFuncTable_MethodParams(t *testing.T) {
252 + // MethodParams now always returns nil since each function is a separate endpoint
253 + c := New()
254 + c.db, _, _ = sqlmock.New()
255 + c.Config.Functions = []ConfigFunction{
256 + {ID: "func1", Query: "SELECT 1"},
257 + {ID: "func2", Query: "SELECT 2"},
258 + }
259 + defer func() { _ = c.db.Close() }()
260 +
261 + ft := &funcTable{collector: c}
262 + params, err := ft.MethodParams(context.Background(), "postgres_test:func1")
263 +
264 + require.NoError(t, err)
265 + assert.Nil(t, params)
266 +}
267 +
268 +func TestFuncTable_Handle(t *testing.T) {
269 + tests := map[string]struct {
270 + functions []ConfigFunction
271 + functionID string
272 + dbNil bool
273 + prepareMock func(sqlmock.Sqlmock)
274 + checkResp func(*testing.T, *funcapi.FunctionResponse)
275 + }{
276 + "db not initialized": {
277 + functions: []ConfigFunction{{ID: "test", Query: "SELECT 1"}},
278 + functionID: "test",
279 + dbNil: true,
280 + checkResp: func(t *testing.T, resp *funcapi.FunctionResponse) {
281 + assert.Equal(t, 503, resp.Status)
282 + assert.Contains(t, resp.Message, "not initialized")
283 + },
284 + },
285 + "unknown function": {
286 + functions: []ConfigFunction{{ID: "known", Query: "SELECT 1"}},
287 + functionID: "unknown",
288 + checkResp: func(t *testing.T, resp *funcapi.FunctionResponse) {
289 + assert.Equal(t, 404, resp.Status)
290 + assert.Contains(t, resp.Message, "unknown function")
291 + },
292 + },
293 + "successful query": {
294 + functions: []ConfigFunction{
295 + {ID: "test", Query: "SELECT id, name FROM users", Description: "Test query"},
296 + },
297 + functionID: "test",
298 + prepareMock: func(m sqlmock.Sqlmock) {
299 + rows := sqlmock.NewRows([]string{"id", "name"}).
300 + AddRow(1, "Alice").
301 + AddRow(2, "Bob")
302 + m.ExpectQuery("SELECT id, name FROM users").WillReturnRows(rows)
303 + },
304 + checkResp: func(t *testing.T, resp *funcapi.FunctionResponse) {
305 + assert.Equal(t, 200, resp.Status)
306 + assert.Equal(t, "Test query", resp.Help)
307 + assert.Len(t, resp.Data, 2)
308 + },
309 + },
310 + "query error": {
311 + functions: []ConfigFunction{{ID: "test", Query: "SELECT 1"}},
312 + functionID: "test",
313 + prepareMock: func(m sqlmock.Sqlmock) {
314 + m.ExpectQuery("SELECT 1").WillReturnError(assert.AnError)
315 + },
316 + checkResp: func(t *testing.T, resp *funcapi.FunctionResponse) {
317 + assert.Equal(t, 500, resp.Status)
318 + assert.Contains(t, resp.Message, "query failed")
319 + },
320 + },
321 + "limit applied": {
322 + functions: []ConfigFunction{
323 + {ID: "test", Query: "SELECT n", Limit: 2},
324 + },
325 + functionID: "test",
326 + prepareMock: func(m sqlmock.Sqlmock) {
327 + rows := sqlmock.NewRows([]string{"n"}).
328 + AddRow(1).AddRow(2).AddRow(3).AddRow(4).AddRow(5)
329 + m.ExpectQuery("SELECT n").WillReturnRows(rows)
330 + },
331 + checkResp: func(t *testing.T, resp *funcapi.FunctionResponse) {
332 + assert.Equal(t, 200, resp.Status)
333 + assert.Len(t, resp.Data, 2) // limited to 2
334 + },
335 + },
336 + "default limit applied": {
337 + functions: []ConfigFunction{
338 + {ID: "test", Query: "SELECT n"}, // no limit set
339 + },
340 + functionID: "test",
341 + prepareMock: func(m sqlmock.Sqlmock) {
342 + rows := sqlmock.NewRows([]string{"n"})
343 + for i := 0; i < 150; i++ {
344 + rows.AddRow(i)
345 + }
346 + m.ExpectQuery("SELECT n").WillReturnRows(rows)
347 + },
348 + checkResp: func(t *testing.T, resp *funcapi.FunctionResponse) {
349 + assert.Equal(t, 200, resp.Status)
350 + assert.Len(t, resp.Data, defaultFunctionLimit) // 100
351 + },
352 + },
353 + "default_sort valid column": {
354 + functions: []ConfigFunction{
355 + {ID: "test", Query: "SELECT id, name", DefaultSort: "id"},
356 + },
357 + functionID: "test",
358 + prepareMock: func(m sqlmock.Sqlmock) {
359 + rows := sqlmock.NewRows([]string{"id", "name"}).AddRow(1, "test")
360 + m.ExpectQuery("SELECT id, name").WillReturnRows(rows)
361 + },
362 + checkResp: func(t *testing.T, resp *funcapi.FunctionResponse) {
363 + assert.Equal(t, 200, resp.Status)
364 + assert.Equal(t, "id", resp.DefaultSortColumn)
365 + },
366 + },
367 + "default_sort invalid column": {
368 + functions: []ConfigFunction{
369 + {ID: "test", Query: "SELECT id, name", DefaultSort: "nonexistent"},
370 + },
371 + functionID: "test",
372 + prepareMock: func(m sqlmock.Sqlmock) {
373 + rows := sqlmock.NewRows([]string{"id", "name"}).AddRow(1, "test")
374 + m.ExpectQuery("SELECT id, name").WillReturnRows(rows)
375 + },
376 + checkResp: func(t *testing.T, resp *funcapi.FunctionResponse) {
377 + assert.Equal(t, 200, resp.Status)
378 + assert.Equal(t, "", resp.DefaultSortColumn) // cleared because invalid
379 + },
380 + },
381 + }
382 +
383 + for name, tc := range tests {
384 + t.Run(name, func(t *testing.T) {
385 + c := New()
386 + c.Config.Functions = tc.functions
387 +
388 + var mock sqlmock.Sqlmock
389 + if !tc.dbNil {
390 + var db *sql.DB
391 + var err error
392 + db, mock, err = sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherEqual))
393 + require.NoError(t, err)
394 + defer func() { _ = db.Close() }()
395 + c.db = db
396 +
397 + if tc.prepareMock != nil {
398 + tc.prepareMock(mock)
399 + }
400 + }
401 +
402 + ft := &funcTable{collector: c}
403 +
404 + // Method format is "jobName:functionID" - function ID is extracted from method
405 + method := "test_job:" + tc.functionID
406 + resp := ft.Handle(context.Background(), method, nil)
407 +
408 + tc.checkResp(t, resp)
409 + if mock != nil {
410 + assert.NoError(t, mock.ExpectationsWereMet())
411 + }
412 + })
413 + }
414 +}
415 +
416 +func TestInferType(t *testing.T) {
417 + tests := []struct {
418 + dbType string
419 + expected funcapi.FieldType
420 + }{
421 + // MySQL (uppercase)
422 + {"INT", funcapi.FieldTypeInteger},
423 + {"BIGINT", funcapi.FieldTypeInteger},
424 + {"VARCHAR", funcapi.FieldTypeString},
425 + {"DATETIME", funcapi.FieldTypeTimestamp},
426 + {"TIME", funcapi.FieldTypeDuration},
427 + {"FLOAT", funcapi.FieldTypeFloat},
428 + {"DOUBLE", funcapi.FieldTypeFloat},
429 + {"TEXT", funcapi.FieldTypeString},
430 +
431 + // PostgreSQL (lowercase)
432 + {"int4", funcapi.FieldTypeInteger},
433 + {"int8", funcapi.FieldTypeInteger},
434 + {"varchar", funcapi.FieldTypeString},
435 + {"text", funcapi.FieldTypeString},
436 + {"timestamp", funcapi.FieldTypeTimestamp},
437 + {"timestamptz", funcapi.FieldTypeTimestamp},
438 + {"bool", funcapi.FieldTypeBoolean},
439 + {"boolean", funcapi.FieldTypeBoolean},
440 + {"float8", funcapi.FieldTypeFloat},
441 + {"numeric", funcapi.FieldTypeFloat},
442 + {"interval", funcapi.FieldTypeDuration},
443 +
444 + // SQL Server
445 + {"NVARCHAR", funcapi.FieldTypeString},
446 + {"DATETIME2", funcapi.FieldTypeTimestamp},
447 + {"BIT", funcapi.FieldTypeBoolean},
448 + {"REAL", funcapi.FieldTypeFloat},
449 +
450 + // Oracle
451 + {"VARCHAR2", funcapi.FieldTypeString},
452 + {"NUMBER", funcapi.FieldTypeFloat},
453 + {"BINARY_DOUBLE", funcapi.FieldTypeFloat},
454 +
455 + // Case insensitive
456 + {"int", funcapi.FieldTypeInteger},
457 + {"Int", funcapi.FieldTypeInteger},
458 + {"VARCHAR", funcapi.FieldTypeString},
459 + {"Varchar", funcapi.FieldTypeString},
460 +
461 + // Unknown -> string
462 + {"UNKNOWN_TYPE", funcapi.FieldTypeString},
463 + {"custom", funcapi.FieldTypeString},
464 + {"", funcapi.FieldTypeString},
465 + }
466 +
467 + for _, tc := range tests {
468 + t.Run(tc.dbType, func(t *testing.T) {
469 + result := inferType(tc.dbType)
470 + assert.Equal(t, tc.expected, result)
471 + })
472 + }
473 +}
474 +
475 +func TestParseFieldType(t *testing.T) {
476 + tests := []struct {
477 + input string
478 + expected funcapi.FieldType
479 + }{
480 + {"string", funcapi.FieldTypeString},
481 + {"STRING", funcapi.FieldTypeString},
482 + {"integer", funcapi.FieldTypeInteger},
483 + {"INTEGER", funcapi.FieldTypeInteger},
484 + {"float", funcapi.FieldTypeFloat},
485 + {"FLOAT", funcapi.FieldTypeFloat},
486 + {"boolean", funcapi.FieldTypeBoolean},
487 + {"BOOLEAN", funcapi.FieldTypeBoolean},
488 + {"duration", funcapi.FieldTypeDuration},
489 + {"DURATION", funcapi.FieldTypeDuration},
490 + {"timestamp", funcapi.FieldTypeTimestamp},
491 + {"TIMESTAMP", funcapi.FieldTypeTimestamp},
492 + {"unknown", funcapi.FieldTypeString},
493 + {"", funcapi.FieldTypeString},
494 + }
495 +
496 + for _, tc := range tests {
497 + t.Run(tc.input, func(t *testing.T) {
498 + result := parseFieldType(tc.input)
499 + assert.Equal(t, tc.expected, result)
500 + })
501 + }
502 +}
503 +
504 +func TestDeriveTransform(t *testing.T) {
505 + tests := []struct {
506 + fieldType funcapi.FieldType
507 + expected funcapi.FieldTransform
508 + }{
509 + {funcapi.FieldTypeInteger, funcapi.FieldTransformNumber},
510 + {funcapi.FieldTypeFloat, funcapi.FieldTransformNumber},
511 + {funcapi.FieldTypeDuration, funcapi.FieldTransformDuration},
512 + {funcapi.FieldTypeTimestamp, funcapi.FieldTransformDatetime},
513 + {funcapi.FieldTypeString, funcapi.FieldTransformNone},
514 + {funcapi.FieldTypeBoolean, funcapi.FieldTransformNone},
515 + }
516 +
517 + for _, tc := range tests {
518 + t.Run(tc.fieldType.String(), func(t *testing.T) {
519 + result := deriveTransform(tc.fieldType)
520 + assert.Equal(t, tc.expected, result)
521 + })
522 + }
523 +}
524 +
525 +func TestDeriveFilterSummary(t *testing.T) {
526 + tests := []struct {
527 + fieldType funcapi.FieldType
528 + expectedFilter funcapi.FieldFilter
529 + expectedSummary funcapi.FieldSummary
530 + }{
531 + {funcapi.FieldTypeInteger, funcapi.FieldFilterRange, funcapi.FieldSummarySum},
532 + {funcapi.FieldTypeFloat, funcapi.FieldFilterRange, funcapi.FieldSummarySum},
533 + {funcapi.FieldTypeDuration, funcapi.FieldFilterRange, funcapi.FieldSummarySum},
534 + {funcapi.FieldTypeTimestamp, funcapi.FieldFilterRange, funcapi.FieldSummaryMax},
535 + {funcapi.FieldTypeBoolean, funcapi.FieldFilterMultiselect, funcapi.FieldSummaryCount},
536 + {funcapi.FieldTypeString, funcapi.FieldFilterMultiselect, funcapi.FieldSummaryCount},
537 + }
538 +
539 + for _, tc := range tests {
540 + t.Run(tc.fieldType.String(), func(t *testing.T) {
541 + filter, summary := deriveFilterSummary(tc.fieldType)
542 + assert.Equal(t, tc.expectedFilter, filter)
543 + assert.Equal(t, tc.expectedSummary, summary)
544 + })
545 + }
546 +}
547 +
548 +func TestDeriveNameFromID(t *testing.T) {
549 + tests := []struct {
550 + id string
551 + expected string
552 + }{
553 + {"slow-queries", "Slow Queries"},
554 + {"top_connections", "Top Connections"},
555 + {"active-sessions", "Active Sessions"},
556 + {"simple", "Simple"},
557 + {"multi-word-id", "Multi Word Id"},
558 + {"", ""},
559 + }
560 +
561 + for _, tc := range tests {
562 + t.Run(tc.id, func(t *testing.T) {
563 + result := deriveNameFromID(tc.id)
564 + assert.Equal(t, tc.expected, result)
565 + })
566 + }
567 +}
568 +
569 +func TestConfigFunction_derivedName(t *testing.T) {
570 + tests := []struct {
571 + name string
572 + cfg ConfigFunction
573 + expected string
574 + }{
575 + {
576 + name: "uses explicit name",
577 + cfg: ConfigFunction{ID: "slow-queries", Name: "My Custom Name"},
578 + expected: "My Custom Name",
579 + },
580 + {
581 + name: "derives from ID",
582 + cfg: ConfigFunction{ID: "slow-queries"},
583 + expected: "Slow Queries",
584 + },
585 + }
586 +
587 + for _, tc := range tests {
588 + t.Run(tc.name, func(t *testing.T) {
589 + result := tc.cfg.derivedName()
590 + assert.Equal(t, tc.expected, result)
591 + })
592 + }
593 +}
src/go/plugin/go.d/collector/sql/metadata.yaml
+273 -2
@@ -38,9 +38,12 @@ modules:
38 configuration**. There is no fixed metric reference: each job can expose
39 different metrics depending on its `metrics` and `queries` blocks.
40
41 - To see what a specific job collects, open that job’s dashboard in Netdata
41 + To see what a specific job collects, open that job's dashboard in Netdata
42 and inspect the charts and dimensions it created.
43
44 + Jobs can also define **functions** that provide interactive table views in
45 + Netdata's Top tab. A job can have metrics only, functions only, or both.
46 +
47 :::tip
48
49 To change what is collected, edit the `metrics` (and optional `queries`)
@@ -61,6 +64,10 @@ modules:
64 `query_ref`), reads the result set, and maps it to Netdata charts and
65 dimensions.
66
67 + Additionally, you can define **functions** that expose SQL query results as
68 + interactive table views in Netdata's Top tab. Functions support filtering,
69 + sorting, and searching without creating persistent metrics.
70 +
71 ### Result Processing Modes
72
73 | Mode | How it works | Best used when |
@@ -204,6 +211,29 @@ modules:
211 equals: <string|number|bool> # Active (1) if value == this literal.
212 # in: [ <v1>, <v2>, ... ] # Active if value is in the list.
213 # match: '^regex$' # Active if value matches this regex.
214 +
215 + # ---------- FUNCTIONS ----------
216 + # Set function_only: true if this job only provides functions (no metrics).
217 + function_only: <true|false> # OPTIONAL. Default: false.
218 +
219 + # Expose SQL queries as interactive table views in Netdata's Top tab.
220 + functions:
221 + - id: <function_id> # REQUIRED. Unique identifier.
222 + name: <display_name> # OPTIONAL. Derived from id if not set.
223 + description: <help_text> # OPTIONAL. Shown in the UI.
224 + query: | # REQUIRED. SQL to execute.
225 + SELECT ...
226 + timeout: <seconds> # OPTIONAL. Query timeout.
227 + limit: <max_rows> # OPTIONAL. Default: 100.
228 + default_sort: <column_name> # OPTIONAL. Initial sort column.
229 + default_sort_desc: <true|false> # OPTIONAL. Default: true.
230 + columns: # OPTIONAL. Override column metadata.
231 + <column_name>:
232 + type: <string|integer|float|boolean|duration|timestamp>
233 + units: <unit_string>
234 + tooltip: <hover_text>
235 + visible: <true|false>
236 + sortable: <true|false>
237 ```
238 folding:
239 title: Config options
@@ -260,9 +290,72 @@ modules:
290 description: >
291 A list of metric blocks. Each block defines how a query is executed and how its result is transformed into one or more charts. See [Configuration Structure](#configuration) for details.
292 default_value: "[]"
263 - required: true
293 + required: false
294 group: Queries & Metrics
295
296 + - name: functions
297 + description: >
298 + A list of SQL functions exposed as interactive table views in Netdata's Top tab.
299 + Each function runs a SQL query and displays results in a filterable, sortable table.
300 + See [Functions](#functions) for details.
301 + default_value: "[]"
302 + required: false
303 + group: Functions
304 + - name: functions[].id
305 + description: Unique identifier for this function.
306 + default_value: ""
307 + required: true
308 + group: Functions
309 + - name: functions[].name
310 + description: Display name shown in the UI. Auto-derived from ID if not set.
311 + default_value: ""
312 + required: false
313 + group: Functions
314 + - name: functions[].description
315 + description: Help text shown in the UI.
316 + default_value: ""
317 + required: false
318 + group: Functions
319 + - name: functions[].query
320 + description: SQL query to execute when this function is called.
321 + default_value: ""
322 + required: true
323 + group: Functions
324 + - name: functions[].timeout
325 + description: Query timeout (seconds). Uses collector timeout if not set.
326 + default_value: ""
327 + required: false
328 + group: Functions
329 + - name: functions[].limit
330 + description: Maximum rows to return.
331 + default_value: 100
332 + required: false
333 + group: Functions
334 + - name: functions[].default_sort
335 + description: Column name for initial sort order.
336 + default_value: ""
337 + required: false
338 + group: Functions
339 + - name: functions[].default_sort_desc
340 + description: Sort in descending order by default.
341 + default_value: true
342 + required: false
343 + group: Functions
344 + - name: functions[].columns
345 + description: >
346 + Override auto-detected column metadata. Map of column name to settings
347 + (type, units, tooltip, visible, sortable).
348 + default_value: "{}"
349 + required: false
350 + group: Functions
351 + - name: function_only
352 + description: >
353 + Set to true if this job only provides functions (no metrics).
354 + When enabled, metrics configuration is not required and no charts are created.
355 + default_value: false
356 + required: false
357 + group: Functions
358 +
359 - name: vnode
360 description: Associates this data collection job with a Virtual Node.
361 default_value: ""
@@ -508,10 +601,188 @@ modules:
601 source: pg_is_in_recovery
602 status_when:
603 equals: "f"
604 +
605 + - name: Function-only mode – slow query analysis
606 + description: |
607 + PostgreSQL example that provides an interactive slow query analysis view
608 + without collecting any time-series metrics.
609 +
610 + This is useful for ad-hoc troubleshooting via the Netdata **Top** tab.
611 + The function queries `pg_stat_statements` to show the slowest queries
612 + sorted by total execution time.
613 + config: |
614 + jobs:
615 + - name: pg_slow_queries
616 + driver: pgx
617 + dsn: 'postgresql://netdata:password@127.0.0.1:5432/postgres'
618 + timeout: 10
619 + function_only: true
620 +
621 + functions:
622 + - id: slow-queries
623 + name: Slow Queries
624 + description: Top queries by total execution time from pg_stat_statements
625 + query: |
626 + SELECT
627 + queryid,
628 + LEFT(query, 100) AS query,
629 + calls,
630 + total_exec_time,
631 + mean_exec_time,
632 + rows
633 + FROM pg_stat_statements
634 + ORDER BY total_exec_time DESC
635 + limit: 100
636 + default_sort: total_exec_time
637 + default_sort_desc: true
638 + columns:
639 + total_exec_time:
640 + type: duration
641 + units: milliseconds
642 + tooltip: Total time spent executing this query
643 + mean_exec_time:
644 + type: duration
645 + units: milliseconds
646 + tooltip: Average execution time per call
647 +
648 + - name: Combined metrics and functions
649 + description: |
650 + PostgreSQL example that collects time-series metrics AND provides
651 + interactive function views in the same job.
652 +
653 + - The `metrics` block creates charts for connection states.
654 + - The `functions` block provides an interactive activity view.
655 + config: |
656 + jobs:
657 + - name: pg_combined
658 + driver: pgx
659 + dsn: 'postgresql://netdata:password@127.0.0.1:5432/postgres'
660 + timeout: 5
661 +
662 + # Time-series metrics
663 + metrics:
664 + - id: connections
665 + mode: kv
666 + query: |
667 + SELECT state, count(*) AS cnt
668 + FROM pg_stat_activity
669 + GROUP BY state
670 + kv_mode:
671 + name_col: state
672 + value_col: cnt
673 + charts:
674 + - title: "Connection states"
675 + context: sql.pg_connections
676 + family: connections
677 + units: connections
678 + type: stacked
679 + dims:
680 + - name: active
681 + source: active
682 + - name: idle
683 + source: idle
684 +
685 + # Interactive functions
686 + functions:
687 + - id: active-sessions
688 + name: Active Sessions
689 + description: Currently running queries
690 + query: |
691 + SELECT
692 + pid,
693 + usename,
694 + datname,
695 + state,
696 + query_start,
697 + LEFT(query, 200) AS query
698 + FROM pg_stat_activity
699 + WHERE state = 'active'
700 + limit: 50
701 + columns:
702 + query_start:
703 + type: timestamp
704 troubleshooting:
705 problems:
706 list: []
707 alerts: []
708 + functions:
709 + description: |
710 + This collector supports user-defined SQL functions that expose query results as
711 + interactive table views in Netdata's **Top** tab.
712 +
713 + Unlike metrics which create time-series charts, functions provide on-demand,
714 + tabular data views that users can filter, sort, and search.
715 +
716 + **How Functions Appear in the UI**:
717 +
718 + Functions are organized hierarchically in the **Top** tab:
719 +
720 + ```
721 + Databases
722 + └── SQL
723 + └── <job_name>
724 + ├── <function_name_1>
725 + └── <function_name_2>
726 + ```
727 +
728 + Each job creates its own group, and each function within that job appears as a
729 + selectable item. The function's `name` field (or auto-derived name from `id`)
730 + is displayed in the UI.
731 +
732 + > **Note:** Function IDs cannot contain colons (`:`) as they are used internally
733 + > as delimiters. Use hyphens (`-`) or underscores (`_`) instead.
734 +
735 + **Use Cases**:
736 +
737 + - **Slow query analysis**: Query `pg_stat_statements` or `performance_schema` to show queries by execution time.
738 + - **Active connections**: List current sessions from `pg_stat_activity` or `information_schema.processlist`.
739 + - **Lock monitoring**: Display blocking locks from `pg_locks` or `sys.dm_tran_locks`.
740 + - **Table statistics**: Show per-table row counts, bloat, or index usage.
741 +
742 + **Configuration**:
743 +
744 + Functions are defined in the `functions` list in your job configuration:
745 +
746 + ```yaml
747 + functions:
748 + - id: slow-queries # Required: unique identifier
749 + name: Slow Queries # Optional: display name (derived from id if not set)
750 + description: Top queries # Optional: help text
751 + query: | # Required: SQL to execute
752 + SELECT query, calls, total_time
753 + FROM pg_stat_statements
754 + ORDER BY total_time DESC
755 + timeout: 10 # Optional: query timeout (seconds)
756 + limit: 100 # Optional: max rows (default: 100)
757 + default_sort: total_time # Optional: initial sort column
758 + default_sort_desc: true # Optional: descending sort (default: true)
759 + columns: # Optional: column overrides
760 + total_time:
761 + type: duration
762 + units: milliseconds
763 + ```
764 +
765 + **Column Types**:
766 +
767 + Column types are auto-detected from database metadata but can be overridden:
768 +
769 + | Type | Description | Example columns |
770 + |-------------|------------------------------------------|----------------------------|
771 + | `string` | Text values | query, schema, username |
772 + | `integer` | Whole numbers | calls, rows, connections |
773 + | `float` | Decimal numbers | ratio, percentage |
774 + | `boolean` | True/false values | is_active, enabled |
775 + | `duration` | Time intervals (specify units) | execution_time, wait_time |
776 + | `timestamp` | Date/time values | start_time, last_seen |
777 +
778 + **Function-Only Mode**:
779 +
780 + Jobs can be configured to only provide functions by setting `function_only: true`.
781 + In this mode:
782 + - No charts are created (metrics configuration is not required)
783 + - The job only provides interactive table views via the Top tab
784 + - Useful for ad-hoc analysis without ongoing metric collection
785 + list: []
786 metrics:
787 folding:
788 title: Metrics