@cryptotaxi247 / netdata-1 / commits / ff5815062

refactor(go.d): agent decoupling (#21821)

Ilya Mashchenko committed Feb 26, 2026 at 10:29 UTC ff58150626400701d5c603a6d55323d203850de5
41 files changed +863 -278
src/go/cmd/godplugin/main.go
+47 -8
@@ -14,10 +14,14 @@ import (
14 "strings"
15 "time"
16
17 + "github.com/netdata/netdata/go/plugins/cmd/internal/agenthost"
18 + "github.com/netdata/netdata/go/plugins/cmd/internal/discoveryproviders"
19 "github.com/netdata/netdata/go/plugins/plugin/agent"
20 + "github.com/netdata/netdata/go/plugins/plugin/agent/discovery"
21 "github.com/netdata/netdata/go/plugins/plugin/agent/discovery/dummy"
22 "github.com/netdata/netdata/go/plugins/plugin/agent/discovery/file"
23 "github.com/netdata/netdata/go/plugins/plugin/agent/jobmgr"
24 + "github.com/netdata/netdata/go/plugins/plugin/agent/policy"
25 "go.uber.org/automaxprocs/maxprocs"
26 "golang.org/x/net/http/httpproxy"
27
@@ -25,14 +29,17 @@ import (
29 "github.com/netdata/netdata/go/plugins/pkg/buildinfo"
30 "github.com/netdata/netdata/go/plugins/pkg/cli"
31 "github.com/netdata/netdata/go/plugins/pkg/executable"
32 + "github.com/netdata/netdata/go/plugins/pkg/hostinfo"
33 "github.com/netdata/netdata/go/plugins/pkg/multipath"
34 "github.com/netdata/netdata/go/plugins/pkg/netdataapi"
35 "github.com/netdata/netdata/go/plugins/pkg/pluginconfig"
36 + "github.com/netdata/netdata/go/plugins/pkg/terminal"
37 "github.com/netdata/netdata/go/plugins/plugin/framework/collectorapi"
38 "github.com/netdata/netdata/go/plugins/plugin/framework/confgroup"
39 "github.com/netdata/netdata/go/plugins/plugin/framework/dyncfg"
40 "github.com/netdata/netdata/go/plugins/plugin/framework/functions"
41 _ "github.com/netdata/netdata/go/plugins/plugin/go.d/collector"
42 + "github.com/netdata/netdata/go/plugins/plugin/go.d/discovery/sdext"
43 )
44
45 func init() {
@@ -64,6 +71,9 @@ func main() {
71 if opts.Debug {
72 logger.Level.Set(slog.LevelDebug)
73 }
74 + isTerminal := terminal.IsTerminal()
75 + isInsideK8s := hostinfo.IsInsideK8sCluster()
76 + moduleRegistry := moduleRegistryWithSystemdPolicy(collectorapi.DefaultRegistry, hostinfo.SystemdVersion)
77
78 a := agent.New(agent.Config{
79 Name: executable.Name,
@@ -72,10 +82,22 @@ func main() {
82 ServiceDiscoveryConfigDir: pluginconfig.ServiceDiscoveryDir(),
83 CollectorsConfigWatchPath: pluginconfig.CollectorsConfigWatchPaths(),
84 VarLibDir: pluginconfig.VarLibDir(),
75 - RunModule: opts.Module,
76 - RunJob: opts.Job,
77 - MinUpdateEvery: opts.UpdateEvery,
78 - DumpSummary: opts.DumpSummary,
85 + ModuleRegistry: moduleRegistry,
86 + IsInsideK8s: isInsideK8s,
87 + RunModePolicy: policy.RunModePolicy{
88 + IsTerminal: isTerminal,
89 + AutoEnableDiscovered: isTerminal,
90 + UseFileStatusPersistence: !isTerminal,
91 + },
92 + DiscoveryProviders: []discovery.ProviderFactory{
93 + discoveryproviders.File(),
94 + discoveryproviders.Dummy(),
95 + discoveryproviders.SD(sdext.Registry(!isInsideK8s)),
96 + },
97 + RunModule: opts.Module,
98 + RunJob: opts.Job,
99 + MinUpdateEvery: opts.UpdateEvery,
100 + DumpSummary: opts.DumpSummary,
101 })
102
103 a.Infof("plugin: name=%s, %s", a.Name, buildinfo.Info())
@@ -89,7 +111,7 @@ func main() {
111 a.Infof("directories → config: %s | collectors: %s | sd: %s | varlib: %s",
112 a.ConfigDir, a.CollectorsConfDir, a.ServiceDiscoveryConfigDir, a.VarLibDir)
113
92 - a.Run()
114 + agenthost.Run(a)
115 }
116
117 func parseCLI() *cli.Option {
@@ -104,6 +126,19 @@ func parseCLI() *cli.Option {
126 return opt
127 }
128
129 +func moduleRegistryWithSystemdPolicy(base collectorapi.Registry, systemdVersion int) collectorapi.Registry {
130 + registry := make(collectorapi.Registry, len(base))
131 + for name, creator := range base {
132 + if name == "logind" && systemdVersion == 239 {
133 + // Known issue: go.d/logind high CPU usage on Alma Linux8.
134 + // Keep policy in cmd wiring, not inside generic agent package.
135 + creator.Disabled = true
136 + }
137 + registry[name] = creator
138 + }
139 + return registry
140 +}
141 +
142 func runFunctionCLI(opts *cli.Option) int {
143 functionName := strings.TrimSpace(opts.Function)
144 if functionName == "" {
@@ -165,9 +200,13 @@ func runFunctionCLI(opts *cli.Option) int {
200 defer cancel()
201
202 jobMgr := jobmgr.New(jobmgr.Config{
168 - // Force-enable configs in function CLI runs (non-TTY by default).
169 - PluginName: "nodyncfg",
170 - Out: io.Discard,
203 + PluginName: executable.Name,
204 + Out: io.Discard,
205 + RunModePolicy: policy.RunModePolicy{
206 + IsTerminal: false,
207 + AutoEnableDiscovered: true,
208 + UseFileStatusPersistence: true,
209 + },
210 VarLibDir: pluginconfig.VarLibDir(),
211 Modules: collectorapi.Registry{moduleName: creator},
212 ConfigDefaults: reg,
src/go/cmd/godplugin/main_test.go new
+26
@@ -0,0 +1,26 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package main
4 +
5 +import (
6 + "testing"
7 +
8 + "github.com/netdata/netdata/go/plugins/plugin/framework/collectorapi"
9 + "github.com/stretchr/testify/assert"
10 +)
11 +
12 +func TestModuleRegistryWithSystemdPolicy(t *testing.T) {
13 + base := collectorapi.Registry{
14 + "logind": collectorapi.Creator{},
15 + "other": collectorapi.Creator{},
16 + }
17 +
18 + withPolicy := moduleRegistryWithSystemdPolicy(base, 239)
19 + assert.True(t, withPolicy["logind"].Disabled)
20 + assert.False(t, withPolicy["other"].Disabled)
21 + assert.False(t, base["logind"].Disabled, "base registry must remain unchanged")
22 +
23 + withoutPolicy := moduleRegistryWithSystemdPolicy(base, 250)
24 + assert.False(t, withoutPolicy["logind"].Disabled)
25 + assert.False(t, withoutPolicy["other"].Disabled)
26 +}
src/go/cmd/ibmdplugin/main.go
+24 -5
@@ -14,7 +14,11 @@ import (
14 "strings"
15 "time"
16
17 + "github.com/netdata/netdata/go/plugins/cmd/internal/agenthost"
18 + "github.com/netdata/netdata/go/plugins/cmd/internal/discoveryproviders"
19 "github.com/netdata/netdata/go/plugins/plugin/agent"
20 + "github.com/netdata/netdata/go/plugins/plugin/agent/discovery"
21 + "github.com/netdata/netdata/go/plugins/plugin/agent/policy"
22 "go.uber.org/automaxprocs/maxprocs"
23 "golang.org/x/net/http/httpproxy"
24
@@ -22,7 +26,10 @@ import (
26 "github.com/netdata/netdata/go/plugins/pkg/buildinfo"
27 "github.com/netdata/netdata/go/plugins/pkg/cli"
28 "github.com/netdata/netdata/go/plugins/pkg/executable"
29 + "github.com/netdata/netdata/go/plugins/pkg/hostinfo"
30 "github.com/netdata/netdata/go/plugins/pkg/pluginconfig"
31 + "github.com/netdata/netdata/go/plugins/pkg/terminal"
32 + "github.com/netdata/netdata/go/plugins/plugin/framework/collectorapi"
33 // Register IBM ecosystem collectors
34 _ "github.com/netdata/netdata/go/plugins/plugin/ibm.d/modules/as400" // Requires CGO
35 _ "github.com/netdata/netdata/go/plugins/plugin/ibm.d/modules/db2" // Requires CGO
@@ -73,6 +80,7 @@ func main() {
80 if opts.Debug {
81 logger.Level.Set(slog.LevelDebug)
82 }
83 + isTerminal := terminal.IsTerminal()
84
85 // Parse dump duration if provided
86 var dumpMode time.Duration
@@ -86,10 +94,21 @@ func main() {
94 }
95
96 a := agent.New(agent.Config{
89 - Name: executable.Name,
90 - PluginConfigDir: pluginconfig.ConfigDir(),
91 - CollectorsConfigDir: pluginconfig.CollectorsDir(),
92 - VarLibDir: pluginconfig.VarLibDir(),
97 + Name: executable.Name,
98 + PluginConfigDir: pluginconfig.ConfigDir(),
99 + CollectorsConfigDir: pluginconfig.CollectorsDir(),
100 + VarLibDir: pluginconfig.VarLibDir(),
101 + ModuleRegistry: collectorapi.DefaultRegistry,
102 + IsInsideK8s: hostinfo.IsInsideK8sCluster(),
103 + RunModePolicy: policy.RunModePolicy{
104 + IsTerminal: isTerminal,
105 + AutoEnableDiscovered: isTerminal,
106 + UseFileStatusPersistence: !isTerminal,
107 + },
108 + DiscoveryProviders: []discovery.ProviderFactory{
109 + discoveryproviders.File(),
110 + discoveryproviders.Dummy(),
111 + },
112 RunModule: opts.Module,
113 RunJob: opts.Job,
114 MinUpdateEvery: opts.UpdateEvery,
@@ -110,7 +129,7 @@ func main() {
129 a.Infof("directories → config: %s | collectors: %s | sd: %s | varlib: %s",
130 a.ConfigDir, a.CollectorsConfDir, a.ServiceDiscoveryConfigDir, a.VarLibDir)
131
113 - a.Run()
132 + agenthost.Run(a)
133 }
134
135 func parseCLI() *cli.Option {
src/go/cmd/internal/agenthost/host.go new
+112
@@ -0,0 +1,112 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package agenthost
4 +
5 +import (
6 + "context"
7 + "os"
8 + "os/signal"
9 + "sync"
10 + "syscall"
11 + "time"
12 +
13 + "github.com/netdata/netdata/go/plugins/plugin/agent"
14 + "github.com/netdata/netdata/go/plugins/plugin/framework/collectorapi"
15 +)
16 +
17 +// Run hosts an agent process lifecycle (signals, restart, quit, dump timer).
18 +func Run(a *agent.Agent) {
19 + ch := make(chan os.Signal, 1)
20 + signal.Notify(ch, syscall.SIGHUP, syscall.SIGINT, syscall.SIGTERM)
21 + signal.Ignore(syscall.SIGPIPE)
22 +
23 + var keepAliveErr <-chan error
24 + if !a.IsTerminalMode() {
25 + ch := make(chan error, 1)
26 + keepAliveErr = ch
27 + go func() {
28 + if err := a.RunKeepAlive(context.Background()); err != nil {
29 + select {
30 + case ch <- err:
31 + default:
32 + }
33 + }
34 + }()
35 + }
36 +
37 + var wg sync.WaitGroup
38 + var exit bool
39 +
40 + var dumpTimer *time.Timer
41 + var dumpTimerCh <-chan time.Time
42 + if mode := a.DumpModeDuration(); mode > 0 {
43 + dumpTimer = time.NewTimer(mode)
44 + dumpTimerCh = dumpTimer.C
45 + defer dumpTimer.Stop()
46 + }
47 +
48 + for {
49 + collectorapi.ObsoleteCharts(true)
50 +
51 + ctx, cancel := context.WithCancel(context.Background())
52 + runDone := make(chan struct{})
53 + wg.Add(1)
54 + go func() {
55 + defer wg.Done()
56 + defer close(runDone)
57 + a.RunContext(ctx)
58 + }()
59 +
60 + select {
61 + case sig := <-ch:
62 + switch sig {
63 + case syscall.SIGHUP:
64 + a.Infof("received %s signal (%d). Restarting running instance", sig, sig)
65 + default:
66 + a.Infof("received %s signal (%d). Terminating...", sig, sig)
67 + exit = true
68 + }
69 + case <-a.QuitCh():
70 + a.Infof("received QUIT command. Terminating...")
71 + exit = true
72 + case <-dumpTimerCh:
73 + a.Infof("dump mode duration expired, collecting analysis...")
74 + a.TriggerDumpAnalysis()
75 + exit = true
76 + case <-keepAliveErr:
77 + a.Info("too many keepAlive errors. Terminating...")
78 + exit = true
79 + case <-runDone:
80 + a.Info("agent run loop stopped. Terminating...")
81 + exit = true
82 + }
83 +
84 + if exit {
85 + collectorapi.ObsoleteCharts(false)
86 + }
87 +
88 + cancel()
89 +
90 + func() {
91 + timeout := time.Second * 10
92 + t := time.NewTimer(timeout)
93 + defer t.Stop()
94 + done := make(chan struct{})
95 +
96 + go func() { wg.Wait(); close(done) }()
97 +
98 + select {
99 + case <-t.C:
100 + a.Errorf("stopping all goroutines timed out after %s. Exiting...", timeout)
101 + os.Exit(0)
102 + case <-done:
103 + }
104 + }()
105 +
106 + if exit {
107 + os.Exit(0)
108 + }
109 +
110 + time.Sleep(time.Second)
111 + }
112 +}
src/go/cmd/internal/discoveryproviders/providers.go new
+67
@@ -0,0 +1,67 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package discoveryproviders
4 +
5 +import (
6 + "github.com/netdata/netdata/go/plugins/plugin/agent/discovery"
7 + "github.com/netdata/netdata/go/plugins/plugin/agent/discovery/dummy"
8 + "github.com/netdata/netdata/go/plugins/plugin/agent/discovery/file"
9 + "github.com/netdata/netdata/go/plugins/plugin/agent/discovery/sd"
10 +)
11 +
12 +func File() discovery.ProviderFactory {
13 + return discovery.NewProviderFactory("file", func(ctx discovery.BuildContext) (discovery.Discoverer, bool, error) {
14 + if len(ctx.ReadPaths)+len(ctx.Paths.CollectorsConfigWatchPath) == 0 {
15 + return nil, false, nil
16 + }
17 +
18 + d, err := file.NewDiscovery(file.Config{
19 + Registry: ctx.Registry,
20 + Read: ctx.ReadPaths,
21 + Watch: ctx.Paths.CollectorsConfigWatchPath,
22 + })
23 + if err != nil {
24 + return nil, false, err
25 + }
26 + return d, true, nil
27 + })
28 +}
29 +
30 +func Dummy() discovery.ProviderFactory {
31 + return discovery.NewProviderFactory("dummy", func(ctx discovery.BuildContext) (discovery.Discoverer, bool, error) {
32 + if len(ctx.DummyNames) == 0 {
33 + return nil, false, nil
34 + }
35 +
36 + d, err := dummy.NewDiscovery(dummy.Config{
37 + Registry: ctx.Registry,
38 + Names: ctx.DummyNames,
39 + })
40 + if err != nil {
41 + return nil, false, err
42 + }
43 + return d, true, nil
44 + })
45 +}
46 +
47 +func SD(registry sd.Registry) discovery.ProviderFactory {
48 + return discovery.NewProviderFactory("sd", func(ctx discovery.BuildContext) (discovery.Discoverer, bool, error) {
49 + if len(ctx.Paths.ServiceDiscoveryConfigDir) == 0 {
50 + return nil, false, nil
51 + }
52 +
53 + d, err := sd.NewServiceDiscovery(sd.Config{
54 + ConfigDefaults: ctx.Registry,
55 + PluginName: ctx.Identity.Name,
56 + RunModePolicy: ctx.RunMode,
57 + Out: ctx.Out,
58 + ConfDir: ctx.Paths.ServiceDiscoveryConfigDir,
59 + FnReg: ctx.FnReg,
60 + Discoverers: registry,
61 + })
62 + if err != nil {
63 + return nil, false, err
64 + }
65 + return d, true, nil
66 + })
67 +}
src/go/cmd/scriptsdplugin/main.go
+25 -6
@@ -10,7 +10,11 @@ import (
10 "path/filepath"
11 "strings"
12
13 + "github.com/netdata/netdata/go/plugins/cmd/internal/agenthost"
14 + "github.com/netdata/netdata/go/plugins/cmd/internal/discoveryproviders"
15 "github.com/netdata/netdata/go/plugins/plugin/agent"
16 + "github.com/netdata/netdata/go/plugins/plugin/agent/discovery"
17 + "github.com/netdata/netdata/go/plugins/plugin/agent/policy"
18 "go.uber.org/automaxprocs/maxprocs"
19 "golang.org/x/net/http/httpproxy"
20
@@ -18,7 +22,10 @@ import (
22 "github.com/netdata/netdata/go/plugins/pkg/buildinfo"
23 "github.com/netdata/netdata/go/plugins/pkg/cli"
24 "github.com/netdata/netdata/go/plugins/pkg/executable"
25 + "github.com/netdata/netdata/go/plugins/pkg/hostinfo"
26 "github.com/netdata/netdata/go/plugins/pkg/pluginconfig"
27 + "github.com/netdata/netdata/go/plugins/pkg/terminal"
28 + "github.com/netdata/netdata/go/plugins/plugin/framework/collectorapi"
29 _ "github.com/netdata/netdata/go/plugins/plugin/scripts.d/modules/nagios"
30 _ "github.com/netdata/netdata/go/plugins/plugin/scripts.d/modules/scheduler"
31 )
@@ -55,6 +62,7 @@ func main() {
62 if opts.Debug {
63 logger.Level.Set(slog.LevelDebug)
64 }
65 + isTerminal := terminal.IsTerminal()
66
67 a := agent.New(agent.Config{
68 Name: executable.Name,
@@ -63,11 +71,22 @@ func main() {
71 ServiceDiscoveryConfigDir: nil,
72 CollectorsConfigWatchPath: watchPaths,
73 VarLibDir: pluginconfig.VarLibDir(),
66 - RunModule: opts.Module,
67 - RunJob: opts.Job,
68 - MinUpdateEvery: opts.UpdateEvery,
69 - DumpSummary: opts.DumpSummary,
70 - DisableServiceDiscovery: true,
74 + ModuleRegistry: collectorapi.DefaultRegistry,
75 + IsInsideK8s: hostinfo.IsInsideK8sCluster(),
76 + RunModePolicy: policy.RunModePolicy{
77 + IsTerminal: isTerminal,
78 + AutoEnableDiscovered: isTerminal,
79 + UseFileStatusPersistence: !isTerminal,
80 + },
81 + DiscoveryProviders: []discovery.ProviderFactory{
82 + discoveryproviders.File(),
83 + discoveryproviders.Dummy(),
84 + },
85 + RunModule: opts.Module,
86 + RunJob: opts.Job,
87 + MinUpdateEvery: opts.UpdateEvery,
88 + DumpSummary: opts.DumpSummary,
89 + DisableServiceDiscovery: true,
90 })
91
92 a.Debugf("plugin: name=%s, %s", a.Name, buildinfo.Info())
@@ -81,7 +100,7 @@ func main() {
100 a.Infof("directories → config: %s | collectors: %s | varlib: %s",
101 a.ConfigDir, a.CollectorsConfDir, a.VarLibDir)
102
84 - a.Run()
103 + agenthost.Run(a)
104 }
105
106 func parseCLI() *cli.Option {
src/go/pkg/pluginconfig/pluginconfig.go
+6 -1
@@ -56,7 +56,12 @@ type directories struct {
56 }
57
58 func IsStock(path string) bool {
59 - return strings.HasPrefix(path, StockConfigDir())
59 + stock := StockConfigDir()
60 + if stock == "" {
61 + // Fallback for contexts that haven't called MustInit yet (mostly unit tests).
62 + return !strings.Contains(path, "/etc/")
63 + }
64 + return strings.HasPrefix(path, stock)
65 }
66
67 // MustInit parses env, applies CLI overrides, discovers directories, and stores them.
src/go/pkg/pluginconfig/pluginconfig_test.go
+18
@@ -344,3 +344,21 @@ func TestDirectoriesBuildValidation(t *testing.T) {
344 })
345 }
346 }
347 +
348 +func TestIsStock(t *testing.T) {
349 + orig := dirs
350 + t.Cleanup(func() { dirs = orig })
351 +
352 + t.Run("fallback heuristic when stock dir is not initialized", func(t *testing.T) {
353 + dirs = directories{}
354 + assert.True(t, IsStock("/usr/lib/netdata/conf.d/go.d/module.conf"))
355 + assert.False(t, IsStock("/etc/netdata/go.d/module.conf"))
356 + })
357 +
358 + t.Run("uses configured stock root when available", func(t *testing.T) {
359 + dirs = directories{stockConfigDir: "/custom/stock"}
360 + assert.True(t, IsStock("/custom/stock/go.d/module.conf"))
361 + assert.False(t, IsStock("/usr/lib/netdata/conf.d/go.d/module.conf"))
362 + assert.False(t, IsStock("/etc/netdata/go.d/module.conf"))
363 + })
364 +}
src/go/plugin/agent/README.md
+11
@@ -25,6 +25,17 @@ Package provides:
25
26 You are responsible only for __creating modules__.
27
28 +## Architecture Boundaries
29 +
30 +The `agent` package is framework/core orchestration only.
31 +
32 +- `src/go/plugin/agent/**`: context-driven orchestration and adapter contracts.
33 +- `src/go/cmd/*`: composition root and process host responsibilities:
34 + - signal handling / exits / keepalive lifecycle
35 + - module registry selection
36 + - discovery provider wiring and policy selection
37 +- Provider implementations (for example go.d SD discoverers) are selected explicitly in `cmd` wiring, not by implicit imports in `agent`.
38 +
39 ## Custom plugin example
40
41 [Yep! So easy!](https://github.com/netdata/netdata/blob/master/src/go/plugin/go.d/examples/simple/main.go)
src/go/plugin/agent/agent.go
+54 -102
@@ -4,21 +4,19 @@ package agent
4
5 import (
6 "context"
7 + "fmt"
8 "io"
9 "log/slog"
9 - "os"
10 - "os/signal"
10 "sync"
12 - "syscall"
11 "time"
12
13 "github.com/netdata/netdata/go/plugins/logger"
14 "github.com/netdata/netdata/go/plugins/pkg/multipath"
15 "github.com/netdata/netdata/go/plugins/pkg/netdataapi"
16 "github.com/netdata/netdata/go/plugins/pkg/safewriter"
19 - "github.com/netdata/netdata/go/plugins/pkg/terminal"
17 "github.com/netdata/netdata/go/plugins/plugin/agent/discovery"
18 "github.com/netdata/netdata/go/plugins/plugin/agent/jobmgr"
19 + "github.com/netdata/netdata/go/plugins/plugin/agent/policy"
20 "github.com/netdata/netdata/go/plugins/plugin/agent/runtimemgr"
21 "github.com/netdata/netdata/go/plugins/plugin/framework/collectorapi"
22 "github.com/netdata/netdata/go/plugins/plugin/framework/confgroup"
@@ -42,6 +40,12 @@ type Config struct {
40
41 DisableServiceDiscovery bool
42
43 + IsInsideK8s bool
44 +
45 + RunModePolicy policy.RunModePolicy
46 +
47 + DiscoveryProviders []discovery.ProviderFactory
48 +
49 DumpMode time.Duration
50 DumpSummary bool
51 DumpDataDir string
@@ -66,6 +70,12 @@ type Agent struct {
70
71 DisableServiceDiscovery bool
72
73 + IsInsideK8s bool
74 +
75 + runModePolicy policy.RunModePolicy
76 +
77 + DiscoveryProviders []discovery.ProviderFactory
78 +
79 ModuleRegistry collectorapi.Registry
80 Out io.Writer
81
@@ -98,7 +108,10 @@ func New(cfg Config) *Agent {
108 RunModule: cfg.RunModule,
109 RunJob: cfg.RunJob,
110 MinUpdateEvery: cfg.MinUpdateEvery,
101 - ModuleRegistry: collectorapi.DefaultRegistry,
111 + IsInsideK8s: cfg.IsInsideK8s,
112 + runModePolicy: cfg.RunModePolicy,
113 + ModuleRegistry: cfg.ModuleRegistry,
114 + DiscoveryProviders: cfg.DiscoveryProviders,
115 Out: safewriter.Stdout,
116 api: netdataapi.New(safewriter.Stdout),
117 quitCh: make(chan struct{}, 1),
@@ -127,83 +140,52 @@ func New(cfg Config) *Agent {
140 return a
141 }
142
130 -// Run starts the Agent.
131 -func (a *Agent) Run() {
132 - go a.keepAlive()
133 - serve(a)
143 +// RunContext runs one agent instance lifecycle on the provided context.
144 +func (a *Agent) RunContext(ctx context.Context) {
145 + a.run(ctx)
146 }
147
136 -func serve(a *Agent) {
137 - ch := make(chan os.Signal, 1)
138 - signal.Notify(ch, syscall.SIGHUP, syscall.SIGINT, syscall.SIGTERM)
139 - signal.Ignore(syscall.SIGPIPE)
140 -
141 - var wg sync.WaitGroup
142 -
143 - var exit bool
148 +// IsTerminalMode reports whether run-mode policy is interactive terminal.
149 +func (a *Agent) IsTerminalMode() bool {
150 + return a.runModePolicy.IsTerminal
151 +}
152
145 - // Set up dump mode timer if enabled
146 - var dumpTimer *time.Timer
147 - var dumpTimerCh <-chan time.Time
148 - if a.dumpMode > 0 {
149 - dumpTimer = time.NewTimer(a.dumpMode)
150 - dumpTimerCh = dumpTimer.C
151 - }
153 +// RunKeepAlive runs keepalive loop until context cancellation or too many failures.
154 +func (a *Agent) RunKeepAlive(ctx context.Context) error {
155 + tk := time.NewTicker(time.Second)
156 + defer tk.Stop()
157
158 + var n int
159 for {
154 - collectorapi.ObsoleteCharts(true)
155 -
156 - ctx, cancel := context.WithCancel(context.Background())
157 -
158 - wg.Add(1)
159 - go func() { defer wg.Done(); a.run(ctx) }()
160 -
160 select {
162 - case sig := <-ch:
163 - switch sig {
164 - case syscall.SIGHUP:
165 - a.Infof("received %s signal (%d). Restarting running instance", sig, sig)
166 - default:
167 - a.Infof("received %s signal (%d). Terminating...", sig, sig)
168 - exit = true
161 + case <-ctx.Done():
162 + return nil
163 + case <-tk.C:
164 + if err := a.api.EMPTYLINE(); err != nil {
165 + n++
166 + } else {
167 + n = 0
168 + }
169 + if n >= 30 {
170 + return fmt.Errorf("too many keepAlive errors")
171 }
170 - case <-a.quitCh:
171 - a.Infof("received QUIT command. Terminating...")
172 - exit = true
173 - case <-dumpTimerCh:
174 - a.Infof("dump mode duration expired, collecting analysis...")
175 - a.collectDumpAnalysis()
176 - exit = true
177 - }
178 -
179 - if exit {
180 - collectorapi.ObsoleteCharts(false)
172 }
173 + }
174 +}
175
183 - cancel()
184 -
185 - func() {
186 - timeout := time.Second * 10
187 - t := time.NewTimer(timeout)
188 - defer t.Stop()
189 - done := make(chan struct{})
190 -
191 - go func() { wg.Wait(); close(done) }()
192 -
193 - select {
194 - case <-t.C:
195 - a.Errorf("stopping all goroutines timed out after %s. Exiting...", timeout)
196 - os.Exit(0)
197 - case <-done:
198 - }
199 - }()
176 +// QuitCh returns agent quit notifications (e.g., dump completion).
177 +func (a *Agent) QuitCh() <-chan struct{} {
178 + return a.quitCh
179 +}
180
201 - if exit {
202 - os.Exit(0)
203 - }
181 +// DumpModeDuration returns configured dump mode duration.
182 +func (a *Agent) DumpModeDuration() time.Duration {
183 + return a.dumpMode
184 +}
185
205 - time.Sleep(time.Second)
206 - }
186 +// TriggerDumpAnalysis prints dump analysis report.
187 +func (a *Agent) TriggerDumpAnalysis() {
188 + a.collectDumpAnalysis()
189 }
190
191 func (a *Agent) run(ctx context.Context) {
@@ -215,9 +197,6 @@ func (a *Agent) run(ctx context.Context) {
197
198 if !cfg.Enabled {
199 a.Info("plugin is disabled in the configuration file, exiting...")
218 - if terminal.IsTerminal() {
219 - os.Exit(0)
220 - }
200 a.api.DISABLE()
201 return
202 }
@@ -225,9 +204,6 @@ func (a *Agent) run(ctx context.Context) {
204 enabledModules := a.loadEnabledModules(cfg)
205 if len(enabledModules) == 0 {
206 a.Info("no modules to run")
228 - if terminal.IsTerminal() {
229 - os.Exit(0)
230 - }
207 a.api.DISABLE()
208 return
209 }
@@ -239,9 +215,6 @@ func (a *Agent) run(ctx context.Context) {
215 discMgr, err := discovery.NewManager(discCfg)
216 if err != nil {
217 a.Error(err)
242 - if terminal.IsTerminal() {
243 - os.Exit(0)
244 - }
218 return
219 }
220
@@ -257,6 +230,7 @@ func (a *Agent) run(ctx context.Context) {
230 jobMgr := jobmgr.New(jobmgr.Config{
231 PluginName: a.Name,
232 Out: a.Out,
233 + RunModePolicy: a.runModePolicy,
234 Modules: enabledModules,
235 RunJob: runJob,
236 ConfigDefaults: discCfg.Registry,
@@ -288,28 +262,6 @@ func (a *Agent) run(ctx context.Context) {
262 <-ctx.Done()
263 }
264
291 -func (a *Agent) keepAlive() {
292 - if terminal.IsTerminal() {
293 - return
294 - }
295 -
296 - tk := time.NewTicker(time.Second)
297 - defer tk.Stop()
298 -
299 - var n int
300 - for range tk.C {
301 - if err := a.api.EMPTYLINE(); err != nil {
302 - n++
303 - } else {
304 - n = 0
305 - }
306 - if n >= 30 {
307 - a.Info("too many keepAlive errors. Terminating...")
308 - os.Exit(0)
309 - }
310 - }
311 -}
312 -
265 func (a *Agent) collectDumpAnalysis() {
266 if a.dumpAnalyzer == nil || a.mgr == nil {
267 a.Error("dump analyzer or job manager not initialized")
src/go/plugin/agent/agent_test.go
+35 -2
@@ -10,18 +10,51 @@ import (
10 "time"
11
12 "github.com/netdata/netdata/go/plugins/pkg/safewriter"
13 + "github.com/netdata/netdata/go/plugins/plugin/agent/discovery"
14 + "github.com/netdata/netdata/go/plugins/plugin/agent/discovery/dummy"
15 + "github.com/netdata/netdata/go/plugins/plugin/agent/policy"
16 "github.com/netdata/netdata/go/plugins/plugin/framework/collectorapi"
17
18 "github.com/stretchr/testify/assert"
19 )
20
18 -// TODO: tech debt
21 func TestNew(t *testing.T) {
22 + t.Run("uses injected module registry", func(t *testing.T) {
23 + reg := prepareRegistry(&sync.Mutex{}, map[string]int{}, "module1")
24 + a := New(Config{Name: "test", ModuleRegistry: reg})
25 + assert.Equal(t, reg, a.ModuleRegistry)
26 + })
27
28 + t.Run("keeps nil module registry when not provided", func(t *testing.T) {
29 + a := New(Config{Name: "test"})
30 + assert.Nil(t, a.ModuleRegistry)
31 + })
32 }
33
34 func TestAgent_Run(t *testing.T) {
24 - a := New(Config{Name: "nodyncfg"})
35 + a := New(Config{
36 + Name: "test",
37 + RunModePolicy: policy.RunModePolicy{
38 + IsTerminal: false,
39 + AutoEnableDiscovered: true,
40 + UseFileStatusPersistence: true,
41 + },
42 + DiscoveryProviders: []discovery.ProviderFactory{
43 + discovery.NewProviderFactory("dummy", func(ctx discovery.BuildContext) (discovery.Discoverer, bool, error) {
44 + if len(ctx.DummyNames) == 0 {
45 + return nil, false, nil
46 + }
47 + d, err := dummy.NewDiscovery(dummy.Config{
48 + Registry: ctx.Registry,
49 + Names: ctx.DummyNames,
50 + })
51 + if err != nil {
52 + return nil, false, err
53 + }
54 + return d, true, nil
55 + }),
56 + },
57 + })
58
59 var buf bytes.Buffer
60 a.Out = safewriter.New(&buf)
src/go/plugin/agent/discovery/config.go
+35 -8
@@ -4,25 +4,52 @@ package discovery
4
5 import (
6 "errors"
7 + "io"
8
8 - "github.com/netdata/netdata/go/plugins/plugin/agent/discovery/dummy"
9 - "github.com/netdata/netdata/go/plugins/plugin/agent/discovery/file"
10 - "github.com/netdata/netdata/go/plugins/plugin/agent/discovery/sd"
9 + "github.com/netdata/netdata/go/plugins/pkg/multipath"
10 + "github.com/netdata/netdata/go/plugins/plugin/agent/policy"
11 "github.com/netdata/netdata/go/plugins/plugin/framework/confgroup"
12 + "github.com/netdata/netdata/go/plugins/plugin/framework/functions"
13 +)
14 +
15 +type (
16 + PlatformPolicy struct {
17 + IsInsideK8s bool
18 + }
19 + PluginIdentity struct {
20 + Name string
21 + }
22 + PathsConfig struct {
23 + PluginConfigDir multipath.MultiPath
24 + CollectorsConfigDir multipath.MultiPath
25 + CollectorsConfigWatchPath []string
26 + ServiceDiscoveryConfigDir multipath.MultiPath
27 + VarLibDir string
28 + }
29 + BuildContext struct {
30 + Policy PlatformPolicy
31 + RunMode policy.RunModePolicy
32 + Identity PluginIdentity
33 + Out io.Writer
34 + Paths PathsConfig
35 + Registry confgroup.Registry
36 + ReadPaths []string
37 + DummyNames []string
38 + FnReg functions.Registry
39 + }
40 )
41
42 type Config struct {
15 - Registry confgroup.Registry
16 - File file.Config
17 - Dummy dummy.Config
18 - SD sd.Config
43 + Registry confgroup.Registry
44 + BuildContext BuildContext
45 + Providers []ProviderFactory
46 }
47
48 func validateConfig(cfg Config) error {
49 if len(cfg.Registry) == 0 {
50 return errors.New("empty config registry")
51 }
25 - if len(cfg.File.Read)+len(cfg.File.Watch) == 0 && len(cfg.Dummy.Names) == 0 {
52 + if len(cfg.Providers) == 0 {
53 return errors.New("discoverers not set")
54 }
55 return nil
src/go/plugin/agent/discovery/file/read.go
+4 -4
@@ -7,9 +7,9 @@ import (
7 "fmt"
8 "os"
9 "path/filepath"
10 - "strings"
10
11 "github.com/netdata/netdata/go/plugins/logger"
12 + "github.com/netdata/netdata/go/plugins/pkg/pluginconfig"
13 "github.com/netdata/netdata/go/plugins/plugin/framework/confgroup"
14 )
15
@@ -91,8 +91,8 @@ func (r *Reader) groups() (groups []*confgroup.Group) {
91 }
92
93 func configSourceType(path string) string {
94 - if strings.Contains(path, "/etc/netdata") {
95 - return "user"
94 + if pluginconfig.IsStock(path) {
95 + return confgroup.TypeStock
96 }
97 - return "stock"
97 + return confgroup.TypeUser
98 }
src/go/plugin/agent/discovery/file/read_test.go
+22
@@ -115,3 +115,25 @@ func TestReader_Run(t *testing.T) {
115 })
116 }
117 }
118 +
119 +func TestConfigSourceType(t *testing.T) {
120 + tests := map[string]struct {
121 + path string
122 + want string
123 + }{
124 + "user path under /etc": {
125 + path: "/etc/netdata/go.d/module.conf",
126 + want: confgroup.TypeUser,
127 + },
128 + "stock path outside /etc": {
129 + path: "/usr/lib/netdata/conf.d/go.d/module.conf",
130 + want: confgroup.TypeStock,
131 + },
132 + }
133 +
134 + for name, tc := range tests {
135 + t.Run(name, func(t *testing.T) {
136 + assert.Equal(t, tc.want, configSourceType(tc.path))
137 + })
138 + }
139 +}
src/go/plugin/agent/discovery/manager.go
+11 -31
@@ -11,9 +11,6 @@ import (
11 "time"
12
13 "github.com/netdata/netdata/go/plugins/logger"
14 - "github.com/netdata/netdata/go/plugins/plugin/agent/discovery/dummy"
15 - "github.com/netdata/netdata/go/plugins/plugin/agent/discovery/file"
16 - "github.com/netdata/netdata/go/plugins/plugin/agent/discovery/sd"
14 "github.com/netdata/netdata/go/plugins/plugin/framework/confgroup"
15 )
16
@@ -28,7 +25,7 @@ func NewManager(cfg Config) (*Manager, error) {
25 ),
26 send: make(chan struct{}, 1),
27 sendEvery: time.Second * 2, // timeout to aggregate changes
31 - discoverers: make([]discoverer, 0),
28 + discoverers: make([]Discoverer, 0),
29 mux: &sync.RWMutex{},
30 cache: newCache(),
31 }
@@ -40,13 +37,9 @@ func NewManager(cfg Config) (*Manager, error) {
37 return mgr, nil
38 }
39
43 -type discoverer interface {
44 - Run(ctx context.Context, in chan<- []*confgroup.Group)
45 -}
46 -
40 type Manager struct {
41 *logger.Logger
49 - discoverers []discoverer
42 + discoverers []Discoverer
43 send chan struct{}
44 sendEvery time.Duration
45 mux *sync.RWMutex
@@ -65,7 +58,7 @@ func (m *Manager) Run(ctx context.Context, in chan<- []*confgroup.Group) {
58
59 for _, d := range m.discoverers {
60 wg.Add(1)
68 - go func(d discoverer) {
61 + go func(d Discoverer) {
62 defer wg.Done()
63 m.runDiscoverer(ctx, d)
64 }(d)
@@ -82,29 +75,16 @@ func (m *Manager) Run(ctx context.Context, in chan<- []*confgroup.Group) {
75 }
76
77 func (m *Manager) registerDiscoverers(cfg Config) error {
85 - if len(cfg.File.Read) > 0 || len(cfg.File.Watch) > 0 {
86 - cfg.File.Registry = cfg.Registry
87 - d, err := file.NewDiscovery(cfg.File)
78 + for _, provider := range cfg.Providers {
79 + d, enabled, err := provider.Build(cfg.BuildContext)
80 if err != nil {
89 - return err
81 + return fmt.Errorf("provider '%s': %w", provider.Name(), err)
82 }
91 - m.discoverers = append(m.discoverers, d)
92 - }
93 -
94 - if len(cfg.Dummy.Names) > 0 {
95 - cfg.Dummy.Registry = cfg.Registry
96 - d, err := dummy.NewDiscovery(cfg.Dummy)
97 - if err != nil {
98 - return err
83 + if !enabled {
84 + continue
85 }
100 - m.discoverers = append(m.discoverers, d)
101 - }
102 -
103 - if len(cfg.SD.ConfDir) != 0 {
104 - cfg.SD.ConfigDefaults = cfg.Registry
105 - d, err := sd.NewServiceDiscovery(cfg.SD)
106 - if err != nil {
107 - return err
86 + if d == nil {
87 + return fmt.Errorf("provider '%s': enabled but returned nil discoverer", provider.Name())
88 }
89 m.discoverers = append(m.discoverers, d)
90 }
@@ -118,7 +98,7 @@ func (m *Manager) registerDiscoverers(cfg Config) error {
98 return nil
99 }
100
121 -func (m *Manager) runDiscoverer(ctx context.Context, d discoverer) {
101 +func (m *Manager) runDiscoverer(ctx context.Context, d Discoverer) {
102 done := make(chan struct{})
103 updates := make(chan []*confgroup.Group)
104
src/go/plugin/agent/discovery/manager_test.go
+41 -4
@@ -9,7 +9,7 @@ import (
9 "testing"
10 "time"
11
12 - "github.com/netdata/netdata/go/plugins/plugin/agent/discovery/file"
12 + "github.com/netdata/netdata/go/plugins/plugin/agent/discovery/sd"
13 "github.com/netdata/netdata/go/plugins/plugin/framework/confgroup"
14 "github.com/stretchr/testify/assert"
15 "github.com/stretchr/testify/require"
@@ -23,12 +23,38 @@ func TestNewManager(t *testing.T) {
23 "valid config": {
24 cfg: Config{
25 Registry: confgroup.Registry{"module1": confgroup.Default{}},
26 - File: file.Config{Read: []string{"path"}},
26 + Providers: []ProviderFactory{
27 + NewProviderFactory("test", func(BuildContext) (Discoverer, bool, error) {
28 + return prepareMockDiscoverer("test", 1, 1), true, nil
29 + }),
30 + },
31 + },
32 + },
33 + "valid config, sd only": {
34 + cfg: Config{
35 + Registry: confgroup.Registry{"module1": confgroup.Default{}},
36 + Providers: []ProviderFactory{
37 + NewProviderFactory("sd", func(BuildContext) (Discoverer, bool, error) {
38 + d, err := sd.NewServiceDiscovery(sd.Config{
39 + PluginName: "test",
40 + ConfDir: []string{"path"},
41 + Discoverers: sd.NewRegistry(),
42 + })
43 + if err != nil {
44 + return nil, false, err
45 + }
46 + return d, true, nil
47 + }),
48 + },
49 },
50 },
51 "invalid config, registry not set": {
52 cfg: Config{
31 - File: file.Config{Read: []string{"path"}},
53 + Providers: []ProviderFactory{
54 + NewProviderFactory("test", func(BuildContext) (Discoverer, bool, error) {
55 + return prepareMockDiscoverer("test", 1, 1), true, nil
56 + }),
57 + },
58 },
59 wantErr: true,
60 },
@@ -38,6 +64,17 @@ func TestNewManager(t *testing.T) {
64 },
65 wantErr: true,
66 },
67 + "invalid config, all providers disabled": {
68 + cfg: Config{
69 + Registry: confgroup.Registry{"module1": confgroup.Default{}},
70 + Providers: []ProviderFactory{
71 + NewProviderFactory("disabled", func(BuildContext) (Discoverer, bool, error) {
72 + return nil, false, nil
73 + }),
74 + },
75 + },
76 + wantErr: true,
77 + },
78 }
79
80 for name, test := range tests {
@@ -142,7 +179,7 @@ func prepareMockDiscoverer(source string, groups, configs int) mockDiscoverer {
179 return d
180 }
181
145 -func prepareManager(discoverers ...discoverer) *Manager {
182 +func prepareManager(discoverers ...Discoverer) *Manager {
183 mgr := &Manager{
184 send: make(chan struct{}, 1),
185 sendEvery: 2 * time.Second,
src/go/plugin/agent/discovery/provider.go new
+42
@@ -0,0 +1,42 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package discovery
4 +
5 +import (
6 + "context"
7 + "fmt"
8 +
9 + "github.com/netdata/netdata/go/plugins/plugin/framework/confgroup"
10 +)
11 +
12 +// Discoverer is a discovery source runner.
13 +type Discoverer interface {
14 + Run(ctx context.Context, in chan<- []*confgroup.Group)
15 +}
16 +
17 +// ProviderFactory builds optional discoverers from a shared build context.
18 +type ProviderFactory interface {
19 + Name() string
20 + Build(ctx BuildContext) (Discoverer, bool, error)
21 +}
22 +
23 +type providerFactoryFunc struct {
24 + name string
25 + build func(ctx BuildContext) (Discoverer, bool, error)
26 +}
27 +
28 +func (p providerFactoryFunc) Name() string {
29 + return p.name
30 +}
31 +
32 +func (p providerFactoryFunc) Build(ctx BuildContext) (Discoverer, bool, error) {
33 + if p.build == nil {
34 + return nil, false, fmt.Errorf("provider %q has nil build function", p.name)
35 + }
36 + return p.build(ctx)
37 +}
38 +
39 +// NewProviderFactory creates a named provider factory.
40 +func NewProviderFactory(name string, build func(ctx BuildContext) (Discoverer, bool, error)) ProviderFactory {
41 + return providerFactoryFunc{name: name, build: build}
42 +}
src/go/plugin/agent/discovery/sd/dyncfg.go
+2 -3
@@ -7,7 +7,6 @@ import (
7 "fmt"
8 "strings"
9
10 - "github.com/netdata/netdata/go/plugins/pkg/executable"
10 "github.com/netdata/netdata/go/plugins/pkg/netdataapi"
11 "github.com/netdata/netdata/go/plugins/plugin/framework/confgroup"
12 "github.com/netdata/netdata/go/plugins/plugin/framework/dyncfg"
@@ -20,7 +19,7 @@ const (
19 )
20
21 func (d *ServiceDiscovery) dyncfgSDPrefixValue() string {
23 - return fmt.Sprintf(dyncfgSDPrefixf, executable.Name)
22 + return fmt.Sprintf(dyncfgSDPrefixf, d.pluginName)
23 }
24
25 func (d *ServiceDiscovery) dyncfgTemplateID(discovererType string) string {
@@ -45,7 +44,7 @@ func (d *ServiceDiscovery) dyncfgSDTemplateCreate(discovererType string) {
44 ID: d.dyncfgTemplateID(discovererType),
45 Status: dyncfg.StatusAccepted.String(),
46 ConfigType: dyncfg.ConfigTypeTemplate.String(),
48 - Path: fmt.Sprintf(dyncfgSDPath, executable.Name),
47 + Path: fmt.Sprintf(dyncfgSDPath, d.pluginName),
48 SourceType: "internal",
49 Source: "internal",
50 SupportedCommands: dyncfgSDTemplateCmds(),
src/go/plugin/agent/discovery/sd/dyncfg_cache.go
+4 -5
@@ -7,6 +7,7 @@ import (
7 "fmt"
8 "strings"
9
10 + "github.com/netdata/netdata/go/plugins/pkg/pluginconfig"
11 "github.com/netdata/netdata/go/plugins/plugin/agent/discovery/sd/pipeline"
12 "github.com/netdata/netdata/go/plugins/plugin/agent/internal/naming"
13 "github.com/netdata/netdata/go/plugins/plugin/framework/confgroup"
@@ -176,10 +177,8 @@ func newSDConfigFromJSON(data []byte, name, source, sourceType, discovererType,
177
178 // sourceTypeFromPath determines the source type (stock/user) from a file path.
179 func sourceTypeFromPath(path string) string {
179 - // User configs are in /etc/ (e.g., /etc/netdata/sd.d/)
180 - // Stock configs are in /usr/lib/ or similar system paths
181 - if strings.Contains(path, "/etc/") {
182 - return confgroup.TypeUser
180 + if pluginconfig.IsStock(path) {
181 + return confgroup.TypeStock
182 }
184 - return confgroup.TypeStock
183 + return confgroup.TypeUser
184 }
src/go/plugin/agent/discovery/sd/dyncfg_cache_test.go new
+32
@@ -0,0 +1,32 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package sd
4 +
5 +import (
6 + "testing"
7 +
8 + "github.com/netdata/netdata/go/plugins/plugin/framework/confgroup"
9 + "github.com/stretchr/testify/assert"
10 +)
11 +
12 +func TestSourceTypeFromPath(t *testing.T) {
13 + tests := map[string]struct {
14 + path string
15 + want string
16 + }{
17 + "user path under /etc": {
18 + path: "/etc/netdata/sd.d/test.conf",
19 + want: confgroup.TypeUser,
20 + },
21 + "stock path outside /etc": {
22 + path: "/usr/lib/netdata/conf.d/sd.d/test.conf",
23 + want: confgroup.TypeStock,
24 + },
25 + }
26 +
27 + for name, tc := range tests {
28 + t.Run(name, func(t *testing.T) {
29 + assert.Equal(t, tc.want, sourceTypeFromPath(tc.path))
30 + })
31 + }
32 +}
src/go/plugin/agent/discovery/sd/dyncfg_test.go
+2 -2
@@ -14,7 +14,6 @@ import (
14
15 "github.com/netdata/netdata/go/plugins/logger"
16 "github.com/netdata/netdata/go/plugins/pkg/confopt"
17 - "github.com/netdata/netdata/go/plugins/pkg/executable"
17 "github.com/netdata/netdata/go/plugins/pkg/netdataapi"
18 "github.com/netdata/netdata/go/plugins/pkg/safewriter"
19 "github.com/netdata/netdata/go/plugins/plugin/agent/discovery/sd/pipeline"
@@ -105,6 +104,7 @@ func (s *dyncfgSim) run(t *testing.T) {
104 var buf bytes.Buffer
105 sd := &ServiceDiscovery{
106 Logger: logger.New(),
107 + pluginName: testPluginName,
108 dyncfgApi: dyncfg.NewResponder(netdataapi.New(safewriter.New(&buf))),
109 seen: dyncfg.NewSeenCache[sdConfig](),
110 exposed: dyncfg.NewExposedCache[sdConfig](),
@@ -122,7 +122,7 @@ func (s *dyncfgSim) run(t *testing.T) {
122 Exposed: sd.exposed,
123 Callbacks: sd.sdCb,
124
125 - Path: fmt.Sprintf(dyncfgSDPath, executable.Name),
125 + Path: fmt.Sprintf(dyncfgSDPath, testPluginName),
126 EnableFailCode: 422,
127 JobCommands: []dyncfg.Command{
128 dyncfg.CommandSchema,
src/go/plugin/agent/discovery/sd/output_test.go new
+38
@@ -0,0 +1,38 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package sd
4 +
5 +import (
6 + "bytes"
7 + "fmt"
8 + "testing"
9 +
10 + "github.com/netdata/netdata/go/plugins/pkg/netdataapi"
11 + "github.com/netdata/netdata/go/plugins/plugin/framework/dyncfg"
12 + "github.com/stretchr/testify/assert"
13 + "github.com/stretchr/testify/require"
14 +)
15 +
16 +func TestNewServiceDiscovery_UsesConfiguredOutForDyncfgResponder(t *testing.T) {
17 + const pluginName = "test"
18 +
19 + var buf bytes.Buffer
20 + sd, err := NewServiceDiscovery(Config{
21 + PluginName: pluginName,
22 + Out: &buf,
23 + Discoverers: NewRegistry(),
24 + })
25 + require.NoError(t, err)
26 +
27 + sd.dyncfgApi.ConfigCreate(netdataapi.ConfigOpts{
28 + ID: "test:sd:discoverer",
29 + Status: dyncfg.StatusAccepted.String(),
30 + ConfigType: dyncfg.ConfigTypeTemplate.String(),
31 + Path: fmt.Sprintf(dyncfgSDPath, pluginName),
32 + SourceType: "internal",
33 + Source: "internal",
34 + SupportedCommands: "schema",
35 + })
36 +
37 + assert.Contains(t, buf.String(), "CONFIG test:sd:discoverer create accepted template /collectors/test/ServiceDiscovery")
38 +}
src/go/plugin/agent/discovery/sd/sd.go
+17 -7
@@ -5,24 +5,26 @@ package sd
5 import (
6 "context"
7 "fmt"
8 + "io"
9 "log/slog"
10 "sync"
11
11 - "github.com/netdata/netdata/go/plugins/pkg/terminal"
12 "github.com/netdata/netdata/go/plugins/plugin/agent/discovery/sd/pipeline"
13 + "github.com/netdata/netdata/go/plugins/plugin/agent/policy"
14 "github.com/netdata/netdata/go/plugins/plugin/framework/confgroup"
15 "github.com/netdata/netdata/go/plugins/plugin/framework/dyncfg"
16 "github.com/netdata/netdata/go/plugins/plugin/framework/functions"
17
18 "github.com/netdata/netdata/go/plugins/logger"
18 - "github.com/netdata/netdata/go/plugins/pkg/executable"
19 "github.com/netdata/netdata/go/plugins/pkg/multipath"
20 "github.com/netdata/netdata/go/plugins/pkg/netdataapi"
21 - "github.com/netdata/netdata/go/plugins/pkg/safewriter"
21 )
22
23 type Config struct {
24 ConfigDefaults confgroup.Registry
25 + PluginName string
26 + RunModePolicy policy.RunModePolicy
27 + Out io.Writer
28 ConfDir multipath.MultiPath
29 FnReg functions.Registry
30 Discoverers Registry
@@ -35,14 +37,20 @@ func NewServiceDiscovery(cfg Config) (*ServiceDiscovery, error) {
37 if cfg.Discoverers == nil {
38 return nil, fmt.Errorf("service discovery discoverer registry is not configured")
39 }
40 + out := cfg.Out
41 + if out == nil {
42 + out = io.Discard
43 + }
44
45 d := &ServiceDiscovery{
46 Logger: log,
47 confProv: newConfFileReader(log, cfg.ConfDir),
48 configDefaults: cfg.ConfigDefaults,
49 + pluginName: cfg.PluginName,
50 + runModePolicy: cfg.RunModePolicy,
51 fnReg: cfg.FnReg,
52 discoverers: cfg.Discoverers,
45 - dyncfgApi: dyncfg.NewResponder(netdataapi.New(safewriter.Stdout)),
53 + dyncfgApi: dyncfg.NewResponder(netdataapi.New(out)),
54 seen: dyncfg.NewSeenCache[sdConfig](),
55 exposed: dyncfg.NewExposedCache[sdConfig](),
56 dyncfgCh: make(chan dyncfg.Function, 1),
@@ -61,7 +69,7 @@ func NewServiceDiscovery(cfg Config) (*ServiceDiscovery, error) {
69 return cfg.PipelineKey()
70 },
71
64 - Path: fmt.Sprintf(dyncfgSDPath, executable.Name),
72 + Path: fmt.Sprintf(dyncfgSDPath, cfg.PluginName),
73 EnableFailCode: 422,
74 JobCommands: []dyncfg.Command{
75 dyncfg.CommandSchema,
@@ -84,6 +92,8 @@ type (
92 confProv confFileProvider
93
94 configDefaults confgroup.Registry
95 + pluginName string
96 + runModePolicy policy.RunModePolicy
97 fnReg functions.Registry
98 discoverers Registry
99 dyncfgApi *dyncfg.Responder
@@ -273,7 +283,7 @@ func (d *ServiceDiscovery) addConfig(ctx context.Context, scfg sdConfig) {
283 d.handler.AddDiscoveredConfig(scfg, dyncfg.StatusAccepted)
284
285 d.handler.NotifyJobCreate(scfg, dyncfg.StatusAccepted)
276 - if terminal.IsTerminal() || d.fnReg == nil || d.dyncfgCh == nil {
286 + if d.runModePolicy.AutoEnableDiscovered || d.fnReg == nil || d.dyncfgCh == nil {
287 // Auto-enable in terminal mode and tests.
288 // Also auto-enable when no function registry is attached, because
289 // no external enable/disable commands can be delivered.
@@ -309,7 +319,7 @@ func (d *ServiceDiscovery) addConfig(ctx context.Context, scfg sdConfig) {
319 d.handler.NotifyJobRemove(entry.Cfg)
320 d.handler.NotifyJobCreate(scfg, dyncfg.StatusAccepted)
321
312 - if terminal.IsTerminal() || d.fnReg == nil || d.dyncfgCh == nil {
322 + if d.runModePolicy.AutoEnableDiscovered || d.fnReg == nil || d.dyncfgCh == nil {
323 d.autoEnableConfig(scfg)
324 } else {
325 d.handler.WaitForDecision(scfg)
src/go/plugin/agent/discovery/sd/sim_test.go
+8 -5
@@ -12,7 +12,6 @@ import (
12 "time"
13
14 "github.com/netdata/netdata/go/plugins/logger"
15 - "github.com/netdata/netdata/go/plugins/pkg/executable"
15 "github.com/netdata/netdata/go/plugins/pkg/netdataapi"
16 "github.com/netdata/netdata/go/plugins/pkg/safewriter"
17 "github.com/netdata/netdata/go/plugins/plugin/agent/discovery/sd/pipeline"
@@ -24,6 +23,8 @@ import (
23
24 var lock = &sync.Mutex{}
25
26 +const testPluginName = "test"
27 +
28 type discoverySim struct {
29 configs []confFile
30 wantPipelines []*mockPipeline
@@ -48,7 +49,8 @@ func (sim *discoverySimExt) run(t *testing.T) {
49 fact := &mockFactory{}
50 var buf bytes.Buffer
51 mgr := &ServiceDiscovery{
51 - Logger: logger.New(),
52 + Logger: logger.New(),
53 + pluginName: testPluginName,
54 newPipeline: func(config pipeline.Config) (sdPipeline, error) {
55 return fact.create(config)
56 },
@@ -70,7 +72,7 @@ func (sim *discoverySimExt) run(t *testing.T) {
72 Exposed: mgr.exposed,
73 Callbacks: mgr.sdCb,
74
73 - Path: fmt.Sprintf(dyncfgSDPath, executable.Name),
75 + Path: fmt.Sprintf(dyncfgSDPath, testPluginName),
76 EnableFailCode: 422,
77 JobCommands: []dyncfg.Command{
78 dyncfg.CommandSchema,
@@ -123,7 +125,8 @@ func (sim *discoverySim) run(t *testing.T) {
125 fact := &mockFactory{}
126 var buf bytes.Buffer
127 mgr := &ServiceDiscovery{
126 - Logger: logger.New(),
128 + Logger: logger.New(),
129 + pluginName: testPluginName,
130 newPipeline: func(config pipeline.Config) (sdPipeline, error) {
131 return fact.create(config)
132 },
@@ -146,7 +149,7 @@ func (sim *discoverySim) run(t *testing.T) {
149 Exposed: mgr.exposed,
150 Callbacks: mgr.sdCb,
151
149 - Path: fmt.Sprintf(dyncfgSDPath, executable.Name),
152 + Path: fmt.Sprintf(dyncfgSDPath, testPluginName),
153 EnableFailCode: 422,
154 JobCommands: []dyncfg.Command{
155 dyncfg.CommandSchema,
src/go/plugin/agent/jobmgr/dyncfg_collector.go
+2 -3
@@ -13,7 +13,6 @@ import (
13 "gopkg.in/yaml.v2"
14
15 "github.com/netdata/netdata/go/plugins/logger"
16 - "github.com/netdata/netdata/go/plugins/pkg/executable"
16 "github.com/netdata/netdata/go/plugins/pkg/netdataapi"
17 "github.com/netdata/netdata/go/plugins/plugin/framework/confgroup"
18 "github.com/netdata/netdata/go/plugins/plugin/framework/dyncfg"
@@ -25,7 +24,7 @@ const (
24 )
25
26 func (m *Manager) dyncfgCollectorPrefixValue() string {
28 - return fmt.Sprintf(dyncfgCollectorPrefixf, executable.Name)
27 + return fmt.Sprintf(dyncfgCollectorPrefixf, m.pluginName)
28 }
29
30 func (m *Manager) dyncfgModID(name string) string {
@@ -51,7 +50,7 @@ func (m *Manager) dyncfgCollectorModuleCreate(name string) {
50 ID: m.dyncfgModID(name),
51 Status: dyncfg.StatusAccepted.String(),
52 ConfigType: dyncfg.ConfigTypeTemplate.String(),
54 - Path: fmt.Sprintf(dyncfgCollectorPath, executable.Name),
53 + Path: fmt.Sprintf(dyncfgCollectorPath, m.pluginName),
54 SourceType: "internal",
55 Source: "internal",
56 SupportedCommands: dyncfgCollectorModCmds(),
src/go/plugin/agent/jobmgr/dyncfg_collector_test.go
+1 -1
@@ -30,7 +30,7 @@ func TestDyncfgConfigUserconfig_InvalidPayload_Returns400Only(t *testing.T) {
30 t.Run(name, func(t *testing.T) {
31 var buf bytes.Buffer
32
33 - mgr := New(Config{})
33 + mgr := New(Config{PluginName: testPluginName})
34 mgr.modules = prepareMockRegistry()
35 mgr.SetDyncfgResponder(dyncfg.NewResponder(netdataapi.New(safewriter.New(&buf))))
36
src/go/plugin/agent/jobmgr/dyncfg_test.go
+1 -1
@@ -32,7 +32,7 @@ func TestDyncfgConfig_ShutdownDoesNotQueue(t *testing.T) {
32 t.Run(name, func(t *testing.T) {
33 var buf bytes.Buffer
34
35 - mgr := New(Config{})
35 + mgr := New(Config{PluginName: testPluginName})
36 mgr.SetDyncfgResponder(dyncfg.NewResponder(netdataapi.New(safewriter.New(&buf))))
37
38 ctx, cancel := context.WithCancel(context.Background())
src/go/plugin/agent/jobmgr/dyncfg_vnode.go
+3 -4
@@ -11,7 +11,6 @@ import (
11 "github.com/netdata/netdata/go/plugins/plugin/framework/vnodes"
12 "gopkg.in/yaml.v2"
13
14 - "github.com/netdata/netdata/go/plugins/pkg/executable"
14 "github.com/netdata/netdata/go/plugins/pkg/netdataapi"
15 "github.com/netdata/netdata/go/plugins/plugin/framework/confgroup"
16 "github.com/netdata/netdata/go/plugins/plugin/framework/dyncfg"
@@ -23,7 +22,7 @@ const (
22 )
23
24 func (m *Manager) dyncfgVnodePrefixValue() string {
26 - return fmt.Sprintf(dyncfgVnodeIDf, executable.Name)
25 + return fmt.Sprintf(dyncfgVnodeIDf, m.pluginName)
26 }
27
28 func dyncfgVnodeModCmds() string {
@@ -53,7 +52,7 @@ func (m *Manager) dyncfgVnodeModuleCreate() {
52 ID: m.dyncfgVnodePrefixValue(),
53 Status: dyncfg.StatusAccepted.String(),
54 ConfigType: dyncfg.ConfigTypeTemplate.String(),
56 - Path: fmt.Sprintf(dyncfgVnodePath, executable.Name),
55 + Path: fmt.Sprintf(dyncfgVnodePath, m.pluginName),
56 SourceType: "internal",
57 Source: "internal",
58 SupportedCommands: dyncfgVnodeModCmds(),
@@ -65,7 +64,7 @@ func (m *Manager) dyncfgVnodeJobCreate(cfg *vnodes.VirtualNode, status dyncfg.St
64 ID: fmt.Sprintf("%s:%s", m.dyncfgVnodePrefixValue(), cfg.Name),
65 Status: status.String(),
66 ConfigType: dyncfg.ConfigTypeJob.String(),
68 - Path: fmt.Sprintf(dyncfgVnodePath, executable.Name),
67 + Path: fmt.Sprintf(dyncfgVnodePath, m.pluginName),
68 SourceType: cfg.SourceType,
69 Source: cfg.Source,
70 SupportedCommands: dyncfgVnodeJobCmds(cfg.SourceType == confgroup.TypeDyncfg),
src/go/plugin/agent/jobmgr/filestatus.go
+6 -8
@@ -11,25 +11,23 @@ import (
11 "strings"
12 "sync"
13
14 - "github.com/netdata/netdata/go/plugins/pkg/executable"
15 - "github.com/netdata/netdata/go/plugins/pkg/terminal"
14 "github.com/netdata/netdata/go/plugins/plugin/framework/confgroup"
15 "github.com/netdata/netdata/go/plugins/plugin/framework/filepersister"
16 )
17
20 -func statusFileName(dir string) string {
21 - name := strings.ReplaceAll(executable.Name, ".", "")
18 +func statusFileName(dir, pluginName string) string {
19 + name := strings.ReplaceAll(pluginName, ".", "")
20 return filepath.Join(dir, fmt.Sprintf("%s-jobs-statuses.json", name))
21 }
22
23 func (m *Manager) loadFileStatus() {
24 m.fileStatus = newFileStatus()
25
28 - if terminal.IsTerminal() || m.varLibDir == "" {
26 + if !m.runModePolicy.UseFileStatusPersistence || m.varLibDir == "" {
27 return
28 }
29
32 - s, err := loadFileStatus(statusFileName(m.varLibDir))
30 + s, err := loadFileStatus(statusFileName(m.varLibDir, m.pluginName))
31 if err != nil {
32 m.Warningf("failed to load state file: %v", err)
33 return
@@ -38,11 +36,11 @@ func (m *Manager) loadFileStatus() {
36 }
37
38 func (m *Manager) runFileStatusPersistence() {
41 - if m.varLibDir == "" {
39 + if !m.runModePolicy.UseFileStatusPersistence || m.varLibDir == "" {
40 return
41 }
42
45 - p := filepersister.New(statusFileName(m.varLibDir))
43 + p := filepersister.New(statusFileName(m.varLibDir, m.pluginName))
44
45 p.Run(m.ctx, m.fileStatus)
46 }
src/go/plugin/agent/jobmgr/manager.go
+10 -9
@@ -14,13 +14,11 @@ import (
14 "time"
15
16 "github.com/netdata/netdata/go/plugins/logger"
17 - "github.com/netdata/netdata/go/plugins/pkg/executable"
17 "github.com/netdata/netdata/go/plugins/pkg/funcapi"
18 "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/terminal"
19 "github.com/netdata/netdata/go/plugins/pkg/ticker"
20 "github.com/netdata/netdata/go/plugins/plugin/agent/internal/naming"
21 + "github.com/netdata/netdata/go/plugins/plugin/agent/policy"
22 "github.com/netdata/netdata/go/plugins/plugin/framework/collectorapi"
23 "github.com/netdata/netdata/go/plugins/plugin/framework/confgroup"
24 "github.com/netdata/netdata/go/plugins/plugin/framework/dyncfg"
@@ -34,6 +32,7 @@ import (
32 type Config struct {
33 PluginName string
34 Out io.Writer
35 + RunModePolicy policy.RunModePolicy
36 Modules collectorapi.Registry
37 RunJob []string
38 ConfigDefaults confgroup.Registry
@@ -48,14 +47,14 @@ type Config struct {
47 }
48
49 func New(cfg Config) *Manager {
51 - seen := dyncfg.NewSeenCache[confgroup.Config]()
52 - exposed := dyncfg.NewExposedCache[confgroup.Config]()
53 - api := dyncfg.NewResponder(netdataapi.New(safewriter.Stdout))
54 -
50 out := cfg.Out
51 if out == nil {
52 out = io.Discard
53 }
54 +
55 + seen := dyncfg.NewSeenCache[confgroup.Config]()
56 + exposed := dyncfg.NewExposedCache[confgroup.Config]()
57 + api := dyncfg.NewResponder(netdataapi.New(out))
58 fnReg := cfg.FnReg
59 if fnReg == nil {
60 fnReg = noop{}
@@ -71,6 +70,7 @@ func New(cfg Config) *Manager {
70 ),
71 pluginName: cfg.PluginName,
72 out: out,
73 + runModePolicy: cfg.RunModePolicy,
74 modules: cfg.Modules,
75 runJob: cfg.RunJob,
76 configDefaults: cfg.ConfigDefaults,
@@ -109,7 +109,7 @@ func New(cfg Config) *Manager {
109 return cfg.FullName()
110 },
111
112 - Path: fmt.Sprintf(dyncfgCollectorPath, executable.Name),
112 + Path: fmt.Sprintf(dyncfgCollectorPath, cfg.PluginName),
113 EnableFailCode: 200,
114 RemoveStockOnEnableFail: true,
115 JobCommands: []dyncfg.Command{
@@ -137,6 +137,7 @@ type Manager struct {
137
138 pluginName string
139 out io.Writer
140 + runModePolicy policy.RunModePolicy
141 modules collectorapi.Registry
142 runJob []string
143 configDefaults confgroup.Registry
@@ -358,7 +359,7 @@ func (m *Manager) addConfig(cfg confgroup.Config) {
359
360 m.handler.NotifyJobCreate(entry.Cfg, entry.Status)
361
361 - if terminal.IsTerminal() || m.pluginName == "nodyncfg" { // FIXME: quick fix of TestAgent_Run (agent_test.go)
362 + if m.runModePolicy.AutoEnableDiscovered {
363 m.handler.CmdEnable(dyncfg.NewFunction(functions.Function{Args: []string{m.dyncfgJobID(entry.Cfg), "enable"}}))
364 } else {
365 m.handler.WaitForDecision(entry.Cfg)
src/go/plugin/agent/jobmgr/manager_process_test.go
+1 -1
@@ -30,7 +30,7 @@ func TestRunProcessConfGroups_ChannelCloseDoesNotSpin(t *testing.T) {
30
31 for name, tc := range tests {
32 t.Run(name, func(t *testing.T) {
33 - mgr := New(Config{})
33 + mgr := New(Config{PluginName: testPluginName})
34 ctx, cancel := context.WithCancel(context.Background())
35 t.Cleanup(cancel)
36 mgr.ctx = ctx
src/go/plugin/agent/jobmgr/manager_v2_test.go
+1 -1
@@ -103,7 +103,7 @@ func TestManagerCreateCollectorJobV2Branching(t *testing.T) {
103
104 for name, tc := range tests {
105 t.Run(name, func(t *testing.T) {
106 - mgr := New(Config{})
106 + mgr := New(Config{PluginName: testPluginName})
107 mgr.modules = collectorapi.Registry{
108 "testmod": tc.creator,
109 }
src/go/plugin/agent/jobmgr/output_test.go new
+35
@@ -0,0 +1,35 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package jobmgr
4 +
5 +import (
6 + "bytes"
7 + "fmt"
8 + "testing"
9 +
10 + "github.com/netdata/netdata/go/plugins/pkg/netdataapi"
11 + "github.com/netdata/netdata/go/plugins/plugin/framework/dyncfg"
12 + "github.com/stretchr/testify/assert"
13 +)
14 +
15 +func TestNew_UsesConfiguredOutForDyncfgResponder(t *testing.T) {
16 + const pluginName = "test"
17 +
18 + var buf bytes.Buffer
19 + mgr := New(Config{
20 + PluginName: pluginName,
21 + Out: &buf,
22 + })
23 +
24 + mgr.dyncfgApi.ConfigCreate(netdataapi.ConfigOpts{
25 + ID: "test:collector:module",
26 + Status: dyncfg.StatusAccepted.String(),
27 + ConfigType: dyncfg.ConfigTypeTemplate.String(),
28 + Path: fmt.Sprintf(dyncfgCollectorPath, pluginName),
29 + SourceType: "internal",
30 + Source: "internal",
31 + SupportedCommands: "schema",
32 + })
33 +
34 + assert.Contains(t, buf.String(), "CONFIG test:collector:module create accepted template /collectors/test/Jobs")
35 +}
src/go/plugin/agent/jobmgr/sim_test.go
+3 -1
@@ -21,6 +21,8 @@ import (
21 "github.com/stretchr/testify/require"
22 )
23
24 +const testPluginName = "test"
25 +
26 type wantExposedEntry struct {
27 cfg confgroup.Config
28 status dyncfg.Status
@@ -42,7 +44,7 @@ func (s *runSim) run(t *testing.T) {
44 require.NotNil(t, s.do, "s.do is nil")
45
46 var buf bytes.Buffer
45 - mgr := New(Config{})
47 + mgr := New(Config{PluginName: testPluginName})
48 mgr.SetDyncfgResponder(dyncfg.NewResponder(netdataapi.New(safewriter.New(&buf))))
49 mgr.modules = prepareMockRegistry()
50
src/go/plugin/agent/policy/runmode.go new
+10
@@ -0,0 +1,10 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package policy
4 +
5 +// RunModePolicy defines runtime-mode behavior gates injected from composition root.
6 +type RunModePolicy struct {
7 + IsTerminal bool
8 + AutoEnableDiscovered bool
9 + UseFileStatusPersistence bool
10 +}
src/go/plugin/agent/setup.go
+48 -43
@@ -5,18 +5,13 @@ package agent
5 import (
6 "io"
7 "os"
8 - "strings"
8
10 - hostinfo2 "github.com/netdata/netdata/go/plugins/pkg/hostinfo"
9 + "github.com/netdata/netdata/go/plugins/pkg/pluginconfig"
10 "github.com/netdata/netdata/go/plugins/plugin/agent/discovery"
12 - "github.com/netdata/netdata/go/plugins/plugin/agent/discovery/dummy"
13 - "github.com/netdata/netdata/go/plugins/plugin/agent/discovery/file"
14 - "github.com/netdata/netdata/go/plugins/plugin/agent/discovery/sd"
11 "github.com/netdata/netdata/go/plugins/plugin/framework/collectorapi"
12 "github.com/netdata/netdata/go/plugins/plugin/framework/confgroup"
13 "github.com/netdata/netdata/go/plugins/plugin/framework/functions"
14 "github.com/netdata/netdata/go/plugins/plugin/framework/vnodes"
19 - "github.com/netdata/netdata/go/plugins/plugin/go.d/discovery/sdext"
15 "gopkg.in/yaml.v2"
16 )
17
@@ -58,8 +53,7 @@ func (a *Agent) loadEnabledModules(cfg config) collectorapi.Registry {
53 continue
54 }
55 if all {
61 - // Known issue: go.d/logind high CPU usage on Alma Linux8 (https://github.com/netdata/netdata/issues/15930)
62 - if !cfg.isExplicitlyEnabled(name) && (creator.Disabled || name == "logind" && hostinfo2.SystemdVersion == 239) {
56 + if !cfg.isExplicitlyEnabled(name) && creator.Disabled {
57 a.Infof("'%s' module disabled by default, should be explicitly enabled in the config", name)
58 continue
59 }
@@ -91,18 +85,52 @@ func (a *Agent) buildDiscoveryConf(enabled collectorapi.Registry, fnReg function
85
86 var readPaths, dummyPaths []string
87
88 + watchPaths := a.CollectorsConfigWatchPath
89 + sdConfDir := a.ServiceDiscoveryConfigDir
90 + if a.DisableServiceDiscovery {
91 + dummyPaths = nil
92 + sdConfDir = nil
93 + }
94 +
95 + cfg := discovery.Config{
96 + Registry: reg,
97 + BuildContext: discovery.BuildContext{
98 + Policy: discovery.PlatformPolicy{
99 + IsInsideK8s: a.IsInsideK8s,
100 + },
101 + RunMode: a.runModePolicy,
102 + Identity: discovery.PluginIdentity{
103 + Name: a.Name,
104 + },
105 + Out: a.Out,
106 + Paths: discovery.PathsConfig{
107 + PluginConfigDir: a.ConfigDir,
108 + CollectorsConfigDir: a.CollectorsConfDir,
109 + CollectorsConfigWatchPath: watchPaths,
110 + ServiceDiscoveryConfigDir: sdConfDir,
111 + VarLibDir: a.VarLibDir,
112 + },
113 + Registry: reg,
114 + ReadPaths: readPaths,
115 + DummyNames: dummyPaths,
116 + FnReg: fnReg,
117 + },
118 + Providers: append([]discovery.ProviderFactory(nil), a.DiscoveryProviders...),
119 + }
120 +
121 if len(a.CollectorsConfDir) == 0 {
95 - if hostinfo2.IsInsideK8sCluster() {
96 - return discovery.Config{Registry: reg}
122 + if a.IsInsideK8s {
123 + cfg.Providers = nil
124 + return cfg
125 }
126 a.Info("modules conf dir not provided, will use default config for all enabled modules")
127 for name := range enabled {
128 dummyPaths = append(dummyPaths, name)
129 }
102 - return discovery.Config{
103 - Registry: reg,
104 - Dummy: dummy.Config{Names: dummyPaths},
105 - }
130 + watchPaths = nil
131 + cfg.BuildContext.Paths.CollectorsConfigWatchPath = watchPaths
132 + cfg.BuildContext.DummyNames = dummyPaths
133 + return cfg
134 }
135
136 for name := range enabled {
@@ -110,7 +138,7 @@ func (a *Agent) buildDiscoveryConf(enabled collectorapi.Registry, fnReg function
138 a.Debugf("looking for '%s' in %v", cfgName, a.CollectorsConfDir)
139
140 path, err := a.CollectorsConfDir.Find(cfgName)
113 - if hostinfo2.IsInsideK8sCluster() {
141 + if a.IsInsideK8s {
142 if err != nil {
143 a.Infof("not found '%s', won't use default (reading stock configs is disabled in k8s)", cfgName)
144 continue
@@ -129,26 +157,10 @@ func (a *Agent) buildDiscoveryConf(enabled collectorapi.Registry, fnReg function
157 }
158
159 a.Infof("dummy/read/watch paths: %d/%d/%d", len(dummyPaths), len(readPaths), len(a.CollectorsConfigWatchPath))
132 -
133 - cfg := discovery.Config{
134 - Registry: reg,
135 - File: file.Config{
136 - Read: readPaths,
137 - Watch: a.CollectorsConfigWatchPath,
138 - },
139 - }
140 -
141 - if !a.DisableServiceDiscovery {
142 - cfg.Dummy = dummy.Config{
143 - Names: dummyPaths,
144 - }
145 - cfg.SD = sd.Config{
146 - ConfigDefaults: reg,
147 - ConfDir: a.ServiceDiscoveryConfigDir,
148 - FnReg: fnReg,
149 - Discoverers: sdext.Registry(),
150 - }
151 - }
160 + cfg.BuildContext.Paths.CollectorsConfigWatchPath = watchPaths
161 + cfg.BuildContext.Paths.ServiceDiscoveryConfigDir = sdConfDir
162 + cfg.BuildContext.ReadPaths = readPaths
163 + cfg.BuildContext.DummyNames = dummyPaths
164
165 return cfg
166 }
@@ -186,13 +198,6 @@ func loadYAML(conf any, path string) error {
198 return nil
199 }
200
189 -var (
190 - envNDStockConfigDir = os.Getenv("NETDATA_STOCK_CONFIG_DIR")
191 -)
192 -
201 func isStockConfig(path string) bool {
194 - if envNDStockConfigDir == "" {
195 - return false
196 - }
197 - return strings.HasPrefix(path, envNDStockConfigDir)
202 + return pluginconfig.IsStock(path)
203 }
src/go/plugin/agent/setup_test.go
+33 -1
@@ -5,6 +5,7 @@ package agent
5 import (
6 "testing"
7
8 + "github.com/netdata/netdata/go/plugins/plugin/agent/discovery"
9 "github.com/netdata/netdata/go/plugins/plugin/framework/collectorapi"
10
11 "github.com/stretchr/testify/assert"
@@ -203,7 +204,38 @@ func TestAgent_loadEnabledModules(t *testing.T) {
204 }
205 }
206
206 -// TODO: tech debt
207 +func TestIsStockConfig(t *testing.T) {
208 + assert.True(t, isStockConfig("/usr/lib/netdata/conf.d/go.d/module.conf"))
209 + assert.False(t, isStockConfig("/etc/netdata/go.d/module.conf"))
210 +}
211 +
212 func TestAgent_buildDiscoveryConf(t *testing.T) {
213 + providers := []discovery.ProviderFactory{
214 + discovery.NewProviderFactory("noop", nil),
215 + }
216 + enabled := collectorapi.Registry{
217 + "module1": collectorapi.Creator{},
218 + }
219 +
220 + t.Run("inside k8s with no collectors dir disables provider list", func(t *testing.T) {
221 + a := &Agent{
222 + IsInsideK8s: true,
223 + DiscoveryProviders: providers,
224 + }
225 +
226 + cfg := a.buildDiscoveryConf(enabled, nil)
227 + assert.True(t, cfg.BuildContext.Policy.IsInsideK8s)
228 + assert.Empty(t, cfg.Providers)
229 + })
230 +
231 + t.Run("outside k8s keeps injected providers", func(t *testing.T) {
232 + a := &Agent{
233 + IsInsideK8s: false,
234 + DiscoveryProviders: providers,
235 + }
236
237 + cfg := a.buildDiscoveryConf(enabled, nil)
238 + assert.False(t, cfg.BuildContext.Policy.IsInsideK8s)
239 + assert.Len(t, cfg.Providers, 1)
240 + })
241 }
src/go/plugin/framework/vnodes/vnodes.go
+2 -9
@@ -11,12 +11,12 @@ import (
11 "maps"
12 "os"
13 "path/filepath"
14 - "strings"
14
15 "github.com/google/uuid"
16 "gopkg.in/yaml.v2"
17
18 "github.com/netdata/netdata/go/plugins/logger"
19 + "github.com/netdata/netdata/go/plugins/pkg/pluginconfig"
20 )
21
22 //go:embed "config_schema.json"
@@ -171,13 +171,6 @@ func loadConfigFile(conf any, path string) error {
171 return nil
172 }
173
174 -var (
175 - envNDStockConfigDir = os.Getenv("NETDATA_STOCK_CONFIG_DIR")
176 -)
177 -
174 func isStockConfig(path string) bool {
179 - if envNDStockConfigDir == "" {
180 - return false
181 - }
182 - return strings.HasPrefix(path, envNDStockConfigDir)
175 + return pluginconfig.IsStock(path)
176 }
src/go/plugin/framework/vnodes/vnodes_test.go
+5
@@ -12,3 +12,8 @@ func TestLoad(t *testing.T) {
12 assert.NotNil(t, Load("testdata"))
13 assert.NotNil(t, Load("not_exist"))
14 }
15 +
16 +func TestIsStockConfig(t *testing.T) {
17 + assert.True(t, isStockConfig("/usr/lib/netdata/conf.d/vnodes/test.conf"))
18 + assert.False(t, isStockConfig("/etc/netdata/vnodes/test.conf"))
19 +}
src/go/plugin/go.d/discovery/sdext/registry.go
+2 -3
@@ -6,7 +6,6 @@ import (
6 "encoding/json"
7 "fmt"
8
9 - "github.com/netdata/netdata/go/plugins/pkg/hostinfo"
9 "github.com/netdata/netdata/go/plugins/plugin/agent/discovery/sd"
10 "github.com/netdata/netdata/go/plugins/plugin/agent/discovery/sd/model"
11 "github.com/netdata/netdata/go/plugins/plugin/go.d/discovery/sdext/discoverer/dockersd"
@@ -22,7 +21,7 @@ const (
21 discovererSNMP = "snmp"
22 )
23
25 -func Registry() sd.Registry {
24 +func Registry(includeDocker bool) sd.Registry {
25 descs := []sd.Descriptor{
26 sd.NewDescriptor(
27 discovererNetListeners,
@@ -43,7 +42,7 @@ func Registry() sd.Registry {
42 newSNMPDiscoverers,
43 ),
44 }
46 - if !hostinfo.IsInsideK8sCluster() {
45 + if includeDocker {
46 descs = append(descs, sd.NewDescriptor(
47 discovererDocker,
48 schemaDocker,
src/go/plugin/go.d/discovery/sdext/registry_test.go new
+17
@@ -0,0 +1,17 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package sdext
4 +
5 +import (
6 + "testing"
7 +
8 + "github.com/stretchr/testify/assert"
9 +)
10 +
11 +func TestRegistry_DockerInclusion(t *testing.T) {
12 + withDocker := Registry(true)
13 + assert.Contains(t, withDocker.Types(), discovererDocker)
14 +
15 + withoutDocker := Registry(false)
16 + assert.NotContains(t, withoutDocker.Types(), discovererDocker)
17 +}