@cryptotaxi247 / netdata-1 / commits / 8a4a53626

refactor(go/plugin): move dump analyzer to metricsaudit (#21830)

Ilya Mashchenko committed Feb 26, 2026 at 18:39 UTC 8a4a53626d9d528547f16a05048051bdc4ea70ce
20 files changed +1026 -548
src/go/cmd/godplugin/main.go
+4 -2
@@ -59,7 +59,10 @@ func main() {
59 return
60 }
61
62 - pluginconfig.MustInit(opts)
62 + pluginconfig.MustInit(pluginconfig.InitInput{
63 + ConfDir: opts.ConfDir,
64 + WatchPath: opts.WatchPath,
65 + })
66
67 if opts.Function != "" {
68 os.Exit(runFunctionCLI(opts))
@@ -97,7 +100,6 @@ func main() {
100 RunModule: opts.Module,
101 RunJob: opts.Job,
102 MinUpdateEvery: opts.UpdateEvery,
100 - DumpSummary: opts.DumpSummary,
103 })
104
105 a.Infof("plugin: name=%s, %s", a.Name, buildinfo.Info())
src/go/cmd/ibmdplugin/main.go
+74 -33
@@ -11,9 +11,11 @@ import (
11 "os"
12 "os/user"
13 "path/filepath"
14 + "strconv"
15 "strings"
16 "time"
17
18 + "github.com/jessevdk/go-flags"
19 "github.com/netdata/netdata/go/plugins/cmd/internal/agenthost"
20 "github.com/netdata/netdata/go/plugins/cmd/internal/discoveryproviders"
21 "github.com/netdata/netdata/go/plugins/plugin/agent"
@@ -51,28 +53,27 @@ func init() {
53 func main() {
54 _, _ = maxprocs.Set(maxprocs.Logger(func(s string, args ...interface{}) {}))
55
54 - opts := parseCLI()
56 + opts, err := parseCLI()
57 + if err != nil {
58 + if cli.IsHelp(err) {
59 + os.Exit(0)
60 + }
61 + os.Exit(1)
62 + }
63
64 if opts.Version {
65 fmt.Printf("%s.plugin, version: %s\n", executable.Name, buildinfo.Version)
66 return
67 }
68
61 - if opts.DumpDataDir != "" && opts.DumpMode == "" {
62 - opts.DumpMode = "10m"
69 + if opts.MetricsAuditDataDir != "" && opts.MetricsAuditDuration == "" {
70 + opts.MetricsAuditDuration = "10m"
71 }
72
65 - pluginconfig.MustInit(opts)
66 -
67 - dumpDataDir := ""
68 - if opts.DumpDataDir != "" {
69 - var err error
70 - dumpDataDir, err = prepareDumpDataDir(opts.DumpDataDir, pluginconfig.VarLibDir())
71 - if err != nil {
72 - logger.Errorf("error preparing dump-data directory: %v", err)
73 - os.Exit(1)
74 - }
75 - }
73 + pluginconfig.MustInit(pluginconfig.InitInput{
74 + ConfDir: opts.ConfDir,
75 + WatchPath: opts.WatchPath,
76 + })
77
78 if lvl := pluginconfig.EnvLogLevel(); lvl != "" {
79 logger.Level.SetByName(lvl)
@@ -82,13 +83,32 @@ func main() {
83 }
84 isTerminal := terminal.IsTerminal()
85
85 - // Parse dump duration if provided
86 - var dumpMode time.Duration
87 - if opts.DumpMode != "" {
86 + // Parse metrics-audit duration if provided.
87 + var auditDuration time.Duration
88 + if opts.MetricsAuditDuration != "" {
89 var err error
89 - dumpMode, err = time.ParseDuration(opts.DumpMode)
90 + auditDuration, err = time.ParseDuration(opts.MetricsAuditDuration)
91 if err != nil {
91 - logger.Errorf("error: invalid dump duration '%s': %v", opts.DumpMode, err)
92 + logger.Errorf("error: invalid --metrics-audit duration '%s': %v", opts.MetricsAuditDuration, err)
93 + os.Exit(1)
94 + }
95 + if auditDuration <= 0 {
96 + logger.Errorf("error: invalid --metrics-audit duration '%s': duration must be > 0", opts.MetricsAuditDuration)
97 + os.Exit(1)
98 + }
99 + }
100 +
101 + if opts.MetricsAuditDataDir != "" && auditDuration <= 0 {
102 + logger.Errorf("error: --metrics-audit-data requires positive --metrics-audit duration")
103 + os.Exit(1)
104 + }
105 +
106 + metricsAuditDataDir := ""
107 + if opts.MetricsAuditDataDir != "" {
108 + var err error
109 + metricsAuditDataDir, err = prepareMetricsAuditDataDir(opts.MetricsAuditDataDir, pluginconfig.VarLibDir())
110 + if err != nil {
111 + logger.Errorf("error preparing --metrics-audit-data directory: %v", err)
112 os.Exit(1)
113 }
114 }
@@ -112,9 +132,9 @@ func main() {
132 RunModule: opts.Module,
133 RunJob: opts.Job,
134 MinUpdateEvery: opts.UpdateEvery,
115 - DumpMode: dumpMode,
116 - DumpSummary: opts.DumpSummary,
117 - DumpDataDir: dumpDataDir,
135 + AuditDuration: auditDuration,
136 + AuditSummary: opts.MetricsAuditSummary,
137 + AuditDataDir: metricsAuditDataDir,
138 DisableServiceDiscovery: true,
139 })
140
@@ -132,19 +152,39 @@ func main() {
152 agenthost.Run(a)
153 }
154
135 -func parseCLI() *cli.Option {
136 - opt, err := cli.Parse(os.Args)
155 +type options struct {
156 + cli.Option
157 + MetricsAuditDuration string `long:"metrics-audit" description:"run metrics-audit mode for specified duration (e.g. 30s, 5m) and analyze metric structure"`
158 + MetricsAuditSummary bool `long:"metrics-audit-summary" description:"show consolidated metrics-audit summary across all jobs"`
159 + MetricsAuditDataDir string `long:"metrics-audit-data" description:"write structured metrics-audit artifacts for the selected module to the given directory"`
160 +}
161 +
162 +func parseCLI() (*options, error) {
163 + opt := &options{
164 + Option: cli.Option{
165 + UpdateEvery: 1,
166 + },
167 + }
168 +
169 + parser := flags.NewParser(opt, flags.Default)
170 + parser.Name = executable.Name
171 + parser.Usage = "[OPTIONS] [update every]"
172 +
173 + rest, err := parser.ParseArgs(os.Args)
174 if err != nil {
138 - if cli.IsHelp(err) {
139 - os.Exit(0)
175 + return nil, err
176 + }
177 +
178 + if len(rest) > 1 {
179 + if opt.UpdateEvery, err = strconv.Atoi(rest[1]); err != nil {
180 + return nil, err
181 }
141 - os.Exit(1)
182 }
183
144 - return opt
184 + return opt, nil
185 }
186
147 -func prepareDumpDataDir(path string, varLibDir string) (string, error) {
187 +func prepareMetricsAuditDataDir(path string, varLibDir string) (string, error) {
188 if path == "" {
189 return "", nil
190 }
@@ -157,13 +197,14 @@ func prepareDumpDataDir(path string, varLibDir string) (string, error) {
197 return "", err
198 }
199 if absolute == "" || absolute == "/" {
160 - return "", fmt.Errorf("refusing to use unsafe dump-data directory '%s'", absolute)
200 + return "", fmt.Errorf("refusing to use unsafe metrics-audit-data directory '%s'", absolute)
201 }
162 - if err := os.RemoveAll(absolute); err != nil {
202 + if err := os.MkdirAll(absolute, 0o755); err != nil {
203 return "", err
204 }
165 - if err := os.MkdirAll(absolute, 0o755); err != nil {
205 + runDir := filepath.Join(absolute, fmt.Sprintf("run-%s", time.Now().UTC().Format("20060102-150405.000000000")))
206 + if err := os.MkdirAll(runDir, 0o755); err != nil {
207 return "", err
208 }
168 - return absolute, nil
209 + return runDir, nil
210 }
src/go/cmd/internal/agenthost/host.go
+27 -13
@@ -14,7 +14,7 @@ import (
14 "github.com/netdata/netdata/go/plugins/plugin/framework/collectorapi"
15 )
16
17 -// Run hosts an agent process lifecycle (signals, restart, quit, dump timer).
17 +// Run hosts an agent process lifecycle (signals, restart, quit, metrics-audit timer).
18 func Run(a *agent.Agent) {
19 ch := make(chan os.Signal, 1)
20 signal.Notify(ch, syscall.SIGHUP, syscall.SIGINT, syscall.SIGTERM)
@@ -36,13 +36,14 @@ func Run(a *agent.Agent) {
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()
39 + var finalizeReason string
40 +
41 + var auditTimer *time.Timer
42 + var auditTimerCh <-chan time.Time
43 + if mode := a.AuditDuration(); mode > 0 {
44 + auditTimer = time.NewTimer(mode)
45 + auditTimerCh = auditTimer.C
46 + defer auditTimer.Stop()
47 }
48
49 for {
@@ -65,20 +66,24 @@ func Run(a *agent.Agent) {
66 default:
67 a.Infof("received %s signal (%d). Terminating...", sig, sig)
68 exit = true
69 + finalizeReason = sig.String()
70 }
71 case <-a.QuitCh():
72 a.Infof("received QUIT command. Terminating...")
73 exit = true
72 - case <-dumpTimerCh:
73 - a.Infof("dump mode duration expired, collecting analysis...")
74 - a.TriggerDumpAnalysis()
74 + finalizeReason = "quit"
75 + case <-auditTimerCh:
76 + a.Infof("metrics-audit duration expired, finalizing metrics audit...")
77 exit = true
78 + finalizeReason = "audit timer expired"
79 case <-keepAliveErr:
80 a.Info("too many keepAlive errors. Terminating...")
81 exit = true
82 + finalizeReason = "keepalive error"
83 case <-runDone:
84 a.Info("agent run loop stopped. Terminating...")
85 exit = true
86 + finalizeReason = "run loop stopped"
87 }
88
89 if exit {
@@ -87,7 +92,7 @@ func Run(a *agent.Agent) {
92
93 cancel()
94
90 - func() {
95 + stopped := func() bool {
96 timeout := time.Second * 10
97 t := time.NewTimer(timeout)
98 defer t.Stop()
@@ -98,12 +103,21 @@ func Run(a *agent.Agent) {
103 select {
104 case <-t.C:
105 a.Errorf("stopping all goroutines timed out after %s. Exiting...", timeout)
101 - os.Exit(0)
106 + return false
107 case <-done:
108 + return true
109 }
110 }()
111
112 + if !stopped {
113 + if exit {
114 + a.FinalizeMetricsAudit(finalizeReason + ", forced shutdown")
115 + }
116 + os.Exit(0)
117 + }
118 +
119 if exit {
120 + a.FinalizeMetricsAudit(finalizeReason)
121 os.Exit(0)
122 }
123
src/go/cmd/scriptsdplugin/main.go
+4 -2
@@ -47,7 +47,10 @@ func main() {
47 return
48 }
49
50 - pluginconfig.MustInit(opts)
50 + pluginconfig.MustInit(pluginconfig.InitInput{
51 + ConfDir: opts.ConfDir,
52 + WatchPath: opts.WatchPath,
53 + })
54
55 watchPaths := pluginconfig.CollectorsConfigWatchPaths()
56 if len(watchPaths) == 0 {
@@ -85,7 +88,6 @@ func main() {
88 RunModule: opts.Module,
89 RunJob: opts.Job,
90 MinUpdateEvery: opts.UpdateEvery,
88 - DumpSummary: opts.DumpSummary,
91 DisableServiceDiscovery: true,
92 })
93
src/go/pkg/cli/cli.go
-3
@@ -19,9 +19,6 @@ type Option struct {
19 WatchPath []string `short:"w" long:"watch-path" description:"config path to watch"`
20 Debug bool `short:"d" long:"debug" description:"debug mode"`
21 Version bool `short:"v" long:"version" description:"display the version and exit"`
22 - DumpMode string `long:"dump" description:"run in dump mode for specified duration (e.g. 30s, 5m) and analyze metric structure"`
23 - DumpSummary bool `long:"dump-summary" description:"show consolidated summary across all jobs in dump mode"`
24 - DumpDataDir string `long:"dump-data" description:"write structured dump artifacts for the selected module to the given directory"`
22 Function string `long:"function" description:"execute function once (module name)"`
23 FunctionArgs []string `long:"function-args" description:"function args (repeatable, e.g. info)"`
24 FunctionPayload string `long:"function-payload" description:"function payload JSON or @file.json"`
src/go/pkg/pluginconfig/pluginconfig.go
+15 -10
@@ -16,7 +16,6 @@ import (
16 "sync"
17
18 "github.com/netdata/netdata/go/plugins/pkg/buildinfo"
19 - "github.com/netdata/netdata/go/plugins/pkg/cli"
19 "github.com/netdata/netdata/go/plugins/pkg/executable"
20 "github.com/netdata/netdata/go/plugins/pkg/multipath"
21 "github.com/netdata/netdata/go/plugins/pkg/terminal"
@@ -55,6 +54,12 @@ type directories struct {
54 varLibDir string
55 }
56
57 +// InitInput is the minimal CLI-derived input required to initialize paths.
58 +type InitInput struct {
59 + ConfDir []string
60 + WatchPath []string
61 +}
62 +
63 func IsStock(path string) bool {
64 stock := StockConfigDir()
65 if stock == "" {
@@ -66,11 +71,11 @@ func IsStock(path string) bool {
71
72 // MustInit parses env, applies CLI overrides, discovers directories, and stores them.
73 // Safe to call multiple times; only the first call has effect.
69 -func MustInit(opts *cli.Option) {
74 +func MustInit(input InitInput) {
75 initOnce.Do(func() {
76 env = readEnvFromOS(executable.Directory)
77 var d directories
73 - if err := d.build(opts, env, executable.Name, executable.Directory); err != nil {
78 + if err := d.build(input, env, executable.Name, executable.Directory); err != nil {
79 // fail fast during startup (internal invariant was broken)
80 panic(fmt.Errorf("pluginconfig initialization failed: %w", err))
81 }
@@ -116,22 +121,22 @@ func (d *directories) serviceDiscoveryDir() multipath.MultiPath {
121 return multipath.New(combined...)
122 }
123
119 -func (d *directories) build(opts *cli.Option, env envData, execName, execDir string) error {
120 - d.initUserRoots(opts, env, execDir)
124 +func (d *directories) build(input InitInput, env envData, execName, execDir string) error {
125 + d.initUserRoots(input, env, execDir)
126 d.initStockRoot(env, execDir)
127 d.deriveCollectors(execName)
128 d.deriveServiceDiscovery(execName)
124 - d.initWatchPaths(opts, env)
129 + d.initWatchPaths(input, env)
130 d.initVarLib(env)
131 return d.validate()
132 }
133
134 // Build step 1: initialize "user" roots as a multipath: CLI (highest), env, fallback.
130 -func (d *directories) initUserRoots(opts *cli.Option, env envData, execDir string) {
135 +func (d *directories) initUserRoots(input InitInput, env envData, execDir string) {
136 var roots multipath.MultiPath
137
138 // 1) CLI dirs
134 - for _, p := range opts.ConfDir {
139 + for _, p := range input.ConfDir {
140 p = safePathClean(handleDirOnWin(env.cygwinBase, p, execDir))
141 roots = append(roots, p)
142 }
@@ -228,8 +233,8 @@ func (d *directories) deriveServiceDiscovery(execName string) {
233 }
234
235 // Build step 5: init watchers (normalize + dedupe via multipath.New)
231 -func (d *directories) initWatchPaths(opts *cli.Option, env envData) {
232 - in := append([]string{}, opts.WatchPath...)
236 +func (d *directories) initWatchPaths(input InitInput, env envData) {
237 + in := append([]string{}, input.WatchPath...)
238 if env.watchPath != "" {
239 in = append(in, env.watchPath)
240 }
src/go/pkg/pluginconfig/pluginconfig_test.go
+13 -15
@@ -7,8 +7,6 @@ import (
7
8 "github.com/stretchr/testify/assert"
9 "github.com/stretchr/testify/require"
10 -
11 - "github.com/netdata/netdata/go/plugins/pkg/cli"
10 )
11
12 // helpers
@@ -34,14 +32,14 @@ func TestDirectoriesBuild(t *testing.T) {
32 mkdir(t, probeStock)
33
34 tests := map[string]struct {
37 - opts *cli.Option
35 + input InitInput
36 env envData
37 want directories
38 wantErr bool
39 }{
40 // ─────────────────────────────────── core (no cygwinBase) ───────────────────────────────────
41 "cli_dirs_only": {
44 - opts: &cli.Option{
42 + input: InitInput{
43 ConfDir: []string{"/tmp/user1", "/tmp/user2"},
44 },
45 env: envData{
@@ -59,7 +57,7 @@ func TestDirectoriesBuild(t *testing.T) {
57 },
58 },
59 "env_dirs_only": {
62 - opts: &cli.Option{},
60 + input: InitInput{},
61 env: envData{
62 userDir: "/tmp/env/user",
63 stockDir: "/tmp/env/stock",
@@ -76,7 +74,7 @@ func TestDirectoriesBuild(t *testing.T) {
74 },
75 },
76 "cli_overrides_env": {
79 - opts: &cli.Option{
77 + input: InitInput{
78 ConfDir: []string{"/tmp/cli/dir"},
79 },
80 env: envData{
@@ -95,8 +93,8 @@ func TestDirectoriesBuild(t *testing.T) {
93 },
94 },
95 "fallback_dirs_when_no_config": {
98 - opts: &cli.Option{},
99 - env: envData{},
96 + input: InitInput{},
97 + env: envData{},
98 // build-relative fallback from execDir
99 want: directories{
100 userConfigDirs: []string{filepath.Join(execDir, "..", "..", "..", "..", "etc", "netdata")},
@@ -110,7 +108,7 @@ func TestDirectoriesBuild(t *testing.T) {
108 },
109 },
110 "multiple_cli_dirs": {
113 - opts: &cli.Option{
111 + input: InitInput{
112 ConfDir: []string{"/tmp/dir1", "/tmp/dir2", "/tmp/dir3"},
113 },
114 env: envData{
@@ -138,7 +136,7 @@ func TestDirectoriesBuild(t *testing.T) {
136
137 // ─────────────────────────────── explicit cygwinBase cases ───────────────────────────────
138 "watch_paths_from_cli_and_env (cygwinBase)": {
141 - opts: &cli.Option{
139 + input: InitInput{
140 WatchPath: []string{"/tmp/watch1", "/tmp/watch2"},
141 },
142 env: envData{
@@ -157,7 +155,7 @@ func TestDirectoriesBuild(t *testing.T) {
155 },
156 },
157 "varlib_from_env (cygwinBase + probes)": {
160 - opts: &cli.Option{},
158 + input: InitInput{},
159 env: envData{
160 cygwinBase: tmp,
161 varLibDir: "/var/lib/netdata",
@@ -174,7 +172,7 @@ func TestDirectoriesBuild(t *testing.T) {
172 },
173 },
174 "empty_stock_dir_from_env (cygwinBase + probes)": {
177 - opts: &cli.Option{},
175 + input: InitInput{},
176 env: envData{
177 userDir: "/tmp/user", // explicit user
178 stockDir: "", // missing -> probe for stock
@@ -192,7 +190,7 @@ func TestDirectoriesBuild(t *testing.T) {
190 },
191 },
192 "env_user_pre_normalized (cygwinBase set)": {
195 - opts: &cli.Option{},
193 + input: InitInput{},
194 env: envData{
195 cygwinBase: tmp, // present but build() expects env already normalized
196 userDir: filepath.Join(tmp, "etc", "netdata"), // pre-normalized
@@ -210,7 +208,7 @@ func TestDirectoriesBuild(t *testing.T) {
208 },
209 },
210 "cli_dirs_remapped (cygwinBase)": {
213 - opts: &cli.Option{
211 + input: InitInput{
212 ConfDir: []string{"/etc/netdata", "/opt/netdata/etc/netdata"},
213 },
214 env: envData{
@@ -243,7 +241,7 @@ func TestDirectoriesBuild(t *testing.T) {
241 t.Run(name, func(t *testing.T) {
242 t.Parallel() // safe: build() uses only inputs, no globals
243 var got directories
246 - err := got.build(tc.opts, tc.env, testPluginName, execDir)
244 + err := got.build(tc.input, tc.env, testPluginName, execDir)
245
246 if tc.wantErr {
247 require.Error(t, err)
src/go/plugin/agent/agent.go
+50 -45
@@ -21,6 +21,7 @@ import (
21 "github.com/netdata/netdata/go/plugins/plugin/framework/collectorapi"
22 "github.com/netdata/netdata/go/plugins/plugin/framework/confgroup"
23 "github.com/netdata/netdata/go/plugins/plugin/framework/functions"
24 + "github.com/netdata/netdata/go/plugins/plugin/framework/metricsaudit"
25 )
26
27 // Config is an Agent configuration.
@@ -46,9 +47,9 @@ type Config struct {
47
48 DiscoveryProviders []discovery.ProviderFactory
49
49 - DumpMode time.Duration
50 - DumpSummary bool
51 - DumpDataDir string
50 + AuditDuration time.Duration
51 + AuditSummary bool
52 + AuditDataDir string
53 }
54
55 // Agent represents orchestrator.
@@ -83,14 +84,14 @@ type Agent struct {
84
85 quitCh chan struct{}
86
86 - // Dump mode
87 - dumpMode time.Duration
88 - dumpSummary bool
89 - dumpAnalyzer *DumpAnalyzer
90 - mgr *jobmgr.Manager
87 + // Metrics-audit mode.
88 + auditDuration time.Duration
89 + auditSummary bool
90 + auditAnalyzer *metricsaudit.Auditor
91
92 - dumpDataDir string
93 - dumpOnce sync.Once
92 + auditDataDir string
93 + quitOnce sync.Once
94 + auditOnce sync.Once
95 }
96
97 // New creates a new Agent.
@@ -115,26 +116,26 @@ func New(cfg Config) *Agent {
116 Out: safewriter.Stdout,
117 api: netdataapi.New(safewriter.Stdout),
118 quitCh: make(chan struct{}, 1),
118 - dumpMode: cfg.DumpMode,
119 - dumpSummary: cfg.DumpSummary,
119 + auditDuration: cfg.AuditDuration,
120 + auditSummary: cfg.AuditSummary,
121 DisableServiceDiscovery: cfg.DisableServiceDiscovery,
122 }
123
123 - if a.dumpMode > 0 {
124 - a.dumpAnalyzer = NewDumpAnalyzer()
125 - a.Infof("dump mode enabled: will run for %v and analyze metric structure", a.dumpMode)
126 - if a.dumpSummary {
127 - a.Infof("dump summary enabled: will show consolidated summary across all jobs")
124 + if a.auditDuration > 0 {
125 + a.auditAnalyzer = metricsaudit.New()
126 + a.Infof("metrics-audit mode enabled: will run for %v and analyze metric structure", a.auditDuration)
127 + if a.auditSummary {
128 + a.Infof("metrics-audit summary enabled: will show consolidated summary across all jobs")
129 }
130 }
131
131 - if cfg.DumpDataDir != "" {
132 - a.dumpDataDir = cfg.DumpDataDir
133 - if a.dumpAnalyzer == nil {
134 - a.dumpAnalyzer = NewDumpAnalyzer()
132 + if cfg.AuditDataDir != "" {
133 + a.auditDataDir = cfg.AuditDataDir
134 + if a.auditAnalyzer == nil {
135 + a.auditAnalyzer = metricsaudit.New()
136 }
136 - a.dumpAnalyzer.EnableDataCapture(cfg.DumpDataDir, a.signalDumpComplete)
137 - a.Infof("dump data directory: %s", cfg.DumpDataDir)
137 + a.auditAnalyzer.EnableDataCapture(cfg.AuditDataDir, a.signalAuditComplete)
138 + a.Infof("metrics-audit data directory: %s", cfg.AuditDataDir)
139 }
140
141 return a
@@ -173,19 +174,27 @@ func (a *Agent) RunKeepAlive(ctx context.Context) error {
174 }
175 }
176
176 -// QuitCh returns agent quit notifications (e.g., dump completion).
177 +// QuitCh returns agent quit notifications (e.g., metrics-audit completion).
178 func (a *Agent) QuitCh() <-chan struct{} {
179 return a.quitCh
180 }
181
181 -// DumpModeDuration returns configured dump mode duration.
182 -func (a *Agent) DumpModeDuration() time.Duration {
183 - return a.dumpMode
182 +// AuditDuration returns configured metrics-audit timer duration.
183 +func (a *Agent) AuditDuration() time.Duration {
184 + return a.auditDuration
185 }
186
186 -// TriggerDumpAnalysis prints dump analysis report.
187 -func (a *Agent) TriggerDumpAnalysis() {
188 - a.collectDumpAnalysis()
187 +// FinalizeMetricsAudit prints metrics-audit analysis report once.
188 +func (a *Agent) FinalizeMetricsAudit(reason string) {
189 + a.auditOnce.Do(func() {
190 + if a.auditAnalyzer == nil {
191 + return
192 + }
193 + if reason != "" {
194 + a.Infof("finalizing metrics audit (%s)", reason)
195 + }
196 + a.printMetricsAudit()
197 + })
198 }
199
200 func (a *Agent) run(ctx context.Context) {
@@ -237,15 +246,12 @@ func (a *Agent) run(ctx context.Context) {
246 VarLibDir: a.VarLibDir,
247 FnReg: fnMgr,
248 Vnodes: a.setupVnodeRegistry(),
240 - DumpMode: a.dumpMode > 0,
241 - DumpAnalyzer: a.dumpAnalyzer,
242 - DumpDataDir: a.dumpDataDir,
249 + AuditMode: a.auditDuration > 0,
250 + AuditAnalyzer: a.auditAnalyzer,
251 + AuditDataDir: a.auditDataDir,
252 RuntimeService: runtimeSvc,
253 })
254
246 - // Store reference for dump mode and enable dump mode if configured
247 - a.mgr = jobMgr
248 -
255 in := make(chan []*confgroup.Group)
256 var wg sync.WaitGroup
257
@@ -262,23 +268,22 @@ func (a *Agent) run(ctx context.Context) {
268 <-ctx.Done()
269 }
270
265 -func (a *Agent) collectDumpAnalysis() {
266 - if a.dumpAnalyzer == nil || a.mgr == nil {
267 - a.Error("dump analyzer or job manager not initialized")
271 +func (a *Agent) printMetricsAudit() {
272 + if a.auditAnalyzer == nil {
273 return
274 }
275
276 // Print the analysis report
272 - if a.dumpSummary {
273 - a.dumpAnalyzer.PrintSummary()
277 + if a.auditSummary {
278 + a.auditAnalyzer.PrintSummary()
279 } else {
275 - a.dumpAnalyzer.PrintReport()
280 + a.auditAnalyzer.PrintReport()
281 }
282 }
283
279 -func (a *Agent) signalDumpComplete() {
280 - a.dumpOnce.Do(func() {
281 - a.Infof("dump data collection complete, shutting down")
284 +func (a *Agent) signalAuditComplete() {
285 + a.quitOnce.Do(func() {
286 + a.Infof("metrics-audit data collection complete, shutting down")
287 select {
288 case a.quitCh <- struct{}{}:
289 default:
src/go/plugin/agent/dump.go deleted
-6
@@ -1,6 +0,0 @@
1 -// SPDX-License-Identifier: GPL-3.0-or-later
2 -
3 -package agent
4 -
5 -// Dump analyzer implementation is split across dump_model.go,
6 -// dump_capture.go, and dump_report.go.
src/go/plugin/agent/dump_capture.go deleted
-281
@@ -1,281 +0,0 @@
1 -// SPDX-License-Identifier: GPL-3.0-or-later
2 -
3 -package agent
4 -
5 -import (
6 - "encoding/json"
7 - "fmt"
8 - "os"
9 - "path/filepath"
10 - "sort"
11 - "time"
12 -
13 - "github.com/netdata/netdata/go/plugins/plugin/framework/collectorapi"
14 -)
15 -
16 -func (da *DumpAnalyzer) EnableDataCapture(dir string, onComplete func()) {
17 - da.mu.Lock()
18 - defer da.mu.Unlock()
19 - da.dataDir = dir
20 - da.onComplete = onComplete
21 -}
22 -
23 -// RegisterJob registers directory info for a job.
24 -func (da *DumpAnalyzer) RegisterJob(jobName, moduleName, dir string) {
25 - da.mu.Lock()
26 - defer da.mu.Unlock()
27 - if dir == "" {
28 - return
29 - }
30 - if da.jobDirs == nil {
31 - da.jobDirs = make(map[string]string)
32 - }
33 - da.jobDirs[jobName] = dir
34 - if da.jobDone == nil {
35 - da.jobDone = make(map[string]bool)
36 - }
37 - da.jobDone[jobName] = false
38 - // Ensure expected sub-directories exist
39 - _ = os.MkdirAll(filepath.Join(dir, "queries"), 0o755)
40 - _ = os.MkdirAll(filepath.Join(dir, "rows"), 0o755)
41 - _ = os.MkdirAll(filepath.Join(dir, "metrics"), 0o755)
42 - _ = os.MkdirAll(filepath.Join(dir, "meta"), 0o755)
43 -}
44 -
45 -// RecordJobStructure records the initial chart structure for a job
46 -func (da *DumpAnalyzer) RecordJobStructure(jobName, moduleName string, charts *collectorapi.Charts) {
47 - da.mu.Lock()
48 - defer da.mu.Unlock()
49 -
50 - job := &JobAnalysis{
51 - Name: jobName,
52 - Module: moduleName,
53 - Charts: make([]ChartAnalysis, 0),
54 - AllSeenMetrics: make(map[string]bool),
55 - }
56 -
57 - // Copy chart structure
58 - for _, chart := range *charts {
59 - ca := ChartAnalysis{
60 - Chart: chart,
61 - CollectedValues: make(map[string][]int64),
62 - SeenDimensions: make(map[string]bool),
63 - }
64 -
65 - // Initialize dimension tracking
66 - for _, dim := range chart.Dims {
67 - ca.CollectedValues[dim.ID] = make([]int64, 0)
68 - ca.SeenDimensions[dim.ID] = false
69 - }
70 -
71 - job.Charts = append(job.Charts, ca)
72 - }
73 -
74 - da.jobs[jobName] = job
75 - da.writeJobMetadata(jobName, moduleName)
76 -}
77 -
78 -// UpdateJobStructure updates the chart structure for a job with current charts
79 -// This is needed for collectors that create charts dynamically during collection
80 -func (da *DumpAnalyzer) UpdateJobStructure(jobName string, charts *collectorapi.Charts) {
81 - da.mu.Lock()
82 - defer da.mu.Unlock()
83 -
84 - job, exists := da.jobs[jobName]
85 - if !exists {
86 - return // Job not found, cannot update
87 - }
88 -
89 - // Create a map of existing chart data to preserve collected values
90 - existingCharts := make(map[string]*ChartAnalysis)
91 - for i := range job.Charts {
92 - existingCharts[job.Charts[i].Chart.ID] = &job.Charts[i]
93 - }
94 -
95 - // Rebuild chart list while preserving existing data
96 - job.Charts = make([]ChartAnalysis, 0)
97 -
98 - // Copy current chart structure
99 - for _, chart := range *charts {
100 - var ca ChartAnalysis
101 -
102 - // Check if we have existing data for this chart
103 - if existing, exists := existingCharts[chart.ID]; exists {
104 - // Preserve existing chart analysis but update the chart reference
105 - ca = *existing
106 - ca.Chart = chart
107 -
108 - // Add any new dimensions that weren't tracked before
109 - for _, dim := range chart.Dims {
110 - if _, tracked := ca.CollectedValues[dim.ID]; !tracked {
111 - ca.CollectedValues[dim.ID] = make([]int64, 0)
112 - ca.SeenDimensions[dim.ID] = false
113 - }
114 - }
115 - } else {
116 - // New chart - create fresh tracking
117 - ca = ChartAnalysis{
118 - Chart: chart,
119 - CollectedValues: make(map[string][]int64),
120 - SeenDimensions: make(map[string]bool),
121 - }
122 -
123 - // Initialize dimension tracking
124 - for _, dim := range chart.Dims {
125 - ca.CollectedValues[dim.ID] = make([]int64, 0)
126 - ca.SeenDimensions[dim.ID] = false
127 - }
128 - }
129 -
130 - job.Charts = append(job.Charts, ca)
131 - }
132 -}
133 -
134 -// RecordCollection records collected metrics directly from structured data
135 -func (da *DumpAnalyzer) RecordCollection(jobName string, mx map[string]int64) {
136 - da.mu.Lock()
137 - defer da.mu.Unlock()
138 -
139 - job, exists := da.jobs[jobName]
140 - if !exists {
141 - return
142 - }
143 -
144 - job.CollectionCount++
145 - job.LastCollection = time.Now()
146 -
147 - // Track ALL metrics in mx map
148 - for metricID := range mx {
149 - job.AllSeenMetrics[metricID] = true
150 - }
151 -
152 - // Record values for each chart
153 - for i := range job.Charts {
154 - ca := &job.Charts[i]
155 -
156 - // Check each dimension in this chart
157 - for _, dim := range ca.Chart.Dims {
158 - if value, collected := mx[dim.ID]; collected {
159 - ca.SeenDimensions[dim.ID] = true
160 - ca.CollectedValues[dim.ID] = append(ca.CollectedValues[dim.ID], value)
161 - }
162 - }
163 - }
164 -
165 - da.writeMetrics(jobName, job.CollectionCount, mx)
166 - da.markJobCollected(jobName)
167 -}
168 -
169 -func (da *DumpAnalyzer) writeJobMetadata(jobName, moduleName string) {
170 - if da.dataDir == "" {
171 - return
172 - }
173 - dir, ok := da.jobDirs[jobName]
174 - if !ok || dir == "" {
175 - return
176 - }
177 - meta := struct {
178 - Job string `json:"job"`
179 - Module string `json:"module"`
180 - Created time.Time `json:"created_at"`
181 - Metadata map[string]string `json:"metadata"`
182 - }{
183 - Job: jobName,
184 - Module: moduleName,
185 - Created: time.Now(),
186 - Metadata: map[string]string{
187 - "module": moduleName,
188 - },
189 - }
190 - path := filepath.Join(dir, "meta", "job.json")
191 - _ = writeJSON(path, meta)
192 -}
193 -
194 -func (da *DumpAnalyzer) writeMetrics(jobName string, seq int, mx map[string]int64) {
195 - if da.dataDir == "" {
196 - return
197 - }
198 - dir, ok := da.jobDirs[jobName]
199 - if !ok || dir == "" {
200 - return
201 - }
202 - metricsDir := filepath.Join(dir, "metrics")
203 - _ = os.MkdirAll(metricsDir, 0o755)
204 - payload := struct {
205 - CollectedAt time.Time `json:"collected_at"`
206 - Metrics map[string]int64 `json:"metrics"`
207 - }{
208 - CollectedAt: time.Now(),
209 - Metrics: mx,
210 - }
211 - filename := fmt.Sprintf("metrics-%04d.json", seq)
212 - path := filepath.Join(metricsDir, filename)
213 - _ = writeJSON(path, payload)
214 -}
215 -
216 -func (da *DumpAnalyzer) markJobCollected(jobName string) {
217 - if da.dataDir == "" {
218 - return
219 - }
220 - if da.jobDone == nil {
221 - return
222 - }
223 - da.jobDone[jobName] = true
224 - for job, dir := range da.jobDirs {
225 - if dir == "" {
226 - continue
227 - }
228 - if !da.jobDone[job] {
229 - return
230 - }
231 - }
232 - if da.completed {
233 - return
234 - }
235 - da.completed = true
236 - da.writeManifest()
237 - if da.onComplete != nil {
238 - go da.onComplete()
239 - }
240 -}
241 -
242 -func (da *DumpAnalyzer) writeManifest() {
243 - if da.dataDir == "" {
244 - return
245 - }
246 - type manifestJob struct {
247 - Name string `json:"name"`
248 - Module string `json:"module"`
249 - Directory string `json:"directory"`
250 - Collections int `json:"collections"`
251 - }
252 - var jobs []manifestJob
253 - for name, job := range da.jobs {
254 - dir := da.jobDirs[name]
255 - jobs = append(jobs, manifestJob{
256 - Name: name,
257 - Module: job.Module,
258 - Directory: dir,
259 - Collections: job.CollectionCount,
260 - })
261 - }
262 - sort.Slice(jobs, func(i, j int) bool { return jobs[i].Name < jobs[j].Name })
263 - manifest := struct {
264 - GeneratedAt time.Time `json:"generated_at"`
265 - Jobs []manifestJob `json:"jobs"`
266 - }{
267 - GeneratedAt: time.Now(),
268 - Jobs: jobs,
269 - }
270 - _ = writeJSON(filepath.Join(da.dataDir, "manifest.json"), manifest)
271 -}
272 -
273 -func writeJSON(path string, payload any) error {
274 - data, err := json.MarshalIndent(payload, "", " ")
275 - if err != nil {
276 - return err
277 - }
278 - return os.WriteFile(path, data, 0o644)
279 -}
280 -
281 -// contextInfo holds information about a context within a family
src/go/plugin/agent/dump_model.go deleted
-49
@@ -1,49 +0,0 @@
1 -// SPDX-License-Identifier: GPL-3.0-or-later
2 -
3 -package agent
4 -
5 -import (
6 - "sync"
7 - "time"
8 -
9 - "github.com/netdata/netdata/go/plugins/plugin/framework/collectorapi"
10 -)
11 -
12 -// DumpAnalyzer collects and analyzes metric structure from dump mode
13 -type DumpAnalyzer struct {
14 - mu sync.RWMutex
15 - jobs map[string]*JobAnalysis // key: job name
16 - startTime time.Time
17 - dataDir string
18 - jobDirs map[string]string
19 - jobDone map[string]bool
20 - onComplete func()
21 - completed bool
22 -}
23 -
24 -// JobAnalysis holds analysis for a single job
25 -type JobAnalysis struct {
26 - Name string
27 - Module string
28 - Charts []ChartAnalysis
29 - CollectionCount int
30 - LastCollection time.Time
31 - AllSeenMetrics map[string]bool // Track ALL metrics seen in mx map
32 -}
33 -
34 -// ChartAnalysis holds analysis for a single chart
35 -type ChartAnalysis struct {
36 - Chart *collectorapi.Chart
37 - CollectedValues map[string][]int64 // dimension ID -> collected values
38 - SeenDimensions map[string]bool // track which dimensions received data
39 -}
40 -
41 -// NewDumpAnalyzer creates a new dump analyzer
42 -func NewDumpAnalyzer() *DumpAnalyzer {
43 - return &DumpAnalyzer{
44 - jobs: make(map[string]*JobAnalysis),
45 - startTime: time.Now(),
46 - jobDirs: make(map[string]string),
47 - jobDone: make(map[string]bool),
48 - }
49 -}
src/go/plugin/agent/jobmgr/manager.go
+28 -29
@@ -24,6 +24,7 @@ import (
24 "github.com/netdata/netdata/go/plugins/plugin/framework/dyncfg"
25 "github.com/netdata/netdata/go/plugins/plugin/framework/functions"
26 "github.com/netdata/netdata/go/plugins/plugin/framework/jobruntime"
27 + "github.com/netdata/netdata/go/plugins/plugin/framework/metricsaudit"
28 "github.com/netdata/netdata/go/plugins/plugin/framework/runtimecomp"
29 "github.com/netdata/netdata/go/plugins/plugin/framework/vnodes"
30 "gopkg.in/yaml.v2"
@@ -39,9 +40,9 @@ type Config struct {
40 VarLibDir string
41 FnReg FunctionRegistry
42 Vnodes map[string]*vnodes.VirtualNode
42 - DumpMode bool
43 - DumpAnalyzer jobruntime.DumpAnalyzer
44 - DumpDataDir string
43 + AuditMode bool
44 + AuditAnalyzer metricsaudit.Analyzer
45 + AuditDataDir string
46 FunctionJSONWriter func(payload []byte, code int)
47 RuntimeService runtimecomp.Service
48 }
@@ -78,9 +79,9 @@ func New(cfg Config) *Manager {
79 fnReg: fnReg,
80 vnodes: vnodesReg,
81
81 - dumpMode: cfg.DumpMode,
82 - dumpAnalyzer: cfg.DumpAnalyzer,
83 - dumpDataDir: cfg.DumpDataDir,
82 + auditMode: cfg.AuditMode,
83 + auditAnalyzer: cfg.AuditAnalyzer,
84 + auditDataDir: cfg.AuditDataDir,
85 functionJSONWriter: cfg.FunctionJSONWriter,
86 runtimeService: cfg.RuntimeService,
87
@@ -145,10 +146,10 @@ type Manager struct {
146 fnReg FunctionRegistry
147 vnodes map[string]*vnodes.VirtualNode
148
148 - // Dump mode
149 - dumpMode bool
150 - dumpAnalyzer jobruntime.DumpAnalyzer
151 - dumpDataDir string
149 + // Metrics-audit mode.
150 + auditMode bool
151 + auditAnalyzer metricsaudit.Analyzer
152 + auditDataDir string
153
154 fileStatus *fileStatus
155 moduleFuncs *moduleFuncRegistry
@@ -558,18 +559,16 @@ func (m *Manager) createCollectorJob(cfg confgroup.Config) (runtimeJob, error) {
559
560 m.Debugf("creating %s[%s] job, config: %v", cfg.Module(), cfg.Name(), cfg)
561
561 - var jobDumpDir string
562 - if m.dumpDataDir != "" {
563 - jobDumpDir = filepath.Join(m.dumpDataDir, naming.Sanitize(cfg.Module()), naming.Sanitize(cfg.Name()))
564 - if err := os.MkdirAll(jobDumpDir, 0o755); err != nil {
565 - return nil, fmt.Errorf("creating dump directory: %w", err)
566 - }
567 - if m.dumpAnalyzer != nil {
568 - m.dumpAnalyzer.RegisterJob(cfg.Name(), cfg.Module(), jobDumpDir)
562 + useV2 := creator.CreateV2 != nil
563 +
564 + var jobCaptureDir string
565 + if m.auditDataDir != "" && !useV2 {
566 + jobCaptureDir = filepath.Join(m.auditDataDir, naming.Sanitize(cfg.Module()), naming.Sanitize(cfg.Name()))
567 + if err := os.MkdirAll(jobCaptureDir, 0o755); err != nil {
568 + return nil, fmt.Errorf("creating audit directory: %w", err)
569 }
570 }
571
572 - useV2 := creator.CreateV2 != nil
572 if useV2 {
573 mod := creator.CreateV2()
574 if mod == nil {
@@ -578,11 +577,6 @@ func (m *Manager) createCollectorJob(cfg confgroup.Config) (runtimeJob, error) {
577 if err := applyConfig(cfg, mod); err != nil {
578 return nil, err
579 }
581 - if jobDumpDir != "" {
582 - if dumpAware, ok := mod.(interface{ EnableDump(string) }); ok {
583 - dumpAware.EnableDump(jobDumpDir)
584 - }
585 - }
580
581 jobCfg := jobruntime.JobV2Config{
582 PluginName: m.pluginName,
@@ -614,9 +608,14 @@ func (m *Manager) createCollectorJob(cfg confgroup.Config) (runtimeJob, error) {
608 return nil, err
609 }
610
617 - if jobDumpDir != "" {
618 - if dumpAware, ok := mod.(interface{ EnableDump(string) }); ok {
619 - dumpAware.EnableDump(jobDumpDir)
611 + if m.auditAnalyzer != nil && jobCaptureDir != "" {
612 + // Auditing hooks are V1-only; V2 jobs are intentionally excluded.
613 + m.auditAnalyzer.RegisterJob(cfg.Name(), cfg.Module(), jobCaptureDir)
614 + }
615 +
616 + if jobCaptureDir != "" {
617 + if captureAware, ok := mod.(metricsaudit.Capturable); ok {
618 + captureAware.EnableCaptureArtifacts(jobCaptureDir)
619 }
620 }
621
@@ -632,8 +631,8 @@ func (m *Manager) createCollectorJob(cfg confgroup.Config) (runtimeJob, error) {
631 IsStock: cfg.SourceType() == "stock",
632 Module: mod,
633 Out: m.out,
635 - DumpMode: m.dumpMode,
636 - DumpAnalyzer: m.dumpAnalyzer,
634 + AuditMode: m.auditMode,
635 + AuditAnalyzer: m.auditAnalyzer,
636 FunctionOnly: functionOnly,
637 }
638
src/go/plugin/framework/jobruntime/dump_analyzer.go deleted
-14
@@ -1,14 +0,0 @@
1 -// SPDX-License-Identifier: GPL-3.0-or-later
2 -
3 -package jobruntime
4 -
5 -import "github.com/netdata/netdata/go/plugins/plugin/framework/collectorapi"
6 -
7 -// DumpAnalyzer captures dump-mode hooks used by job manager and runtime job.
8 -// Implementations can persist per-job artifacts and summarize metric structures.
9 -type DumpAnalyzer interface {
10 - RegisterJob(jobName, moduleName, dir string)
11 - RecordJobStructure(jobName, moduleName string, charts *collectorapi.Charts)
12 - UpdateJobStructure(jobName string, charts *collectorapi.Charts)
13 - RecordCollection(jobName string, mx map[string]int64)
14 -}
src/go/plugin/framework/jobruntime/job_v1.go
+19 -18
@@ -17,6 +17,7 @@ import (
17 "github.com/netdata/netdata/go/plugins/logger"
18 "github.com/netdata/netdata/go/plugins/pkg/netdataapi"
19 "github.com/netdata/netdata/go/plugins/plugin/framework/collectorapi"
20 + "github.com/netdata/netdata/go/plugins/plugin/framework/metricsaudit"
21 "github.com/netdata/netdata/go/plugins/plugin/framework/tickstate"
22 "github.com/netdata/netdata/go/plugins/plugin/framework/vnodes"
23 "github.com/netdata/netdata/go/plugins/plugin/go.d/pkg/oldmetrix"
@@ -66,8 +67,8 @@ type JobConfig struct {
67 Priority int
68 IsStock bool
69 Vnode vnodes.VirtualNode
69 - DumpMode bool
70 - DumpAnalyzer DumpAnalyzer
70 + AuditMode bool
71 + AuditAnalyzer metricsaudit.Analyzer
72 FunctionOnly bool
73 }
74
@@ -101,8 +102,8 @@ func NewJob(cfg JobConfig) *Job {
102 api: netdataapi.New(&buf),
103 vnode: cfg.Vnode,
104 updVnode: make(chan *vnodes.VirtualNode, 1),
104 - dumpMode: cfg.DumpMode,
105 - dumpAnalyzer: cfg.DumpAnalyzer,
105 + auditMode: cfg.AuditMode,
106 + auditAnalyzer: cfg.AuditAnalyzer,
107 }
108
109 log := logger.New().With(
@@ -161,10 +162,10 @@ type Job struct {
162
163 stopCtrl stopController
164
164 - // Dump mode support
165 - dumpMode bool
166 - dumpAnalyzer DumpAnalyzer
167 - skipTracker tickstate.SkipTracker
165 + // Metrics-audit mode support.
166 + auditMode bool
167 + auditAnalyzer metricsaudit.Analyzer
168 + skipTracker tickstate.SkipTracker
169 }
170
171 type collectedMetrics struct {
@@ -265,9 +266,9 @@ func (j *Job) AutoDetection() (err error) {
266 return err
267 }
268
268 - // Record job structure for dump mode after successful detection
269 - if j.dumpMode && j.dumpAnalyzer != nil && j.charts != nil {
270 - j.dumpAnalyzer.RecordJobStructure(j.name, j.moduleName, j.charts)
269 + // Record job structure for metrics-audit mode after successful detection.
270 + if j.auditMode && j.auditAnalyzer != nil && j.charts != nil {
271 + j.auditAnalyzer.RecordJobStructure(j.name, j.moduleName, j.charts)
272 }
273
274 return nil
@@ -477,10 +478,10 @@ func (j *Job) collect() collectedMetrics {
478 var mx collectedMetrics
479 mx.intMetrics = j.module.Collect(context.TODO())
480
480 - // Record collected metrics for dump mode
481 - // TODO: The dump analyzer only records intMetrics but ignores floatMetrics
482 - if j.dumpMode && j.dumpAnalyzer != nil && mx.intMetrics != nil {
483 - j.dumpAnalyzer.RecordCollection(j.name, mx.intMetrics)
481 + // Record collected metrics for metrics-audit mode.
482 + // TODO: The analyzer only records intMetrics but ignores floatMetrics.
483 + if j.auditMode && j.auditAnalyzer != nil && mx.intMetrics != nil {
484 + j.auditAnalyzer.RecordCollection(j.name, j.moduleName, mx.intMetrics)
485 }
486
487 return mx
@@ -557,9 +558,9 @@ func (j *Job) processMetrics(mx collectedMetrics, startTime time.Time, sinceLast
558 j.createChart(j.collectDurationChart)
559 }
560
560 - // Update dump analyzer with current chart structure for dynamic collectors
561 - if j.dumpMode && j.dumpAnalyzer != nil {
562 - j.dumpAnalyzer.UpdateJobStructure(j.name, j.charts)
561 + // Update analyzer with current chart structure for dynamic collectors.
562 + if j.auditMode && j.auditAnalyzer != nil {
563 + j.auditAnalyzer.UpdateJobStructure(j.name, j.moduleName, j.charts)
564 }
565
566 intMx := collectedMetrics{intMetrics: map[string]int64{"success": oldmetrix.Bool(updated > 0), "failed": oldmetrix.Bool(updated == 0)}}
src/go/plugin/framework/metricsaudit/analyzer.go new
+18
@@ -0,0 +1,18 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package metricsaudit
4 +
5 +import "github.com/netdata/netdata/go/plugins/plugin/framework/collectorapi"
6 +
7 +// Analyzer captures metrics-audit hooks used by job manager and runtime jobs.
8 +type Analyzer interface {
9 + RegisterJob(jobName, moduleName, dir string)
10 + RecordJobStructure(jobName, moduleName string, charts *collectorapi.Charts)
11 + UpdateJobStructure(jobName, moduleName string, charts *collectorapi.Charts)
12 + RecordCollection(jobName, moduleName string, mx map[string]int64)
13 +}
14 +
15 +// Capturable marks collectors that can emit additional capture artifacts.
16 +type Capturable interface {
17 + EnableCaptureArtifacts(string)
18 +}
src/go/plugin/framework/metricsaudit/capture.go new
+433
@@ -0,0 +1,433 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package metricsaudit
4 +
5 +import (
6 + "encoding/json"
7 + "fmt"
8 + "os"
9 + "path/filepath"
10 + "sort"
11 + "time"
12 +
13 + "github.com/netdata/netdata/go/plugins/plugin/framework/collectorapi"
14 +)
15 +
16 +type manifestJob struct {
17 + Name string `json:"name"`
18 + Module string `json:"module"`
19 + Directory string `json:"directory"`
20 + Collections int `json:"collections"`
21 +}
22 +
23 +type manifestPayload struct {
24 + GeneratedAt time.Time `json:"generated_at"`
25 + Jobs []manifestJob `json:"jobs"`
26 +}
27 +
28 +func (da *Auditor) EnableDataCapture(dir string, onComplete func()) {
29 + da.mu.Lock()
30 + defer da.mu.Unlock()
31 +
32 + da.dataDir = dir
33 + da.onComplete = onComplete
34 + if dir == "" || da.writeCh != nil {
35 + return
36 + }
37 +
38 + da.writeCh = make(chan writeTask, writeQueueSize)
39 + go da.runWriter(da.writeCh)
40 +}
41 +
42 +func (da *Auditor) runWriter(ch <-chan writeTask) {
43 + for task := range ch {
44 + if task.flush != nil {
45 + close(task.flush)
46 + continue
47 + }
48 +
49 + if task.run != nil {
50 + if err := task.run(); err != nil {
51 + da.recordWriteError(fmt.Errorf("%s: %w", task.label, err))
52 + }
53 + }
54 +
55 + if task.after != nil {
56 + task.after()
57 + }
58 + }
59 +}
60 +
61 +// RegisterJob registers directory info for a job.
62 +func (da *Auditor) RegisterJob(jobName, moduleName, dir string) {
63 + id := newJobID(moduleName, jobName)
64 +
65 + da.mu.Lock()
66 + if da.jobDirs == nil {
67 + da.jobDirs = make(map[JobID]string)
68 + }
69 + da.jobDirs[id] = dir
70 + if da.jobDone == nil {
71 + da.jobDone = make(map[JobID]bool)
72 + }
73 + da.jobDone[id] = false
74 + da.mu.Unlock()
75 +
76 + if dir == "" {
77 + return
78 + }
79 +
80 + for _, sub := range []string{"queries", "rows", "metrics", "meta"} {
81 + if err := os.MkdirAll(filepath.Join(dir, sub), 0o755); err != nil {
82 + da.recordWriteError(fmt.Errorf("prepare %s directory for %s[%s]: %w", sub, moduleName, jobName, err))
83 + }
84 + }
85 +}
86 +
87 +// RecordJobStructure records the initial chart structure for a job.
88 +func (da *Auditor) RecordJobStructure(jobName, moduleName string, charts *collectorapi.Charts) {
89 + if charts == nil {
90 + return
91 + }
92 +
93 + job := &JobAnalysis{
94 + Name: jobName,
95 + Module: moduleName,
96 + Charts: make([]ChartAnalysis, 0, len(*charts)),
97 + AllSeenMetrics: make(map[string]bool),
98 + }
99 +
100 + for _, chart := range *charts {
101 + ca := ChartAnalysis{
102 + Chart: chart,
103 + CollectedValues: make(map[string][]int64),
104 + SeenDimensions: make(map[string]bool),
105 + }
106 + for _, dim := range chart.Dims {
107 + ca.CollectedValues[dim.ID] = make([]int64, 0)
108 + ca.SeenDimensions[dim.ID] = false
109 + }
110 + job.Charts = append(job.Charts, ca)
111 + }
112 +
113 + id := newJobID(moduleName, jobName)
114 + var dir string
115 + var captureEnabled bool
116 +
117 + da.mu.Lock()
118 + da.jobs[id] = job
119 + dir = da.jobDirs[id]
120 + captureEnabled = da.dataDir != "" && dir != ""
121 + da.mu.Unlock()
122 +
123 + if !captureEnabled {
124 + return
125 + }
126 +
127 + meta := struct {
128 + Job string `json:"job"`
129 + Module string `json:"module"`
130 + Created time.Time `json:"created_at"`
131 + Metadata map[string]string `json:"metadata"`
132 + }{
133 + Job: jobName,
134 + Module: moduleName,
135 + Created: time.Now(),
136 + Metadata: map[string]string{
137 + "module": moduleName,
138 + },
139 + }
140 + path := filepath.Join(dir, "meta", "job.json")
141 + da.enqueueJSONWrite(
142 + fmt.Sprintf("write metadata for %s[%s]", moduleName, jobName),
143 + path,
144 + meta,
145 + nil,
146 + )
147 +}
148 +
149 +// UpdateJobStructure updates the chart structure for a job with current charts.
150 +// This is needed for collectors that create charts dynamically during collection.
151 +func (da *Auditor) UpdateJobStructure(jobName, moduleName string, charts *collectorapi.Charts) {
152 + if charts == nil {
153 + return
154 + }
155 +
156 + id := newJobID(moduleName, jobName)
157 +
158 + da.mu.Lock()
159 + defer da.mu.Unlock()
160 +
161 + job, exists := da.jobs[id]
162 + if !exists {
163 + return
164 + }
165 +
166 + existingCharts := make(map[string]*ChartAnalysis)
167 + for i := range job.Charts {
168 + existingCharts[job.Charts[i].Chart.ID] = &job.Charts[i]
169 + }
170 +
171 + job.Charts = make([]ChartAnalysis, 0, len(*charts))
172 + for _, chart := range *charts {
173 + var ca ChartAnalysis
174 + if existing, ok := existingCharts[chart.ID]; ok {
175 + ca = *existing
176 + ca.Chart = chart
177 + for _, dim := range chart.Dims {
178 + if _, tracked := ca.CollectedValues[dim.ID]; !tracked {
179 + ca.CollectedValues[dim.ID] = make([]int64, 0)
180 + ca.SeenDimensions[dim.ID] = false
181 + }
182 + }
183 + } else {
184 + ca = ChartAnalysis{
185 + Chart: chart,
186 + CollectedValues: make(map[string][]int64),
187 + SeenDimensions: make(map[string]bool),
188 + }
189 + for _, dim := range chart.Dims {
190 + ca.CollectedValues[dim.ID] = make([]int64, 0)
191 + ca.SeenDimensions[dim.ID] = false
192 + }
193 + }
194 + job.Charts = append(job.Charts, ca)
195 + }
196 +}
197 +
198 +// RecordCollection records collected metrics directly from structured data.
199 +func (da *Auditor) RecordCollection(jobName, moduleName string, mx map[string]int64) {
200 + if mx == nil {
201 + return
202 + }
203 +
204 + id := newJobID(moduleName, jobName)
205 +
206 + var seq int
207 + var metricsDir string
208 + var metricsPath string
209 + var captureEnabled bool
210 + var manifest *manifestPayload
211 + var onComplete func()
212 +
213 + da.mu.Lock()
214 + job, exists := da.jobs[id]
215 + if !exists {
216 + da.mu.Unlock()
217 + return
218 + }
219 +
220 + job.CollectionCount++
221 + job.LastCollection = time.Now()
222 + seq = job.CollectionCount
223 +
224 + for metricID := range mx {
225 + job.AllSeenMetrics[metricID] = true
226 + }
227 +
228 + for i := range job.Charts {
229 + ca := &job.Charts[i]
230 + for _, dim := range ca.Chart.Dims {
231 + if value, collected := mx[dim.ID]; collected {
232 + ca.SeenDimensions[dim.ID] = true
233 + ca.CollectedValues[dim.ID] = append(ca.CollectedValues[dim.ID], value)
234 + }
235 + }
236 + }
237 +
238 + if dir := da.jobDirs[id]; da.dataDir != "" && dir != "" {
239 + captureEnabled = true
240 + metricsDir = filepath.Join(dir, "metrics")
241 + metricsPath = filepath.Join(metricsDir, fmt.Sprintf("metrics-%04d.json", seq))
242 + }
243 +
244 + manifest, onComplete = da.markJobCollectedLocked(id)
245 + da.mu.Unlock()
246 +
247 + if captureEnabled {
248 + payload := struct {
249 + CollectedAt time.Time `json:"collected_at"`
250 + Metrics map[string]int64 `json:"metrics"`
251 + }{
252 + CollectedAt: time.Now(),
253 + Metrics: cloneIntMetrics(mx),
254 + }
255 +
256 + da.enqueueWriteTask(writeTask{
257 + label: fmt.Sprintf("write metrics snapshot for %s[%s]", moduleName, jobName),
258 + run: func() error {
259 + if err := os.MkdirAll(metricsDir, 0o755); err != nil {
260 + return err
261 + }
262 + return writeJSON(metricsPath, payload)
263 + },
264 + })
265 + }
266 +
267 + da.handleCompletionWrite(manifest, onComplete)
268 +}
269 +
270 +func (da *Auditor) handleCompletionWrite(manifest *manifestPayload, onComplete func()) {
271 + if onComplete == nil {
272 + return
273 + }
274 + if manifest == nil {
275 + go onComplete()
276 + return
277 + }
278 +
279 + da.mu.RLock()
280 + manifestPath := filepath.Join(da.dataDir, "manifest.json")
281 + da.mu.RUnlock()
282 +
283 + enqueued := da.enqueueJSONWrite("write audit manifest", manifestPath, manifest, func() {
284 + go onComplete()
285 + })
286 + if !enqueued {
287 + go onComplete()
288 + }
289 +}
290 +
291 +func (da *Auditor) markJobCollectedLocked(id JobID) (*manifestPayload, func()) {
292 + if da.jobDone == nil {
293 + return nil, nil
294 + }
295 + if _, tracked := da.jobDone[id]; !tracked {
296 + return nil, nil
297 + }
298 +
299 + da.jobDone[id] = true
300 + for jobID, dir := range da.jobDirs {
301 + if dir == "" {
302 + continue
303 + }
304 + if !da.jobDone[jobID] {
305 + return nil, nil
306 + }
307 + }
308 +
309 + if da.completed {
310 + return nil, nil
311 + }
312 + da.completed = true
313 + if da.dataDir == "" {
314 + return nil, da.onComplete
315 + }
316 +
317 + manifest := da.buildManifestLocked()
318 + return &manifest, da.onComplete
319 +}
320 +
321 +func (da *Auditor) buildManifestLocked() manifestPayload {
322 + jobs := make([]manifestJob, 0, len(da.jobs))
323 + for id, job := range da.jobs {
324 + dir := da.jobDirs[id]
325 + jobs = append(jobs, manifestJob{
326 + Name: id.Name,
327 + Module: id.Module,
328 + Directory: dir,
329 + Collections: job.CollectionCount,
330 + })
331 + }
332 +
333 + sort.Slice(jobs, func(i, j int) bool {
334 + if jobs[i].Module == jobs[j].Module {
335 + return jobs[i].Name < jobs[j].Name
336 + }
337 + return jobs[i].Module < jobs[j].Module
338 + })
339 +
340 + return manifestPayload{
341 + GeneratedAt: time.Now(),
342 + Jobs: jobs,
343 + }
344 +}
345 +
346 +func (da *Auditor) enqueueJSONWrite(label, path string, payload any, after func()) bool {
347 + return da.enqueueWriteTask(writeTask{
348 + label: label,
349 + run: func() error {
350 + return writeJSON(path, payload)
351 + },
352 + after: after,
353 + })
354 +}
355 +
356 +func (da *Auditor) enqueueWriteTask(task writeTask) bool {
357 + da.mu.RLock()
358 + ch := da.writeCh
359 + da.mu.RUnlock()
360 + if ch == nil {
361 + return false
362 + }
363 +
364 + select {
365 + case ch <- task:
366 + return true
367 + default:
368 + da.recordWriteError(fmt.Errorf("%s: write queue is full", task.label))
369 + return false
370 + }
371 +}
372 +
373 +func (da *Auditor) flushWriteQueue(timeout time.Duration) bool {
374 + da.mu.RLock()
375 + ch := da.writeCh
376 + da.mu.RUnlock()
377 + if ch == nil {
378 + return true
379 + }
380 +
381 + ack := make(chan struct{})
382 + task := writeTask{flush: ack}
383 + timer := time.NewTimer(timeout)
384 + defer timer.Stop()
385 +
386 + select {
387 + case ch <- task:
388 + case <-timer.C:
389 + da.recordWriteError(fmt.Errorf("flush write queue: timed out while enqueueing sentinel"))
390 + return false
391 + }
392 +
393 + timer.Reset(timeout)
394 + select {
395 + case <-ack:
396 + return true
397 + case <-timer.C:
398 + da.recordWriteError(fmt.Errorf("flush write queue: timed out waiting for sentinel"))
399 + return false
400 + }
401 +}
402 +
403 +func (da *Auditor) recordWriteError(err error) {
404 + if err == nil {
405 + return
406 + }
407 +
408 + da.mu.Lock()
409 + defer da.mu.Unlock()
410 + da.writeErrorCount++
411 + if len(da.writeErrors) < maxWriteErrorSamples {
412 + da.writeErrors = append(da.writeErrors, err.Error())
413 + }
414 +}
415 +
416 +func cloneIntMetrics(mx map[string]int64) map[string]int64 {
417 + if len(mx) == 0 {
418 + return map[string]int64{}
419 + }
420 + out := make(map[string]int64, len(mx))
421 + for k, v := range mx {
422 + out[k] = v
423 + }
424 + return out
425 +}
426 +
427 +func writeJSON(path string, payload any) error {
428 + data, err := json.MarshalIndent(payload, "", " ")
429 + if err != nil {
430 + return err
431 + }
432 + return os.WriteFile(path, data, 0o644)
433 +}
src/go/plugin/framework/metricsaudit/capture_test.go new
+115
@@ -0,0 +1,115 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package metricsaudit
4 +
5 +import (
6 + "encoding/json"
7 + "os"
8 + "path/filepath"
9 + "testing"
10 + "time"
11 +
12 + "github.com/netdata/netdata/go/plugins/plugin/framework/collectorapi"
13 +)
14 +
15 +func TestAuditorSeparatesSameJobNameAcrossModules(t *testing.T) {
16 + da := New()
17 +
18 + da.RegisterJob("shared", "modA", "")
19 + da.RegisterJob("shared", "modB", "")
20 +
21 + chartsA := testCharts("chart_a", "ctx_a", "dim_a")
22 + chartsB := testCharts("chart_b", "ctx_b", "dim_b")
23 + da.RecordJobStructure("shared", "modA", &chartsA)
24 + da.RecordJobStructure("shared", "modB", &chartsB)
25 +
26 + da.RecordCollection("shared", "modA", map[string]int64{"dim_a": 1})
27 + da.RecordCollection("shared", "modB", map[string]int64{"dim_b": 2})
28 +
29 + idA := newJobID("modA", "shared")
30 + idB := newJobID("modB", "shared")
31 +
32 + if got := len(da.jobs); got != 2 {
33 + t.Fatalf("expected 2 jobs, got %d", got)
34 + }
35 + if da.jobs[idA] == nil || da.jobs[idB] == nil {
36 + t.Fatalf("expected both module-scoped jobs to exist: %+v", da.jobs)
37 + }
38 + if da.jobs[idA].CollectionCount != 1 || da.jobs[idB].CollectionCount != 1 {
39 + t.Fatalf("expected per-module collection counts to be isolated, got %d and %d",
40 + da.jobs[idA].CollectionCount, da.jobs[idB].CollectionCount)
41 + }
42 + if !da.jobs[idA].AllSeenMetrics["dim_a"] {
43 + t.Fatalf("expected modA metrics to include dim_a")
44 + }
45 + if !da.jobs[idB].AllSeenMetrics["dim_b"] {
46 + t.Fatalf("expected modB metrics to include dim_b")
47 + }
48 +}
49 +
50 +func TestAuditorOnCompleteAfterAllRegisteredJobsCollected(t *testing.T) {
51 + da := New()
52 + baseDir := t.TempDir()
53 +
54 + done := make(chan struct{}, 1)
55 + da.EnableDataCapture(baseDir, func() {
56 + select {
57 + case done <- struct{}{}:
58 + default:
59 + }
60 + })
61 +
62 + job1Dir := filepath.Join(baseDir, "mod", "job1")
63 + job2Dir := filepath.Join(baseDir, "mod", "job2")
64 + da.RegisterJob("job1", "mod", job1Dir)
65 + da.RegisterJob("job2", "mod", job2Dir)
66 +
67 + charts1 := testCharts("chart1", "ctx1", "dim1")
68 + charts2 := testCharts("chart2", "ctx2", "dim2")
69 + da.RecordJobStructure("job1", "mod", &charts1)
70 + da.RecordJobStructure("job2", "mod", &charts2)
71 +
72 + da.RecordCollection("job1", "mod", map[string]int64{"dim1": 10})
73 + select {
74 + case <-done:
75 + t.Fatalf("onComplete fired before all registered jobs collected")
76 + default:
77 + }
78 +
79 + da.RecordCollection("job2", "mod", map[string]int64{"dim2": 20})
80 + select {
81 + case <-done:
82 + case <-time.After(2 * time.Second):
83 + t.Fatalf("timed out waiting for onComplete callback")
84 + }
85 +
86 + if !da.flushWriteQueue(2 * time.Second) {
87 + t.Fatalf("failed to flush write queue")
88 + }
89 +
90 + manifestPath := filepath.Join(baseDir, "manifest.json")
91 + data, err := os.ReadFile(manifestPath)
92 + if err != nil {
93 + t.Fatalf("read manifest: %v", err)
94 + }
95 +
96 + var manifest manifestPayload
97 + if err := json.Unmarshal(data, &manifest); err != nil {
98 + t.Fatalf("decode manifest: %v", err)
99 + }
100 + if got := len(manifest.Jobs); got != 2 {
101 + t.Fatalf("expected 2 jobs in manifest, got %d", got)
102 + }
103 +}
104 +
105 +func testCharts(chartID, context, dimID string) collectorapi.Charts {
106 + return collectorapi.Charts{
107 + {
108 + ID: chartID,
109 + Ctx: context,
110 + Dims: collectorapi.Dims{
111 + {ID: dimID},
112 + },
113 + },
114 + }
115 +}
src/go/plugin/framework/metricsaudit/model.go new
+77
@@ -0,0 +1,77 @@
1 +// SPDX-License-Identifier: GPL-3.0-or-later
2 +
3 +package metricsaudit
4 +
5 +import (
6 + "sync"
7 + "time"
8 +
9 + "github.com/netdata/netdata/go/plugins/plugin/framework/collectorapi"
10 +)
11 +
12 +const (
13 + writeQueueSize = 256
14 + maxWriteErrorSamples = 10
15 +)
16 +
17 +// JobID identifies a job uniquely across modules.
18 +type JobID struct {
19 + Module string
20 + Name string
21 +}
22 +
23 +func newJobID(moduleName, jobName string) JobID {
24 + return JobID{
25 + Module: moduleName,
26 + Name: jobName,
27 + }
28 +}
29 +
30 +type writeTask struct {
31 + label string
32 + run func() error
33 + after func()
34 + flush chan struct{}
35 +}
36 +
37 +// Auditor collects and analyzes metric structure from metrics-audit mode.
38 +type Auditor struct {
39 + mu sync.RWMutex
40 + jobs map[JobID]*JobAnalysis
41 + startTime time.Time
42 + dataDir string
43 + jobDirs map[JobID]string
44 + jobDone map[JobID]bool
45 + onComplete func()
46 + completed bool
47 + writeCh chan writeTask
48 + writeErrorCount int
49 + writeErrors []string
50 +}
51 +
52 +// JobAnalysis holds analysis for a single job
53 +type JobAnalysis struct {
54 + Name string
55 + Module string
56 + Charts []ChartAnalysis
57 + CollectionCount int
58 + LastCollection time.Time
59 + AllSeenMetrics map[string]bool // Track ALL metrics seen in mx map
60 +}
61 +
62 +// ChartAnalysis holds analysis for a single chart
63 +type ChartAnalysis struct {
64 + Chart *collectorapi.Chart
65 + CollectedValues map[string][]int64 // dimension ID -> collected values
66 + SeenDimensions map[string]bool // track which dimensions received data
67 +}
68 +
69 +// New creates a new metrics-audit analyzer.
70 +func New() *Auditor {
71 + return &Auditor{
72 + jobs: make(map[JobID]*JobAnalysis),
73 + startTime: time.Now(),
74 + jobDirs: make(map[JobID]string),
75 + jobDone: make(map[JobID]bool),
76 + }
77 +}
src/go/plugin/framework/metricsaudit/report.go renamed
+147 -26
@@ -1,40 +1,60 @@
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 -package agent
3 +package metricsaudit
4
5 import (
6 "fmt"
7 "sort"
8 "strings"
9 + "time"
10
11 "github.com/netdata/netdata/go/plugins/plugin/framework/collectorapi"
12 )
13
13 -func (da *DumpAnalyzer) PrintReport() {
14 - da.mu.RLock()
15 - defer da.mu.RUnlock()
14 +func (da *Auditor) PrintReport() {
15 + _ = da.flushWriteQueue(2 * time.Second)
16
17 - // Sort jobs for consistent output
18 - var jobNames []string
19 - for name := range da.jobs {
20 - jobNames = append(jobNames, name)
17 + type reportJob struct {
18 + id JobID
19 + job JobAnalysis
20 }
22 - sort.Strings(jobNames)
21
24 - for _, jobName := range jobNames {
25 - job := da.jobs[jobName]
26 - da.printJobAnalysis(job)
22 + da.mu.RLock()
23 + jobs := make([]reportJob, 0, len(da.jobs))
24 + for id, job := range da.jobs {
25 + jobs = append(jobs, reportJob{id: id, job: cloneJobAnalysis(job)})
26 }
27 + writeErrorCount := da.writeErrorCount
28 + writeErrors := append([]string(nil), da.writeErrors...)
29 + da.mu.RUnlock()
30 +
31 + sort.Slice(jobs, func(i, j int) bool {
32 + if jobs[i].id.Module == jobs[j].id.Module {
33 + return jobs[i].id.Name < jobs[j].id.Name
34 + }
35 + return jobs[i].id.Module < jobs[j].id.Module
36 + })
37 +
38 + for _, entry := range jobs {
39 + job := entry.job
40 + da.printJobAnalysis(&job)
41 + }
42 +
43 + da.printWriteErrorSummary(writeErrorCount, writeErrors)
44 }
45
46 // PrintSummary prints a consolidated summary across all jobs
31 -func (da *DumpAnalyzer) PrintSummary() {
32 - da.mu.RLock()
33 - defer da.mu.RUnlock()
34 -
47 +func (da *Auditor) PrintSummary() {
48 // First print the regular report
49 da.PrintReport()
50
51 + da.mu.RLock()
52 + jobs := make([]JobAnalysis, 0, len(da.jobs))
53 + for _, job := range da.jobs {
54 + jobs = append(jobs, cloneJobAnalysis(job))
55 + }
56 + da.mu.RUnlock()
57 +
58 // Then print the consolidated summary
59 fmt.Println("\n" + strings.Repeat("═", 80))
60 fmt.Println("CONSOLIDATED SUMMARY ACROSS ALL JOBS")
@@ -56,7 +76,9 @@ func (da *DumpAnalyzer) PrintSummary() {
76
77 contextMap := make(map[string]*contextSummary) // context -> summary
78
59 - for jobName, job := range da.jobs {
79 + for i := range jobs {
80 + job := &jobs[i]
81 + jobLabel := fmt.Sprintf("%s[%s]", job.Module, job.Name)
82 for i := range job.Charts {
83 ca := &job.Charts[i]
84
@@ -104,7 +126,7 @@ func (da *DumpAnalyzer) PrintSummary() {
126
127 // Update instance count and job tracking
128 contextMap[ctx].instances++
107 - contextMap[ctx].jobs[jobName] = true
129 + contextMap[ctx].jobs[jobLabel] = true
130
131 // Update label keys and dimension names if needed
132 for _, label := range ca.Chart.Labels {
@@ -222,6 +244,23 @@ func (da *DumpAnalyzer) PrintSummary() {
244 }
245 }
246
247 +func (da *Auditor) printWriteErrorSummary(count int, samples []string) {
248 + if count == 0 {
249 + return
250 + }
251 +
252 + fmt.Println("\n" + strings.Repeat("═", 80))
253 + fmt.Printf("CAPTURE WRITE ERRORS: %d (showing up to %d)\n", count, len(samples))
254 + fmt.Println(strings.Repeat("═", 80))
255 +
256 + for _, msg := range samples {
257 + fmt.Printf("⚠️ %s\n", msg)
258 + }
259 + if remaining := count - len(samples); remaining > 0 {
260 + fmt.Printf("... %d additional write errors omitted\n", remaining)
261 + }
262 +}
263 +
264 type contextInfo struct {
265 family string
266 context string
@@ -229,7 +268,7 @@ type contextInfo struct {
268 minPriority int
269 }
270
232 -func (da *DumpAnalyzer) printJobAnalysis(job *JobAnalysis) {
271 +func (da *Auditor) printJobAnalysis(job *JobAnalysis) {
272 // First, check for duplicate chart IDs (SEVERE BUG)
273 chartIDCounts := make(map[string]int)
274 for i := range job.Charts {
@@ -293,7 +332,7 @@ func (da *DumpAnalyzer) printJobAnalysis(job *JobAnalysis) {
332 }
333
334 // Check for severe bugs - duplicate chart IDs and contexts in multiple families
296 - fmt.Println("\n" + job.Name)
335 + fmt.Printf("\n%s[%s]\n", job.Module, job.Name)
336
337 // Report duplicate chart IDs first (most severe)
338 for chartID, count := range chartIDCounts {
@@ -484,7 +523,72 @@ func (da *DumpAnalyzer) printJobAnalysis(job *JobAnalysis) {
523 }
524 }
525
487 -func (da *DumpAnalyzer) printContextAnalysis(ctxInfo *contextInfo, isLast bool) []string {
526 +func cloneJobAnalysis(src *JobAnalysis) JobAnalysis {
527 + if src == nil {
528 + return JobAnalysis{}
529 + }
530 +
531 + dst := JobAnalysis{
532 + Name: src.Name,
533 + Module: src.Module,
534 + CollectionCount: src.CollectionCount,
535 + LastCollection: src.LastCollection,
536 + AllSeenMetrics: make(map[string]bool, len(src.AllSeenMetrics)),
537 + Charts: make([]ChartAnalysis, len(src.Charts)),
538 + }
539 + for metricID, seen := range src.AllSeenMetrics {
540 + dst.AllSeenMetrics[metricID] = seen
541 + }
542 + for i := range src.Charts {
543 + dst.Charts[i] = cloneChartAnalysis(src.Charts[i])
544 + }
545 +
546 + return dst
547 +}
548 +
549 +func cloneChartAnalysis(src ChartAnalysis) ChartAnalysis {
550 + dst := ChartAnalysis{
551 + Chart: cloneChart(src.Chart),
552 + CollectedValues: make(map[string][]int64, len(src.CollectedValues)),
553 + SeenDimensions: make(map[string]bool, len(src.SeenDimensions)),
554 + }
555 + for id, values := range src.CollectedValues {
556 + dst.CollectedValues[id] = append([]int64(nil), values...)
557 + }
558 + for id, seen := range src.SeenDimensions {
559 + dst.SeenDimensions[id] = seen
560 + }
561 + return dst
562 +}
563 +
564 +func cloneChart(src *collectorapi.Chart) *collectorapi.Chart {
565 + if src == nil {
566 + return nil
567 + }
568 +
569 + dst := *src
570 + dst.Labels = append([]collectorapi.Label(nil), src.Labels...)
571 + dst.Dims = make(collectorapi.Dims, len(src.Dims))
572 + for i, dim := range src.Dims {
573 + if dim == nil {
574 + continue
575 + }
576 + d := *dim
577 + dst.Dims[i] = &d
578 + }
579 + dst.Vars = make(collectorapi.Vars, len(src.Vars))
580 + for i, v := range src.Vars {
581 + if v == nil {
582 + continue
583 + }
584 + varCopy := *v
585 + dst.Vars[i] = &varCopy
586 + }
587 +
588 + return &dst
589 +}
590 +
591 +func (da *Auditor) printContextAnalysis(ctxInfo *contextInfo, isLast bool) []string {
592 charts := ctxInfo.charts
593 var issues []string
594
@@ -1016,7 +1120,7 @@ func contains(slice []string, item string) bool {
1120 }
1121
1122 // analyzeMetricDimensionMatching performs comprehensive analysis of dimension/metric matching
1019 -func (da *DumpAnalyzer) analyzeMetricDimensionMatching(job *JobAnalysis, allDimIDs map[string][]string, contextIssues map[string][]string) {
1123 +func (da *Auditor) analyzeMetricDimensionMatching(job *JobAnalysis, allDimIDs map[string][]string, contextIssues map[string][]string) {
1124 // 1. Find duplicate dimension IDs across charts (already done above but let's be explicit)
1125 duplicateDimensions := []string{}
1126 for dimID, chartIDs := range allDimIDs {
@@ -1124,7 +1228,7 @@ func gcd(a, b int) int {
1228 }
1229
1230 // analyzeFamilyStructureForJob performs family-level structural analysis for a single job
1127 -func (da *DumpAnalyzer) analyzeFamilyStructureForJob(job *JobAnalysis, contextIssues map[string][]string) {
1231 +func (da *Auditor) analyzeFamilyStructureForJob(job *JobAnalysis, contextIssues map[string][]string) {
1232 // Get all charts from this job
1233 allCharts := []*ChartAnalysis{}
1234 for i := range job.Charts {
@@ -1419,15 +1523,32 @@ func isSnakeCase(s string) bool {
1523 }
1524
1525 // PrintDebugInfo prints additional debug information
1422 -func (da *DumpAnalyzer) PrintDebugInfo() {
1526 +func (da *Auditor) PrintDebugInfo() {
1527 da.mu.RLock()
1528 defer da.mu.RUnlock()
1529
1530 fmt.Println("\n\nDEBUG INFORMATION:")
1531 fmt.Println(strings.Repeat("-", 80))
1532
1429 - for jobName, job := range da.jobs {
1430 - fmt.Printf("\n[%s] Chart Structure:\n", jobName)
1533 + type debugEntry struct {
1534 + id JobID
1535 + job *JobAnalysis
1536 + }
1537 +
1538 + entries := make([]debugEntry, 0, len(da.jobs))
1539 + for id, job := range da.jobs {
1540 + entries = append(entries, debugEntry{id: id, job: job})
1541 + }
1542 + sort.Slice(entries, func(i, j int) bool {
1543 + if entries[i].id.Module == entries[j].id.Module {
1544 + return entries[i].id.Name < entries[j].id.Name
1545 + }
1546 + return entries[i].id.Module < entries[j].id.Module
1547 + })
1548 +
1549 + for _, entry := range entries {
1550 + job := entry.job
1551 + fmt.Printf("\n[%s][%s] Chart Structure:\n", entry.id.Module, entry.id.Name)
1552
1553 for _, ca := range job.Charts {
1554 fmt.Printf("\nChart ID: %s\n", ca.Chart.ID)
src/go/plugin/ibm.d/modules/as400/collector.go
+2 -2
@@ -1110,8 +1110,8 @@ func (c *Collector) Cleanup(ctx context.Context) {
1110 c.Collector.Cleanup(ctx)
1111 }
1112
1113 -// EnableDump allows the collector to emit structured dump artifacts when requested.
1114 -func (c *Collector) EnableDump(dir string) {
1113 +// EnableCaptureArtifacts allows the collector to emit structured capture artifacts when requested.
1114 +func (c *Collector) EnableCaptureArtifacts(dir string) {
1115 ctx, err := newDumpContext(dir, &c.Config)
1116 if err != nil {
1117 c.Errorf("failed to initialise dump context: %v", err)